From f2879aebde70ec56ab6fe453092f94d36f2cbb18 Mon Sep 17 00:00:00 2001 From: John Kerl Date: Wed, 8 Dec 2021 15:50:07 -0500 Subject: [PATCH] Singleton-list step for reader-batching at input --- internal/pkg/auxents/repl/session.go | 2 +- internal/pkg/auxents/repl/types.go | 4 +-- internal/pkg/auxents/repl/verbs.go | 33 +++++++++++------ internal/pkg/cli/option_types.go | 6 ++++ internal/pkg/input/pseudo_reader_gen.go | 20 +++++++---- internal/pkg/input/record_reader.go | 3 +- internal/pkg/input/record_reader_csv.go | 24 ++++++++----- internal/pkg/input/record_reader_csvlite.go | 36 ++++++++++++------- internal/pkg/input/record_reader_dkvp.go | 29 +++++++++------ internal/pkg/input/record_reader_factory.go | 18 +++++----- internal/pkg/input/record_reader_json.go | 30 +++++++++------- internal/pkg/input/record_reader_nidx.go | 22 +++++++----- internal/pkg/input/record_reader_xtab.go | 23 +++++++----- internal/pkg/stream/stream.go | 35 +++++++++++++++--- internal/pkg/transformers/join.go | 11 ++++-- .../transformers/utils/join-bucket-keeper.go | 11 +++--- internal/pkg/types/context.go | 29 +++++++++++++++ todo.txt | 18 ++++------ 18 files changed, 239 insertions(+), 115 deletions(-) diff --git a/internal/pkg/auxents/repl/session.go b/internal/pkg/auxents/repl/session.go index 1c4d1a019..04dfd3015 100644 --- a/internal/pkg/auxents/repl/session.go +++ b/internal/pkg/auxents/repl/session.go @@ -47,7 +47,7 @@ func NewRepl( recordOutputStream *os.File, ) (*Repl, error) { - recordReader, err := input.Create(&options.ReaderOptions) + recordReader, err := input.Create(&options.ReaderOptions, 1) // recordsPerBatch if err != nil { return nil, err } diff --git a/internal/pkg/auxents/repl/types.go b/internal/pkg/auxents/repl/types.go index f33e45a3f..ea20658a6 100644 --- a/internal/pkg/auxents/repl/types.go +++ b/internal/pkg/auxents/repl/types.go @@ -6,6 +6,7 @@ package repl import ( "bufio" + "container/list" "os" "github.com/johnkerl/miller/internal/pkg/cli" @@ -13,7 +14,6 @@ import ( "github.com/johnkerl/miller/internal/pkg/input" "github.com/johnkerl/miller/internal/pkg/output" "github.com/johnkerl/miller/internal/pkg/runtime" - "github.com/johnkerl/miller/internal/pkg/types" ) // ================================================================ @@ -46,7 +46,7 @@ type Repl struct { options *cli.TOptions - readerChannel chan *types.RecordAndContext + readerChannel chan *list.List // list of *types.RecordAndContext errorChannel chan error downstreamDoneChannel chan bool recordReader input.IRecordReader diff --git a/internal/pkg/auxents/repl/verbs.go b/internal/pkg/auxents/repl/verbs.go index 9b4f29de5..5197dc8b6 100644 --- a/internal/pkg/auxents/repl/verbs.go +++ b/internal/pkg/auxents/repl/verbs.go @@ -5,6 +5,7 @@ package repl import ( + "container/list" "fmt" "os" "strings" @@ -218,7 +219,7 @@ func (repl *Repl) openFiles(filenames []string) { // Remember for :reopen repl.options.FileNames = filenames - repl.readerChannel = make(chan *types.RecordAndContext, 10) + repl.readerChannel = make(chan *list.List, 2) // list of *types.RecordAndContext repl.errorChannel = make(chan error, 1) repl.downstreamDoneChannel = make(chan bool, 1) @@ -270,11 +271,11 @@ func handleRead(repl *Repl, args []string) bool { return true } - var recordAndContext *types.RecordAndContext = nil + var recordsAndContexts *list.List // list of *types.RecordAndContext var err error = nil select { - case recordAndContext = <-repl.readerChannel: + case recordsAndContexts = <-repl.readerChannel: break case err = <-repl.errorChannel: break @@ -287,7 +288,11 @@ func handleRead(repl *Repl, args []string) bool { return true } - if recordAndContext != nil { + if recordsAndContexts != nil { + // TODO: comment and make very clear we've set this all up to batch by 1 for the REPL + lib.InternalCodingErrorIf(recordsAndContexts.Len() != 1) + recordAndContext := recordsAndContexts.Front().Value.(*types.RecordAndContext) + skipOrProcessRecord( repl, recordAndContext, @@ -414,12 +419,12 @@ func handleProcess(repl *Repl, args []string) bool { // ---------------------------------------------------------------- func handleSkipOrProcessN(repl *Repl, n int, processingNotSkipping bool) { - var recordAndContext *types.RecordAndContext = nil + var recordsAndContexts *list.List // list of *types.RecordAndContext var err error = nil for i := 1; i <= n; i++ { select { - case recordAndContext = <-repl.readerChannel: + case recordsAndContexts = <-repl.readerChannel: break case err = <-repl.errorChannel: break @@ -434,7 +439,11 @@ func handleSkipOrProcessN(repl *Repl, n int, processingNotSkipping bool) { return } - if recordAndContext != nil { + if recordsAndContexts != nil { + // TODO: comment and make very clear we've set this all up to batch by 1 for the REPL + lib.InternalCodingErrorIf(recordsAndContexts.Len() != 1) + recordAndContext := recordsAndContexts.Front().Value.(*types.RecordAndContext) + shouldBreak := skipOrProcessRecord( repl, recordAndContext, @@ -472,12 +481,12 @@ func handleSkipOrProcessUntil(repl *Repl, dslString string, processingNotSkippin return } - var recordAndContext *types.RecordAndContext = nil + var recordsAndContexts *list.List // list of *types.RecordAndContext for { doubleBreak := false select { - case recordAndContext = <-repl.readerChannel: + case recordsAndContexts = <-repl.readerChannel: break case err = <-repl.errorChannel: break @@ -496,7 +505,11 @@ func handleSkipOrProcessUntil(repl *Repl, dslString string, processingNotSkippin return } - if recordAndContext != nil { + if recordsAndContexts != nil { + // TODO: comment and make very clear we've set this all up to batch by 1 for the REPL + lib.InternalCodingErrorIf(recordsAndContexts.Len() != 1) + recordAndContext := recordsAndContexts.Front().Value.(*types.RecordAndContext) + shouldBreak := skipOrProcessRecord( repl, recordAndContext, diff --git a/internal/pkg/cli/option_types.go b/internal/pkg/cli/option_types.go index d9063cf56..4529a2ac2 100644 --- a/internal/pkg/cli/option_types.go +++ b/internal/pkg/cli/option_types.go @@ -72,6 +72,9 @@ type TReaderOptions struct { PrepipeIsRaw bool // For in-process gunzip/bunzip2/zcat (distinct from prepipe) FileInputEncoding lib.TFileInputEncoding + + // TODO: comment + RecordsPerBatch int } // ---------------------------------------------------------------- @@ -181,6 +184,9 @@ func DefaultReaderOptions() TReaderOptions { StepAsString: DEFAULT_GEN_STEP_AS_STRING, StopAsString: DEFAULT_GEN_STOP_AS_STRING, }, + + // TODO: make a cli option + RecordsPerBatch: 500, } } diff --git a/internal/pkg/input/pseudo_reader_gen.go b/internal/pkg/input/pseudo_reader_gen.go index e3af2a2e0..e7150df84 100644 --- a/internal/pkg/input/pseudo_reader_gen.go +++ b/internal/pkg/input/pseudo_reader_gen.go @@ -1,6 +1,7 @@ package input import ( + "container/list" "errors" "fmt" @@ -9,29 +10,34 @@ import ( ) type PseudoReaderGen struct { - readerOptions *cli.TReaderOptions + readerOptions *cli.TReaderOptions + recordsPerBatch int } -func NewPseudoReaderGen(readerOptions *cli.TReaderOptions) (*PseudoReaderGen, error) { +func NewPseudoReaderGen( + readerOptions *cli.TReaderOptions, + recordsPerBatch int, +) (*PseudoReaderGen, error) { return &PseudoReaderGen{ - readerOptions: readerOptions, + readerOptions: readerOptions, + recordsPerBatch: recordsPerBatch, }, nil } func (reader *PseudoReaderGen) Read( filenames []string, // ignored context types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { reader.process(&context, readerChannel, errorChannel, downstreamDoneChannel) - readerChannel <- types.NewEndOfStreamMarker(&context) + readerChannel <- types.NewEndOfStreamMarkerList(&context) } func (reader *PseudoReaderGen) process( context *types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -91,7 +97,7 @@ func (reader *PseudoReaderGen) process( record.PutCopy(key, value) context.UpdateForInputRecord() - readerChannel <- types.NewRecordAndContext( + readerChannel <- types.NewRecordAndContextList( record, context, ) diff --git a/internal/pkg/input/record_reader.go b/internal/pkg/input/record_reader.go index c6f63f024..f0a58c408 100644 --- a/internal/pkg/input/record_reader.go +++ b/internal/pkg/input/record_reader.go @@ -2,6 +2,7 @@ package input import ( "bufio" + "container/list" "io" "github.com/johnkerl/miller/internal/pkg/types" @@ -19,7 +20,7 @@ type IRecordReader interface { Read( filenames []string, initialContext types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) diff --git a/internal/pkg/input/record_reader_csv.go b/internal/pkg/input/record_reader_csv.go index e4de0081f..b4c46c17a 100644 --- a/internal/pkg/input/record_reader_csv.go +++ b/internal/pkg/input/record_reader_csv.go @@ -2,6 +2,7 @@ package input import ( "bytes" + "container/list" "encoding/csv" "errors" "fmt" @@ -17,11 +18,15 @@ import ( // ---------------------------------------------------------------- type RecordReaderCSV struct { readerOptions *cli.TReaderOptions + recordsPerBatch int ifs0 byte // Go's CSV library only lets its 'Comma' be a single character } // ---------------------------------------------------------------- -func NewRecordReaderCSV(readerOptions *cli.TReaderOptions) (*RecordReaderCSV, error) { +func NewRecordReaderCSV( + readerOptions *cli.TReaderOptions, + recordsPerBatch int, +) (*RecordReaderCSV, error) { if readerOptions.IRS != "\n" && readerOptions.IRS != "\r\n" { return nil, errors.New("CSV IRS cannot be altered; LF vs CR/LF is autodetected") } @@ -29,8 +34,9 @@ func NewRecordReaderCSV(readerOptions *cli.TReaderOptions) (*RecordReaderCSV, er return nil, errors.New("CSV IFS can only be a single character") } return &RecordReaderCSV{ - readerOptions: readerOptions, - ifs0: readerOptions.IFS[0], + readerOptions: readerOptions, + ifs0: readerOptions.IFS[0], + recordsPerBatch: recordsPerBatch, }, nil } @@ -38,7 +44,7 @@ func NewRecordReaderCSV(readerOptions *cli.TReaderOptions) (*RecordReaderCSV, er func (reader *RecordReaderCSV) Read( filenames []string, context types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -70,7 +76,7 @@ func (reader *RecordReaderCSV) Read( } } } - readerChannel <- types.NewEndOfStreamMarker(&context) + readerChannel <- types.NewEndOfStreamMarkerList(&context) } // ---------------------------------------------------------------- @@ -78,7 +84,7 @@ func (reader *RecordReaderCSV) processHandle( handle io.Reader, filename string, context *types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -204,7 +210,7 @@ func (reader *RecordReaderCSV) processHandle( context.UpdateForInputRecord() - readerChannel <- types.NewRecordAndContext( + readerChannel <- types.NewRecordAndContextList( record, context, ) @@ -216,7 +222,7 @@ func (reader *RecordReaderCSV) processHandle( func (reader *RecordReaderCSV) maybeConsumeComment( csvRecord []string, context *types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext ) bool { if reader.readerOptions.CommentHandling == cli.CommentsAreData { // Nothing is to be construed as a comment @@ -249,7 +255,7 @@ func (reader *RecordReaderCSV) maybeConsumeComment( csvWriter.Comma = rune(reader.ifs0) csvWriter.Write(csvRecord) csvWriter.Flush() - readerChannel <- types.NewOutputString(buffer.String(), context) + readerChannel <- types.NewOutputStringList(buffer.String(), context) } else /* reader.readerOptions.CommentHandling == cli.SkipComments */ { // discard entirely } diff --git a/internal/pkg/input/record_reader_csvlite.go b/internal/pkg/input/record_reader_csvlite.go index 7798935fc..5d8629272 100644 --- a/internal/pkg/input/record_reader_csvlite.go +++ b/internal/pkg/input/record_reader_csvlite.go @@ -19,6 +19,7 @@ package input // 3,4,5,6 3,4,5 import ( + "container/list" "errors" "fmt" "io" @@ -32,20 +33,29 @@ import ( // ---------------------------------------------------------------- type RecordReaderCSVLite struct { - readerOptions *cli.TReaderOptions + readerOptions *cli.TReaderOptions + recordsPerBatch int } // ---------------------------------------------------------------- -func NewRecordReaderCSVLite(readerOptions *cli.TReaderOptions) (*RecordReaderCSVLite, error) { +func NewRecordReaderCSVLite( + readerOptions *cli.TReaderOptions, + recordsPerBatch int, +) (*RecordReaderCSVLite, error) { return &RecordReaderCSVLite{ - readerOptions: readerOptions, + readerOptions: readerOptions, + recordsPerBatch: recordsPerBatch, }, nil } // ---------------------------------------------------------------- -func NewRecordReaderPPRINT(readerOptions *cli.TReaderOptions) (*RecordReaderCSVLite, error) { +func NewRecordReaderPPRINT( + readerOptions *cli.TReaderOptions, + recordsPerBatch int, +) (*RecordReaderCSVLite, error) { return &RecordReaderCSVLite{ - readerOptions: readerOptions, + readerOptions: readerOptions, + recordsPerBatch: recordsPerBatch, }, nil } @@ -53,7 +63,7 @@ func NewRecordReaderPPRINT(readerOptions *cli.TReaderOptions) (*RecordReaderCSVL func (reader *RecordReaderCSVLite) Read( filenames []string, context types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -121,7 +131,7 @@ func (reader *RecordReaderCSVLite) Read( } } } - readerChannel <- types.NewEndOfStreamMarker(&context) + readerChannel <- types.NewEndOfStreamMarkerList(&context) } // ---------------------------------------------------------------- @@ -129,7 +139,7 @@ func (reader *RecordReaderCSVLite) processHandleExplicitCSVHeader( handle io.Reader, filename string, context *types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -170,7 +180,7 @@ func (reader *RecordReaderCSVLite) processHandleExplicitCSVHeader( // Check for comments-in-data feature if strings.HasPrefix(line, reader.readerOptions.CommentString) { if reader.readerOptions.CommentHandling == cli.PassComments { - readerChannel <- types.NewOutputString(line+"\n", context) + readerChannel <- types.NewOutputStringList(line+"\n", context) continue } else if reader.readerOptions.CommentHandling == cli.SkipComments { continue @@ -242,7 +252,7 @@ func (reader *RecordReaderCSVLite) processHandleExplicitCSVHeader( } context.UpdateForInputRecord() - readerChannel <- types.NewRecordAndContext( + readerChannel <- types.NewRecordAndContextList( record, context, ) @@ -256,7 +266,7 @@ func (reader *RecordReaderCSVLite) processHandleImplicitCSVHeader( handle io.Reader, filename string, context *types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -293,7 +303,7 @@ func (reader *RecordReaderCSVLite) processHandleImplicitCSVHeader( // Check for comments-in-data feature if strings.HasPrefix(line, reader.readerOptions.CommentString) { if reader.readerOptions.CommentHandling == cli.PassComments { - readerChannel <- types.NewOutputString(line+"\n", context) + readerChannel <- types.NewOutputStringList(line+"\n", context) continue } else if reader.readerOptions.CommentHandling == cli.SkipComments { continue @@ -373,7 +383,7 @@ func (reader *RecordReaderCSVLite) processHandleImplicitCSVHeader( } context.UpdateForInputRecord() - readerChannel <- types.NewRecordAndContext( + readerChannel <- types.NewRecordAndContextList( record, context, ) diff --git a/internal/pkg/input/record_reader_dkvp.go b/internal/pkg/input/record_reader_dkvp.go index 87cb16f9a..bd39d9f5e 100644 --- a/internal/pkg/input/record_reader_dkvp.go +++ b/internal/pkg/input/record_reader_dkvp.go @@ -1,6 +1,7 @@ package input import ( + "container/list" "io" "strconv" "strings" @@ -11,19 +12,24 @@ import ( ) type RecordReaderDKVP struct { - readerOptions *cli.TReaderOptions + readerOptions *cli.TReaderOptions + recordsPerBatch int } -func NewRecordReaderDKVP(readerOptions *cli.TReaderOptions) (*RecordReaderDKVP, error) { +func NewRecordReaderDKVP( + readerOptions *cli.TReaderOptions, + recordsPerBatch int, +) (*RecordReaderDKVP, error) { return &RecordReaderDKVP{ - readerOptions: readerOptions, + readerOptions: readerOptions, + recordsPerBatch: recordsPerBatch, }, nil } func (reader *RecordReaderDKVP) Read( filenames []string, context types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -55,21 +61,22 @@ func (reader *RecordReaderDKVP) Read( } } } - readerChannel <- types.NewEndOfStreamMarker(&context) + readerChannel <- types.NewEndOfStreamMarkerList(&context) } func (reader *RecordReaderDKVP) processHandle( handle io.Reader, filename string, context *types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { context.UpdateForStartOfFile(filename) - scanner := NewLineScanner(handle, reader.readerOptions.IRS) - for scanner.Scan() { + lineScanner := NewLineScanner(handle, reader.readerOptions.IRS) + + for lineScanner.Scan() { // See if downstream processors will be ignoring further data (e.g. mlr // head). If so, stop reading. This makes 'mlr head hugefile' exit @@ -86,12 +93,12 @@ func (reader *RecordReaderDKVP) processHandle( break } - line := scanner.Text() + line := lineScanner.Text() // Check for comments-in-data feature if strings.HasPrefix(line, reader.readerOptions.CommentString) { if reader.readerOptions.CommentHandling == cli.PassComments { - readerChannel <- types.NewOutputString(line+"\n", context) + readerChannel <- types.NewOutputStringList(line+"\n", context) continue } else if reader.readerOptions.CommentHandling == cli.SkipComments { continue @@ -101,7 +108,7 @@ func (reader *RecordReaderDKVP) processHandle( record := reader.recordFromDKVPLine(line) context.UpdateForInputRecord() - readerChannel <- types.NewRecordAndContext( + readerChannel <- types.NewRecordAndContextList( record, context, ) diff --git a/internal/pkg/input/record_reader_factory.go b/internal/pkg/input/record_reader_factory.go index 9c54ce175..b6f4fa829 100644 --- a/internal/pkg/input/record_reader_factory.go +++ b/internal/pkg/input/record_reader_factory.go @@ -7,24 +7,24 @@ import ( "github.com/johnkerl/miller/internal/pkg/cli" ) -func Create(readerOptions *cli.TReaderOptions) (IRecordReader, error) { +func Create(readerOptions *cli.TReaderOptions, recordsPerBatch int) (IRecordReader, error) { switch readerOptions.InputFileFormat { case "csv": - return NewRecordReaderCSV(readerOptions) + return NewRecordReaderCSV(readerOptions, recordsPerBatch) case "csvlite": - return NewRecordReaderCSVLite(readerOptions) + return NewRecordReaderCSVLite(readerOptions, recordsPerBatch) case "dkvp": - return NewRecordReaderDKVP(readerOptions) + return NewRecordReaderDKVP(readerOptions, recordsPerBatch) case "json": - return NewRecordReaderJSON(readerOptions) + return NewRecordReaderJSON(readerOptions, recordsPerBatch) case "nidx": - return NewRecordReaderNIDX(readerOptions) + return NewRecordReaderNIDX(readerOptions, recordsPerBatch) case "pprint": - return NewRecordReaderPPRINT(readerOptions) + return NewRecordReaderPPRINT(readerOptions, recordsPerBatch) case "xtab": - return NewRecordReaderXTAB(readerOptions) + return NewRecordReaderXTAB(readerOptions, recordsPerBatch) case "gen": - return NewPseudoReaderGen(readerOptions) + return NewPseudoReaderGen(readerOptions, recordsPerBatch) default: return nil, errors.New(fmt.Sprintf("input file format \"%s\" not found", readerOptions.InputFileFormat)) } diff --git a/internal/pkg/input/record_reader_json.go b/internal/pkg/input/record_reader_json.go index d3256a302..e805a9dbf 100644 --- a/internal/pkg/input/record_reader_json.go +++ b/internal/pkg/input/record_reader_json.go @@ -2,6 +2,7 @@ package input import ( "bufio" + "container/list" "errors" "fmt" "io" @@ -15,19 +16,24 @@ import ( ) type RecordReaderJSON struct { - readerOptions *cli.TReaderOptions + readerOptions *cli.TReaderOptions + recordsPerBatch int } -func NewRecordReaderJSON(readerOptions *cli.TReaderOptions) (*RecordReaderJSON, error) { +func NewRecordReaderJSON( + readerOptions *cli.TReaderOptions, + recordsPerBatch int, +) (*RecordReaderJSON, error) { return &RecordReaderJSON{ - readerOptions: readerOptions, + readerOptions: readerOptions, + recordsPerBatch: recordsPerBatch, }, nil } func (reader *RecordReaderJSON) Read( filenames []string, context types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -59,14 +65,14 @@ func (reader *RecordReaderJSON) Read( } } } - readerChannel <- types.NewEndOfStreamMarker(&context) + readerChannel <- types.NewEndOfStreamMarkerList(&context) } func (reader *RecordReaderJSON) processHandle( handle io.Reader, filename string, context *types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -116,7 +122,7 @@ func (reader *RecordReaderJSON) processHandle( return } context.UpdateForInputRecord() - readerChannel <- types.NewRecordAndContext( + readerChannel <- types.NewRecordAndContextList( record, context, ) @@ -143,7 +149,7 @@ func (reader *RecordReaderJSON) processHandle( return } context.UpdateForInputRecord() - readerChannel <- types.NewRecordAndContext( + readerChannel <- types.NewRecordAndContextList( record, context, ) @@ -187,8 +193,8 @@ func (reader *RecordReaderJSON) processHandle( type JSONCommentEnabledReader struct { lineScanner *bufio.Scanner readerOptions *cli.TReaderOptions - context *types.Context // Needed for channelized stdout-printing logic - readerChannel chan<- *types.RecordAndContext + context *types.Context // Needed for channelized stdout-printing logic + readerChannel chan<- *list.List // list of *types.RecordAndContext // In case a line was ingested which was longer than the read-buffer passed // to us, in which case we need to split up that line and return it over @@ -199,7 +205,7 @@ type JSONCommentEnabledReader struct { func NewJSONCommentEnabledReader( underlying io.Reader, readerOptions *cli.TReaderOptions, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext ) *JSONCommentEnabledReader { return &JSONCommentEnabledReader{ lineScanner: bufio.NewScanner(underlying), @@ -234,7 +240,7 @@ func (bsr *JSONCommentEnabledReader) Read(p []byte) (n int, err error) { if bsr.readerOptions.CommentHandling == cli.PassComments { // Insert the string into the record-output stream, so that goroutine can // print it, resulting in deterministic output-ordering. - bsr.readerChannel <- types.NewOutputString(line+"\n", bsr.context) + bsr.readerChannel <- types.NewOutputStringList(line+"\n", bsr.context) } } } diff --git a/internal/pkg/input/record_reader_nidx.go b/internal/pkg/input/record_reader_nidx.go index 9ecf71a48..b6fab8eb9 100644 --- a/internal/pkg/input/record_reader_nidx.go +++ b/internal/pkg/input/record_reader_nidx.go @@ -1,6 +1,7 @@ package input import ( + "container/list" "io" "strconv" "strings" @@ -11,19 +12,24 @@ import ( ) type RecordReaderNIDX struct { - readerOptions *cli.TReaderOptions + readerOptions *cli.TReaderOptions + recordsPerBatch int } -func NewRecordReaderNIDX(readerOptions *cli.TReaderOptions) (*RecordReaderNIDX, error) { +func NewRecordReaderNIDX( + readerOptions *cli.TReaderOptions, + recordsPerBatch int, +) (*RecordReaderNIDX, error) { return &RecordReaderNIDX{ - readerOptions: readerOptions, + readerOptions: readerOptions, + recordsPerBatch: recordsPerBatch, }, nil } func (reader *RecordReaderNIDX) Read( filenames []string, context types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -55,14 +61,14 @@ func (reader *RecordReaderNIDX) Read( } } } - readerChannel <- types.NewEndOfStreamMarker(&context) + readerChannel <- types.NewEndOfStreamMarkerList(&context) } func (reader *RecordReaderNIDX) processHandle( handle io.Reader, filename string, context *types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -92,7 +98,7 @@ func (reader *RecordReaderNIDX) processHandle( // Check for comments-in-data feature if strings.HasPrefix(line, reader.readerOptions.CommentString) { if reader.readerOptions.CommentHandling == cli.PassComments { - readerChannel <- types.NewOutputString(line+"\n", context) + readerChannel <- types.NewOutputStringList(line+"\n", context) continue } else if reader.readerOptions.CommentHandling == cli.SkipComments { continue @@ -103,7 +109,7 @@ func (reader *RecordReaderNIDX) processHandle( record := reader.recordFromNIDXLine(line) context.UpdateForInputRecord() - readerChannel <- types.NewRecordAndContext( + readerChannel <- types.NewRecordAndContextList( record, context, ) diff --git a/internal/pkg/input/record_reader_xtab.go b/internal/pkg/input/record_reader_xtab.go index 97765d43a..9de0a3e75 100644 --- a/internal/pkg/input/record_reader_xtab.go +++ b/internal/pkg/input/record_reader_xtab.go @@ -12,14 +12,19 @@ import ( ) type RecordReaderXTAB struct { - readerOptions *cli.TReaderOptions + readerOptions *cli.TReaderOptions + recordsPerBatch int // Note: XTAB uses two consecutive IFS in place of an IRS; IRS is ignored } // ---------------------------------------------------------------- -func NewRecordReaderXTAB(readerOptions *cli.TReaderOptions) (*RecordReaderXTAB, error) { +func NewRecordReaderXTAB( + readerOptions *cli.TReaderOptions, + recordsPerBatch int, +) (*RecordReaderXTAB, error) { return &RecordReaderXTAB{ - readerOptions: readerOptions, + readerOptions: readerOptions, + recordsPerBatch: recordsPerBatch, }, nil } @@ -27,7 +32,7 @@ func NewRecordReaderXTAB(readerOptions *cli.TReaderOptions) (*RecordReaderXTAB, func (reader *RecordReaderXTAB) Read( filenames []string, context types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -59,14 +64,14 @@ func (reader *RecordReaderXTAB) Read( } } } - readerChannel <- types.NewEndOfStreamMarker(&context) + readerChannel <- types.NewEndOfStreamMarkerList(&context) } func (reader *RecordReaderXTAB) processHandle( handle io.Reader, filename string, context *types.Context, - readerChannel chan<- *types.RecordAndContext, + readerChannel chan<- *list.List, // list of *types.RecordAndContext errorChannel chan error, downstreamDoneChannel <-chan bool, // for mlr head ) { @@ -102,7 +107,7 @@ func (reader *RecordReaderXTAB) processHandle( return } context.UpdateForInputRecord() - readerChannel <- types.NewRecordAndContext(record, context) + readerChannel <- types.NewRecordAndContextList(record, context) linesForRecord = list.New() } @@ -114,7 +119,7 @@ func (reader *RecordReaderXTAB) processHandle( // Check for comments-in-data feature if strings.HasPrefix(line, reader.readerOptions.CommentString) { if reader.readerOptions.CommentHandling == cli.PassComments { - readerChannel <- types.NewOutputString(line+reader.readerOptions.IFS, context) + readerChannel <- types.NewOutputStringList(line+reader.readerOptions.IFS, context) continue } else if reader.readerOptions.CommentHandling == cli.SkipComments { continue @@ -133,7 +138,7 @@ func (reader *RecordReaderXTAB) processHandle( return } context.UpdateForInputRecord() - readerChannel <- types.NewRecordAndContext(record, context) + readerChannel <- types.NewRecordAndContextList(record, context) linesForRecord = list.New() } } diff --git a/internal/pkg/stream/stream.go b/internal/pkg/stream/stream.go index f83e59413..e969fdcee 100644 --- a/internal/pkg/stream/stream.go +++ b/internal/pkg/stream/stream.go @@ -2,6 +2,7 @@ package stream import ( "bufio" + "container/list" "fmt" "io" "os" @@ -47,8 +48,10 @@ func Stream( // passed through the channels along with each record. initialContext := types.NewContext() - // Instantiate the record-reader - recordReader, err := input.Create(&options.ReaderOptions) + // Instantiate the record-reader. + // RecordsPerBatch is tracked separately from ReaderOptions since join/repl + // may use batch size of 1. + recordReader, err := input.Create(&options.ReaderOptions, options.ReaderOptions.RecordsPerBatch) if err != nil { return err } @@ -60,7 +63,8 @@ func Stream( } // Set up the reader-to-transformer and transformer-to-writer channels. - readerChannel := make(chan *types.RecordAndContext, 10) + readerChannel := make(chan *list.List, 2) // list of *types.RecordAndContext + tempChannel := make(chan *types.RecordAndContext, 10) writerChannel := make(chan *types.RecordAndContext, 1) // We're done when a fatal error is registered on input (file not found, @@ -81,7 +85,9 @@ func Stream( bufferedOutputStream := bufio.NewWriter(outputStream) go recordReader.Read(fileNames, *initialContext, readerChannel, errorChannel, readerDownstreamDoneChannel) - go transformers.ChainTransformer(readerChannel, readerDownstreamDoneChannel, recordTransformers, + // TODO: temp for iterative batched-reader refactor + go tempReader(readerChannel, tempChannel) + go transformers.ChainTransformer(tempChannel, readerDownstreamDoneChannel, recordTransformers, writerChannel, options) go output.ChannelWriter(writerChannel, recordWriter, &options.WriterOptions, doneWritingChannel, bufferedOutputStream, outputIsStdout) @@ -102,3 +108,24 @@ func Stream( return nil } + +func tempReader( + readerChannel <-chan *list.List, // list of *types.RecordAndContext + transformerChannel chan<- *types.RecordAndContext, +) { + done := false + for !done { + racs := <-readerChannel + + for e := racs.Front(); e != nil; e = e.Next() { + rac := e.Value.(*types.RecordAndContext) + transformerChannel <- rac + + if rac.EndOfStream { + done = true + break + } + } + + } +} diff --git a/internal/pkg/transformers/join.go b/internal/pkg/transformers/join.go index f1faba9cc..d5c44c7b0 100644 --- a/internal/pkg/transformers/join.go +++ b/internal/pkg/transformers/join.go @@ -458,7 +458,8 @@ func (tr *TransformerJoin) ingestLeftFile() { readerOpts := &tr.opts.joinFlagOptions.ReaderOptions // Instantiate the record-reader - recordReader, err := input.Create(readerOpts) + // TODO: perhaps increase recordsPerBatch, and/or refactor + recordReader, err := input.Create(readerOpts, 1) if recordReader == nil { fmt.Fprintf(os.Stderr, "mlr join: %v\n", err) os.Exit(1) @@ -472,7 +473,7 @@ func (tr *TransformerJoin) ingestLeftFile() { initialContext.UpdateForStartOfFile(tr.opts.leftFileName) // Set up channels for the record-reader. - readerChannel := make(chan *types.RecordAndContext, 10) + readerChannel := make(chan *list.List, 2) // list of *types.RecordAndContext errorChannel := make(chan error, 1) downstreamDoneChannel := make(chan bool, 1) @@ -493,7 +494,11 @@ func (tr *TransformerJoin) ingestLeftFile() { fmt.Fprintln(os.Stderr, "mlr", ": ", err) os.Exit(1) - case leftrecAndContext := <-readerChannel: + case leftrecsAndContexts := <-readerChannel: + // TODO: temp for batch-reader refactor + lib.InternalCodingErrorIf(leftrecsAndContexts.Len() != 1) + leftrecAndContext := leftrecsAndContexts.Front().Value.(*types.RecordAndContext) + if leftrecAndContext.EndOfStream { done = true break // breaks the switch, not the for, in Golang diff --git a/internal/pkg/transformers/utils/join-bucket-keeper.go b/internal/pkg/transformers/utils/join-bucket-keeper.go index 5aa0290fd..6af1e2bd9 100644 --- a/internal/pkg/transformers/utils/join-bucket-keeper.go +++ b/internal/pkg/transformers/utils/join-bucket-keeper.go @@ -125,7 +125,7 @@ type JoinBucketKeeper struct { // For streaming through the left-side file recordReader input.IRecordReader context *types.Context - readerChannel <-chan *types.RecordAndContext + readerChannel <-chan *list.List // list of *types.RecordAndContext errorChannel chan error // TODO: merge with leof flag recordReaderDone bool @@ -165,7 +165,7 @@ func NewJoinBucketKeeper( ) *JoinBucketKeeper { // Instantiate the record-reader - recordReader, err := input.Create(joinReaderOptions) + recordReader, err := input.Create(joinReaderOptions, 1) // TODO: maybe increase records per batch if err != nil { fmt.Fprintf(os.Stderr, "mlr join: %v", err) os.Exit(1) @@ -178,7 +178,7 @@ func NewJoinBucketKeeper( initialContext.UpdateForStartOfFile(leftFileName) // Set up channels for the record-reader - readerChannel := make(chan *types.RecordAndContext, 10) + readerChannel := make(chan *list.List, 10) // list of *types.RecordAndContext errorChannel := make(chan error, 1) downstreamDoneChannel := make(chan bool, 1) @@ -570,7 +570,10 @@ func (keeper *JoinBucketKeeper) readRecord() *types.RecordAndContext { case err := <-keeper.errorChannel: fmt.Fprintln(os.Stderr, "mlr", ": ", err) os.Exit(1) - case leftrecAndContext := <-keeper.readerChannel: + case leftrecsAndContexts := <-keeper.readerChannel: + // TODO: temp + lib.InternalCodingErrorIf(leftrecsAndContexts.Len() != 1) + leftrecAndContext := leftrecsAndContexts.Front().Value.(*types.RecordAndContext) if leftrecAndContext.EndOfStream { // end-of-stream marker keeper.recordReaderDone = true return nil diff --git a/internal/pkg/types/context.go b/internal/pkg/types/context.go index 4faa25982..3a9ac5970 100644 --- a/internal/pkg/types/context.go +++ b/internal/pkg/types/context.go @@ -2,6 +2,7 @@ package types import ( "bytes" + "container/list" "strconv" ) @@ -37,6 +38,16 @@ func NewRecordAndContext( } } +// TODO: temp for batch-reader refactor +func NewRecordAndContextList( + record *Mlrmap, + context *Context, +) *list.List { + ell := list.New() + ell.PushBack(NewRecordAndContext(record, context)) + return ell +} + // For the record-readers to update their initial context as each new record is read. func (rac *RecordAndContext) Copy() *RecordAndContext { if rac == nil { @@ -69,6 +80,16 @@ func NewOutputString( } } +// TODO: temp for batch-reader refactor +func NewOutputStringList( + outputString string, + context *Context, +) *list.List { + ell := list.New() + ell.PushBack(NewOutputString(outputString, context)) + return ell +} + // For the record-readers to update their initial context as each new record is read. func NewEndOfStreamMarker(context *Context) *RecordAndContext { return &RecordAndContext{ @@ -79,6 +100,14 @@ func NewEndOfStreamMarker(context *Context) *RecordAndContext { } } +// TODO: comment +// For the record-readers to update their initial context as each new record is read. +func NewEndOfStreamMarkerList(context *Context) *list.List { + ell := list.New() + ell.PushBack(NewEndOfStreamMarker(context)) + return ell +} + // ---------------------------------------------------------------- type Context struct { FILENAME string diff --git a/todo.txt b/todo.txt index 7ec9ba8d1..c9fa2ba72 100644 --- a/todo.txt +++ b/todo.txt @@ -27,17 +27,15 @@ PUNCHDOWN LIST o dkvp-reader factor-out ... o mods: ? outputChannel -> *list.List at each transformer -- ? profile first - - un-legacy fflush flag :( - > conditional on isatty stdout - > new fflushWasSpecified - > downcase OFSWasSpecified, HaveRandSeed, et al. + ? readerChannel length 1 or 2 ? + ? cli option for records per batch + ? experiment again with hashed/unhashed - do and maybe keep? record-reader return (raclist, err) & refactor repl accordingly > needs factor for-loop to stateful so maybe not - - transfomers w/ reclist: *maybe*, but idchan/odchan too ... invest time after some refactor decions made + - transformers w/ reclist: *maybe*, but idchan/odchan too ... invest time after some refactor decions made - fix record/line sequencing regressions - - tail -f handling - > batch-size 1 on stdin, for repl at least? - > adaptive is-blocking detection -- make sure it's not over-sensitive + - maybe increase records-per-batch in join-bucket-keeper; and/or refactor + - maybe increase records-per-batch in repl; and/or refactor o goals: - keep goroutines -- including per-transformer -- for parallelism - look for flex ideas on how to structure that parallelism @@ -70,10 +68,6 @@ PUNCHDOWN LIST d how to handle tail -f and repl > try unconditional batching and hiding *first* to see how much best-case perf to be had ? lazy type-infer?? needs careful use of accessor-mutators in place of mv.type etc -<<<<<<< HEAD -======= -* note somewhere why NewEndOfStreamMarker instead of channel close -- readers carry final context ->>>>>>> f2e709408 (Rename inputChannel,outputChannel to readerChannel,writerChannel (#772)) * blockers: - keep checking issues