diff --git a/internal/pkg/input/record_reader_csvlite.go b/internal/pkg/input/record_reader_csvlite.go index e75966948..b8150aee2 100644 --- a/internal/pkg/input/record_reader_csvlite.go +++ b/internal/pkg/input/record_reader_csvlite.go @@ -31,13 +31,11 @@ import ( "github.com/johnkerl/miller/internal/pkg/types" ) -// ---------------------------------------------------------------- type RecordReaderCSVLite struct { readerOptions *cli.TReaderOptions recordsPerBatch int } -// ---------------------------------------------------------------- func NewRecordReaderCSVLite( readerOptions *cli.TReaderOptions, recordsPerBatch int, @@ -48,7 +46,6 @@ func NewRecordReaderCSVLite( }, nil } -// ---------------------------------------------------------------- func NewRecordReaderPPRINT( readerOptions *cli.TReaderOptions, recordsPerBatch int, @@ -59,7 +56,6 @@ func NewRecordReaderPPRINT( }, nil } -// ---------------------------------------------------------------- func (reader *RecordReaderCSVLite) Read( filenames []string, context types.Context, @@ -134,7 +130,6 @@ func (reader *RecordReaderCSVLite) Read( readerChannel <- types.NewEndOfStreamMarkerList(&context) } -// ---------------------------------------------------------------- func (reader *RecordReaderCSVLite) processHandleExplicitCSVHeader( handle io.Reader, filename string, @@ -148,8 +143,8 @@ func (reader *RecordReaderCSVLite) processHandleExplicitCSVHeader( 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 @@ -166,7 +161,7 @@ func (reader *RecordReaderCSVLite) processHandleExplicitCSVHeader( break } - line := scanner.Text() + line := lineScanner.Text() inputLineNumber++ @@ -261,7 +256,6 @@ func (reader *RecordReaderCSVLite) processHandleExplicitCSVHeader( } } -// ---------------------------------------------------------------- func (reader *RecordReaderCSVLite) processHandleImplicitCSVHeader( handle io.Reader, filename string, @@ -275,8 +269,8 @@ func (reader *RecordReaderCSVLite) processHandleImplicitCSVHeader( 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 @@ -295,8 +289,7 @@ func (reader *RecordReaderCSVLite) processHandleImplicitCSVHeader( break } - // TODO: IRS - line := scanner.Text() + line := lineScanner.Text() inputLineNumber++ diff --git a/internal/pkg/input/record_reader_dkvp.go b/internal/pkg/input/record_reader_dkvp_nidx.go similarity index 72% rename from internal/pkg/input/record_reader_dkvp.go rename to internal/pkg/input/record_reader_dkvp_nidx.go index 255834cba..3672dedbd 100644 --- a/internal/pkg/input/record_reader_dkvp.go +++ b/internal/pkg/input/record_reader_dkvp_nidx.go @@ -1,3 +1,5 @@ +// This is mostly-identical code for the DKVP and NIDX record-readers. + package input import ( @@ -11,22 +13,39 @@ import ( "github.com/johnkerl/miller/internal/pkg/types" ) -type RecordReaderDKVP struct { +// splitter_DKVP_NIDX is a function type for the one bit of code differing +// between the DKVP reader and the NIDX reader, namely, how it splits lines. +type splitter_DKVP_NIDX func (reader *RecordReaderDKVPNIDX, line string) *types.Mlrmap + +type RecordReaderDKVPNIDX struct { readerOptions *cli.TReaderOptions recordsPerBatch int + splitter splitter_DKVP_NIDX } func NewRecordReaderDKVP( readerOptions *cli.TReaderOptions, recordsPerBatch int, -) (*RecordReaderDKVP, error) { - return &RecordReaderDKVP{ +) (*RecordReaderDKVPNIDX, error) { + return &RecordReaderDKVPNIDX{ readerOptions: readerOptions, recordsPerBatch: recordsPerBatch, + splitter: recordFromDKVPLine, }, nil } -func (reader *RecordReaderDKVP) Read( +func NewRecordReaderNIDX( + readerOptions *cli.TReaderOptions, + recordsPerBatch int, +) (*RecordReaderDKVPNIDX, error) { + return &RecordReaderDKVPNIDX{ + readerOptions: readerOptions, + recordsPerBatch: recordsPerBatch, + splitter: recordFromNIDXLine, + }, nil +} + +func (reader *RecordReaderDKVPNIDX) Read( filenames []string, context types.Context, readerChannel chan<- *list.List, // list of *types.RecordAndContext @@ -64,7 +83,7 @@ func (reader *RecordReaderDKVP) Read( readerChannel <- types.NewEndOfStreamMarkerList(&context) } -func (reader *RecordReaderDKVP) processHandle( +func (reader *RecordReaderDKVPNIDX) processHandle( handle io.Reader, filename string, context *types.Context, @@ -89,7 +108,7 @@ func (reader *RecordReaderDKVP) processHandle( } // TODO: comment copiously we're trying to handle slow/fast/short/long reads: tail -f, smallfile, bigfile. -func (reader *RecordReaderDKVP) getRecordBatch( +func (reader *RecordReaderDKVPNIDX) getRecordBatch( linesChannel <-chan *list.List, maxBatchSize int, context *types.Context, @@ -121,7 +140,7 @@ func (reader *RecordReaderDKVP) getRecordBatch( } } - record := reader.recordFromDKVPLine(line) + record := reader.splitter(reader, line) context.UpdateForInputRecord() recordAndContext := types.NewRecordAndContext(record, context) recordsAndContexts.PushBack(recordAndContext) @@ -130,9 +149,7 @@ func (reader *RecordReaderDKVP) getRecordBatch( return recordsAndContexts, false } -func (reader *RecordReaderDKVP) recordFromDKVPLine( - line string, -) *types.Mlrmap { +func recordFromDKVPLine(reader *RecordReaderDKVPNIDX, line string) *types.Mlrmap { record := types.NewMlrmapAsRecord() var pairs []string @@ -168,3 +185,28 @@ func (reader *RecordReaderDKVP) recordFromDKVPLine( } return record } + +func recordFromNIDXLine(reader *RecordReaderDKVPNIDX, line string) *types.Mlrmap { + record := types.NewMlrmapAsRecord() + + var values []string + // TODO: func-pointer this away + if reader.readerOptions.IFSRegex == nil { // e.g. --no-ifs-regex + values = lib.SplitString(line, reader.readerOptions.IFS) + } else { + values = lib.RegexSplitString(reader.readerOptions.IFSRegex, line, -1) + } + + if reader.readerOptions.AllowRepeatIFS { + values = lib.StripEmpties(values) // left/right trim + } + + var i int = 0 + for _, value := range values { + i++ + key := strconv.Itoa(i) + mval := types.MlrvalFromInferredTypeForDataFiles(value) + record.PutReference(key, mval) + } + return record +} diff --git a/internal/pkg/input/record_reader_nidx.go b/internal/pkg/input/record_reader_nidx.go deleted file mode 100644 index 3aade9194..000000000 --- a/internal/pkg/input/record_reader_nidx.go +++ /dev/null @@ -1,157 +0,0 @@ -package input - -import ( - "container/list" - "io" - "strconv" - "strings" - - "github.com/johnkerl/miller/internal/pkg/cli" - "github.com/johnkerl/miller/internal/pkg/lib" - "github.com/johnkerl/miller/internal/pkg/types" -) - -type RecordReaderNIDX struct { - readerOptions *cli.TReaderOptions - recordsPerBatch int -} - -func NewRecordReaderNIDX( - readerOptions *cli.TReaderOptions, - recordsPerBatch int, -) (*RecordReaderNIDX, error) { - return &RecordReaderNIDX{ - readerOptions: readerOptions, - recordsPerBatch: recordsPerBatch, - }, nil -} - -func (reader *RecordReaderNIDX) Read( - filenames []string, - context types.Context, - readerChannel chan<- *list.List, // list of *types.RecordAndContext - errorChannel chan error, - downstreamDoneChannel <-chan bool, // for mlr head -) { - if filenames != nil { // nil for mlr -n - if len(filenames) == 0 { // read from stdin - handle, err := lib.OpenStdin( - reader.readerOptions.Prepipe, - reader.readerOptions.PrepipeIsRaw, - reader.readerOptions.FileInputEncoding, - ) - if err != nil { - errorChannel <- err - } - reader.processHandle(handle, "(stdin)", &context, readerChannel, errorChannel, downstreamDoneChannel) - } else { - for _, filename := range filenames { - handle, err := lib.OpenFileForRead( - filename, - reader.readerOptions.Prepipe, - reader.readerOptions.PrepipeIsRaw, - reader.readerOptions.FileInputEncoding, - ) - if err != nil { - errorChannel <- err - } else { - reader.processHandle(handle, filename, &context, readerChannel, errorChannel, downstreamDoneChannel) - handle.Close() - } - } - } - } - readerChannel <- types.NewEndOfStreamMarkerList(&context) -} - -func (reader *RecordReaderNIDX) processHandle( - handle io.Reader, - filename string, - context *types.Context, - readerChannel chan<- *list.List, - errorChannel chan<- error, - downstreamDoneChannel <-chan bool, // for mlr head -) { - context.UpdateForStartOfFile(filename) - recordsPerBatch := reader.readerOptions.RecordsPerBatch - - lineScanner := NewLineScanner(handle, reader.readerOptions.IRS) - linesChannel := make(chan *list.List, recordsPerBatch) - go channelizedLineScanner(lineScanner, linesChannel, downstreamDoneChannel, recordsPerBatch) - - for { - recordsAndContexts, eof := reader.getRecordBatch(linesChannel, recordsPerBatch, context) - readerChannel <- recordsAndContexts - if eof { - break - } - } -} - -// TODO: comment copiously we're trying to handle slow/fast/short/long reads: tail -f, smallfile, bigfile. -func (reader *RecordReaderNIDX) getRecordBatch( - linesChannel <-chan *list.List, - maxBatchSize int, - context *types.Context, -) ( - recordsAndContexts *list.List, - eof bool, -) { - recordsAndContexts = list.New() - - lines, more := <-linesChannel - if !more { - return recordsAndContexts, true - } - - for e := lines.Front(); e != nil; e = e.Next() { - line := e.Value.(string) - - // Check for comments-in-data feature - // TODO: function-pointer this away - if reader.readerOptions.CommentHandling != cli.CommentsAreData { - if strings.HasPrefix(line, reader.readerOptions.CommentString) { - if reader.readerOptions.CommentHandling == cli.PassComments { - recordsAndContexts.PushBack(types.NewOutputString(line+"\n", context)) - continue - } else if reader.readerOptions.CommentHandling == cli.SkipComments { - continue - } - // else comments are data - } - } - - record := reader.recordFromNIDXLine(line) - context.UpdateForInputRecord() - recordAndContext := types.NewRecordAndContext(record, context) - recordsAndContexts.PushBack(recordAndContext) - } - - return recordsAndContexts, false -} - - -func (reader *RecordReaderNIDX) recordFromNIDXLine( - line string, -) *types.Mlrmap { - record := types.NewMlrmapAsRecord() - - var values []string - if reader.readerOptions.IFSRegex == nil { // e.g. --no-ifs-regex - values = lib.SplitString(line, reader.readerOptions.IFS) - } else { - values = lib.RegexSplitString(reader.readerOptions.IFSRegex, line, -1) - } - if reader.readerOptions.AllowRepeatIFS { - values = lib.StripEmpties(values) // left/right trim - } - - var i int = 0 - for _, value := range values { - i++ - key := strconv.Itoa(i) - mval := types.MlrvalFromInferredTypeForDataFiles(value) - record.PutReference(key, mval) - } - return record -}