Mirrors/miller
1
0
Fork 0
mirror of https://github.com/johnkerl/miller.git synced 2026-07-25 08:53:55 +00:00
Code Issues Projects Releases Wiki Activity

Singleton-list step for reader-batching at input

Browse source
This commit is contained in:
John Kerl 2021-12-08 15:50:07 -05:00
parent 2f8d2b18f9
commit f2879aebde
18 changed files with 239 additions and 115 deletions
Download patch file Download diff file Expand all files Collapse all files

2
internal/pkg/auxents/repl/session.go
View file

@ -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
}

4
internal/pkg/auxents/repl/types.go
View file

@ -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

33
internal/pkg/auxents/repl/verbs.go
View file

@ -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,

6
internal/pkg/cli/option_types.go
View file

@ -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,
}
}

20
internal/pkg/input/pseudo_reader_gen.go
View file

@ -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,
)

3
internal/pkg/input/record_reader.go
View file

@ -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
)

24
internal/pkg/input/record_reader_csv.go
View file

@ -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
}

36
internal/pkg/input/record_reader_csvlite.go
View file

@ -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,
)

29
internal/pkg/input/record_reader_dkvp.go
View file

@ -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,
)

18
internal/pkg/input/record_reader_factory.go
View file

@ -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))
}

30
internal/pkg/input/record_reader_json.go
View file

@ -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)
}
}
}

22
internal/pkg/input/record_reader_nidx.go
View file

@ -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,
)

23
internal/pkg/input/record_reader_xtab.go
View file

@ -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()
}
}

35
internal/pkg/stream/stream.go
View file

@ -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
}
}
}
}

11
internal/pkg/transformers/join.go
View file

@ -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

11
internal/pkg/transformers/utils/join-bucket-keeper.go
View file

@ -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

29
internal/pkg/types/context.go
View file

@ -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

18
todo.txt
View file

@ -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