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

Dedupe DKVP-reader and NIDX-reader source files

Browse source
This commit is contained in:
John Kerl 2021-12-10 22:02:20 -05:00
parent 2618d63e6a
commit 4df477f734
3 changed files with 58 additions and 180 deletions
Download patch file Download diff file Expand all files Collapse all files

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

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

62
internal/pkg/input/record_reader_dkvp.go → internal/pkg/input/record_reader_dkvp_nidx.go
View file

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

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

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