merge 4.0.0-RC

22 files changed, 926 insertions, 526 deletions

diff --git a/internal/mapr/aggregateset.go b/internal/mapr/aggregateset.go
index a6cc6eb..c50c7a1 100644
--- a/internal/mapr/aggregateset.go
+++ b/internal/mapr/aggregateset.go
@@ -5,6 +5,10 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+
+	"github.com/mimecast/dtail/internal/io/dlog"
+	"github.com/mimecast/dtail/internal/io/pool"
+	"github.com/mimecast/dtail/internal/protocol"
 )
 
 // AggregateSet represents aggregated key/value pairs from the
@@ -33,8 +37,7 @@ func (s *AggregateSet) String() string {
 // Merge one aggregate set into this one.
 func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error {
 	s.Samples += set.Samples
-	//logger.Trace("Merge", set)
-
+	//dlog.Common.Trace("Merge", set)
 	for _, sc := range query.Select {
 		storage := sc.FieldStorage
 		switch sc.Operation {
@@ -66,24 +69,27 @@ func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error {
 
 // Serialize the aggregate set so it can be sent over the wire.
 func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- string) {
-	//logger.Trace("Serialising mapr.AggregateSet", s)
-	var sb strings.Builder
+	dlog.Common.Trace("Serialising mapr.AggregateSet", s)
+	sb := pool.BuilderBuffer.Get().(*strings.Builder)
+	defer pool.RecycleBuilderBuffer(sb)
 
 	sb.WriteString(groupKey)
-	sb.WriteString("➔")
-	sb.WriteString(fmt.Sprintf("%d➔", s.Samples))
+	sb.WriteString(protocol.AggregateDelimiter)
+	sb.WriteString(fmt.Sprintf("%d", s.Samples))
+	sb.WriteString(protocol.AggregateDelimiter)
 
 	for k, v := range s.FValues {
 		sb.WriteString(k)
-		sb.WriteString("=")
-		sb.WriteString(fmt.Sprintf("%v➔", v))
+		sb.WriteString(protocol.AggregateKVDelimiter)
+		sb.WriteString(fmt.Sprintf("%v", v))
+		sb.WriteString(protocol.AggregateDelimiter)
 	}
 
 	for k, v := range s.SValues {
 		sb.WriteString(k)
-		sb.WriteString("=")
+		sb.WriteString(protocol.AggregateKVDelimiter)
 		sb.WriteString(v)
-		sb.WriteString("➔")
+		sb.WriteString(protocol.AggregateDelimiter)
 	}
 
 	select {
@@ -108,7 +114,6 @@ func (s *AggregateSet) addFloatMin(key string, value float64) {
 		s.FValues[key] = value
 		return
 	}
-
 	if f > value {
 		s.FValues[key] = value
 	}
@@ -121,7 +126,6 @@ func (s *AggregateSet) addFloatMax(key string, value float64) {
 		s.FValues[key] = value
 		return
 	}
-
 	if f < value {
 		s.FValues[key] = value
 	}
@@ -140,7 +144,6 @@ func (s *AggregateSet) setFloat(key string, value float64) {
 // Aggregate data to the aggregate set.
 func (s *AggregateSet) Aggregate(key string, agg AggregateOperation, value string, clientAggregation bool) (err error) {
 	var f float64
-
 	// First check if we can aggregate anything without converting value to float.
 	switch agg {
 	case Count:
diff --git a/internal/mapr/client/aggregate.go b/internal/mapr/client/aggregate.go
index 10b34d4..1704d43 100644
--- a/internal/mapr/client/aggregate.go
+++ b/internal/mapr/client/aggregate.go
@@ -1,11 +1,13 @@
 package client
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
-	"github.com/mimecast/dtail/internal/io/logger"
+	"github.com/mimecast/dtail/internal/io/dlog"
 	"github.com/mimecast/dtail/internal/mapr"
+	"github.com/mimecast/dtail/internal/protocol"
 )
 
 // Aggregate mapreduce data on the DTail client side.
@@ -21,7 +23,9 @@ type Aggregate struct {
 }
 
 // NewAggregate create new client aggregator.
-func NewAggregate(server string, query *mapr.Query, globalGroup *mapr.GlobalGroupSet) *Aggregate {
+func NewAggregate(server string, query *mapr.Query,
+	globalGroup *mapr.GlobalGroupSet) *Aggregate {
+
 	return &Aggregate{
 		query:       query,
 		group:       mapr.NewGroupSet(),
@@ -31,20 +35,26 @@ func NewAggregate(server string, query *mapr.Query, globalGroup *mapr.GlobalGrou
 }
 
 // Aggregate data from mapr log line into local (and global) group sets.
-func (a *Aggregate) Aggregate(parts []string) {
+func (a *Aggregate) Aggregate(message string) error {
+	parts := strings.Split(message, protocol.AggregateDelimiter)
+	if len(parts) < 4 {
+		return fmt.Errorf("aggregate message without any real data")
+	}
+
 	groupKey := parts[0]
 	samples, err := strconv.Atoi(parts[1])
 	if err != nil {
-		logger.FatalExit("Unable to parse sample count", parts[1], err, parts)
+		return fmt.Errorf("unable to parse sample count '%s': %v", parts[1], err)
 	}
+
 	fields := a.makeFields(parts[2:])
 	set := a.group.GetSet(groupKey)
-
 	var addedSamples bool
+
 	for _, sc := range a.query.Select {
 		if val, ok := fields[sc.FieldStorage]; ok {
 			if err := set.Aggregate(sc.FieldStorage, sc.Operation, val, true); err != nil {
-				logger.Error(err)
+				dlog.Client.Error(err)
 				continue
 			}
 			addedSamples = true
@@ -63,19 +73,18 @@ func (a *Aggregate) Aggregate(parts []string) {
 		// Re-init local group (make it empty again).
 		a.group.InitSet()
 	}
+	return nil
 }
 
 // Create a map of key-value pairs from a part list such as ["foo=bar",  "bar=baz"].
 func (a *Aggregate) makeFields(parts []string) map[string]string {
 	fields := make(map[string]string, len(parts))
-
 	for _, part := range parts {
-		kv := strings.SplitN(part, "=", 2)
-		if len(kv) < 2 {
+		kv := strings.SplitN(part, protocol.AggregateKVDelimiter, 2)
+		if len(kv) != 2 {
 			continue
 		}
 		fields[kv[0]] = kv[1]
 	}
-
 	return fields
 }
diff --git a/internal/mapr/funcs/function.go b/internal/mapr/funcs/function.go
index 1a89c3a..418d86f 100644
--- a/internal/mapr/funcs/function.go
+++ b/internal/mapr/funcs/function.go
@@ -19,13 +19,12 @@ type Function struct {
 // FunctionStack is a list of functions stacked each other
 type FunctionStack []Function
 
-// NewFunctionStack parses the input string, e.g. foo(bar("arg")) and returns a corresponding function stack.
+// NewFunctionStack parses the input string, e.g. foo(bar("arg")) and returns
+// a corresponding function stack.
 func NewFunctionStack(in string) (FunctionStack, string, error) {
 	var fs FunctionStack
-
 	getCallback := func(name string) (CallbackFunc, error) {
 		var cb CallbackFunc
-
 		switch name {
 		case "md5sum":
 			return Md5Sum, nil
@@ -51,17 +50,15 @@ func NewFunctionStack(in string) (FunctionStack, string, error) {
 		fs = append(fs, Function{name, call})
 		aux = aux[index+1 : len(aux)-1]
 	}
-
 	return fs, aux, nil
 }
 
 // Call the function stack.
 func (fs FunctionStack) Call(str string) string {
 	for i := len(fs) - 1; i >= 0; i-- {
-		//logger.Debug("Call", fs[i].Name, str)
+		//dlog.Common.Debug("Call", fs[i].Name, str)
 		str = fs[i].call(str)
-		//logger.Debug("Call.result", fs[i].Name, str)
+		//dlog.Common.Debug("Call.result", fs[i].Name, str)
 	}
-
 	return str
 }
diff --git a/internal/mapr/funcs/function_test.go b/internal/mapr/funcs/function_test.go
index 415683c..8b5d8b7 100644
--- a/internal/mapr/funcs/function_test.go
+++ b/internal/mapr/funcs/function_test.go
@@ -6,16 +6,19 @@ func TestFunction(t *testing.T) {
 	input := "md5sum($line)"
 	fs, arg, err := NewFunctionStack(input)
 	if err != nil {
-		t.Errorf("error parsing function input '%s': %s (%v)\n", input, err.Error(), fs)
+		t.Errorf("error parsing function input '%s': %s (%v)\n",
+			input, err.Error(), fs)
 	}
 	if arg != "$line" {
-		t.Errorf("error parsing function input '%s': expected argument '$line' but got '%s' (%v)\n", input, arg, fs)
+		t.Errorf("error parsing function input '%s': expected argument '$line' but "+
+			"got '%s' (%v)\n", input, arg, fs)
 	}
 	t.Log(input, fs, arg)
 
 	result := fs.Call(input)
 	if result != "b38699013d79e50d9d122433753959c1" {
-		t.Errorf("error executing function stack '%s': expected result 'b38699013d79e50d9d122433753959c1' but got '%s' (%v)\n", input, result, fs)
+		t.Errorf("error executing function stack '%s': expected result "+
+			"'b38699013d79e50d9d122433753959c1' but got '%s' (%v)\n", input, result, fs)
 	}
 
 	input = "maskdigits(md5sum(maskdigits($line)))"
@@ -24,22 +27,26 @@ func TestFunction(t *testing.T) {
 		t.Errorf("error parsing function input '%s': %s (%v)\n", input, err.Error(), fs)
 	}
 	if arg != "$line" {
-		t.Errorf("error parsing function input '%s': expected argument '$line' but got '%s' (%v)\n", input, arg, fs)
+		t.Errorf("error parsing function input '%s': expected argument '$line' but "+
+			"got '%s' (%v)\n", input, arg, fs)
 	}
 	t.Log(input, fs, arg)
 
 	result = fs.Call(input)
 	if result != ".fac.bbe..bb.........d...a.c..b." {
-		t.Errorf("error executing function stack '%s': expected result '.fac.bbe..bb.........d...a.c..b.' but got '%s' (%v)\n", input, result, fs)
+		t.Errorf("error executing function stack '%s': expected result "+
+			"'.fac.bbe..bb.........d...a.c..b.' but got '%s' (%v)\n", input, result, fs)
 	}
 
 	input = "md5sum$line)"
 	if fs, _, err := NewFunctionStack(input); err == nil {
-		t.Errorf("Expected error parsing function input '%s' (%v) but got no error\n", input, fs)
+		t.Errorf("Expected error parsing function input '%s' (%v) but got no error\n",
+			input, fs)
 	}
 
 	input = "md5sum(makedigits$line))"
 	if fs, _, err := NewFunctionStack(input); err == nil {
-		t.Errorf("Expected error parsing function input '%s' (%v) but got no error\n", input, fs)
+		t.Errorf("Expected error parsing function input '%s' (%v) but got no error\n",
+			input, fs)
 	}
 }
diff --git a/internal/mapr/funcs/maskdigits.go b/internal/mapr/funcs/maskdigits.go
index d51f3d8..925ec4d 100644
--- a/internal/mapr/funcs/maskdigits.go
+++ b/internal/mapr/funcs/maskdigits.go
@@ -3,12 +3,10 @@ package funcs
 // MaskDigits masks all digits (replaces them with .)
 func MaskDigits(input string) string {
 	s := []byte(input)
-
 	for i, b := range s {
 		if '0' <= b && b <= '9' {
 			s[i] = '.'
 		}
 	}
-
 	return string(s)
 }
diff --git a/internal/mapr/globalgroupset.go b/internal/mapr/globalgroupset.go
index cfab506..2d7f10b 100644
--- a/internal/mapr/globalgroupset.go
+++ b/internal/mapr/globalgroupset.go
@@ -17,7 +17,6 @@ func NewGlobalGroupSet() *GlobalGroupSet {
 		semaphore: make(chan struct{}, 1),
 	}
 	g.InitSet()
-
 	return &g
 }
 
@@ -30,7 +29,6 @@ func (g *GlobalGroupSet) String() string {
 func (g *GlobalGroupSet) Merge(query *Query, group *GroupSet) error {
 	g.semaphore <- struct{}{}
 	defer func() { <-g.semaphore }()
-
 	return g.merge(query, group)
 }
 
@@ -54,7 +52,6 @@ func (g *GlobalGroupSet) merge(query *Query, group *GroupSet) error {
 			return err
 		}
 	}
-
 	return nil
 }
 
@@ -67,7 +64,6 @@ func (g *GlobalGroupSet) IsEmpty() bool {
 func (g *GlobalGroupSet) NumSets() int {
 	g.semaphore <- struct{}{}
 	defer func() { <-g.semaphore }()
-
 	return len(g.sets)
 }
 
@@ -79,7 +75,6 @@ func (g *GlobalGroupSet) SwapOut() *GroupSet {
 
 	set := &GroupSet{sets: g.sets}
 	g.InitSet()
-
 	return set
 }
 
@@ -87,14 +82,12 @@ func (g *GlobalGroupSet) SwapOut() *GroupSet {
 func (g *GlobalGroupSet) WriteResult(query *Query) error {
 	g.semaphore <- struct{}{}
 	defer func() { <-g.semaphore }()
-
 	return g.GroupSet.WriteResult(query)
 }
 
 // Result returns the result of the mapreduce aggregation as a string.
-func (g *GlobalGroupSet) Result(query *Query) (string, int, error) {
+func (g *GlobalGroupSet) Result(query *Query, rowsLimit int) (string, int, error) {
 	g.semaphore <- struct{}{}
 	defer func() { <-g.semaphore }()
-
-	return g.GroupSet.Result(query)
+	return g.GroupSet.Result(query, rowsLimit)
 }
diff --git a/internal/mapr/groupset.go b/internal/mapr/groupset.go
index b5c8a48..9d7661a 100644
--- a/internal/mapr/groupset.go
+++ b/internal/mapr/groupset.go
@@ -2,15 +2,9 @@ package mapr
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"io/ioutil"
-	"os"
 	"sort"
 	"strconv"
-	"strings"
-
-	"github.com/mimecast/dtail/internal/io/logger"
 )
 
 // GroupSet represents a map of aggregate sets. The group sets
@@ -22,6 +16,14 @@ type GroupSet struct {
 	sets map[string]*AggregateSet
 }
 
+// Internal helper type
+type result struct {
+	groupKey     string
+	values       []string
+	columnWidths []int
+	orderBy      float64
+}
+
 // NewGroupSet returns a new empty group set.
 func NewGroupSet() *GroupSet {
 	g := GroupSet{}
@@ -56,139 +58,91 @@ func (g *GroupSet) Serialize(ctx context.Context, ch chan<- string) {
 	}
 }
 
-// Result returns a nicely formated result of the query from the group set.
-func (g *GroupSet) Result(query *Query) (string, int, error) {
-	return g.limitedResult(query, query.Limit, "\t", " ", false)
-}
-
-// WriteResult writes the result to an outfile.
-func (g *GroupSet) WriteResult(query *Query) error {
-	if !query.HasOutfile() {
-		return errors.New("No outfile specified")
-	}
-
-	// -1: Don't limit the result, include all data sets
-	result, _, err := g.limitedResult(query, query.Limit, "", ",", true)
-	if err != nil {
-		return err
-	}
-
-	logger.Info("Writing outfile", query.Outfile)
-	tmpOutfile := fmt.Sprintf("%s.tmp", query.Outfile)
-
-	if err := ioutil.WriteFile(tmpOutfile, []byte(result), 0644); err != nil {
-		return err
-	}
-
-	if err := os.Rename(tmpOutfile, query.Outfile); err != nil {
-		os.Remove(tmpOutfile)
-		return err
-	}
-
-	return nil
-}
+// Return a sorted result slice of the query from the group set.
+func (g *GroupSet) result(query *Query, gathercolumnWidths bool) ([]result, []int, error) {
+	var err error
+	var rows []result
 
-// Return a nicely formated result of the query from the group set.
-func (g *GroupSet) limitedResult(query *Query, limit int, lineStarter, fieldSeparator string, addHeader bool) (string, int, error) {
-	type result struct {
-		groupKey  string
-		resultStr string
-		orderBy   float64
-	}
-
-	var resultSlice []result
+	// Helpers for calculating the ASCII table output (output is the terminal and
+	// not a CSV file).
+	columnWidths := make([]int, len(query.Select))
+	var valueStrLen int
 
 	for groupKey, set := range g.sets {
-		var sb strings.Builder
-		r := result{groupKey: groupKey}
+		result := result{groupKey: groupKey}
 
-		lastIndex := len(query.Select) - 1
 		for i, sc := range query.Select {
-			storage := sc.FieldStorage
-			orderByThis := storage == query.OrderBy
-
-			switch sc.Operation {
-			case Count:
-				value := set.FValues[storage]
-				sb.WriteString(fmt.Sprintf("%d", int(value)))
-				if orderByThis {
-					r.orderBy = value
-				}
-			case Len:
-				fallthrough
-			case Sum:
-				fallthrough
-			case Min:
-				fallthrough
-			case Max:
-				value := set.FValues[storage]
-				sb.WriteString(fmt.Sprintf("%f", value))
-				if orderByThis {
-					r.orderBy = value
-				}
-			case Last:
-				value := set.SValues[storage]
-				if orderByThis {
-					f, err := strconv.ParseFloat(value, 64)
-					if err == nil {
-						r.orderBy = f
-					}
-				}
-				sb.WriteString(value)
-			case Avg:
-				value := set.FValues[storage] / float64(set.Samples)
-				sb.WriteString(fmt.Sprintf("%f", value))
-				if orderByThis {
-					r.orderBy = value
-				}
-			default:
-				return "", 0, fmt.Errorf("Unknown aggregation method '%v'", sc.Operation)
+			if valueStrLen, err = g.resultSelect(query, &sc, set, &result); err != nil {
+				return rows, columnWidths, err
+			}
+
+			// Do we want to gather the table withs? This is required to print out a decent
+			// ASCII formated table (table output is the terminal and not a CSV file).
+			if !gathercolumnWidths {
+				continue
+			}
+			if columnWidths[i] < len(sc.FieldStorage) {
+				columnWidths[i] = len(sc.FieldStorage)
 			}
-			if i != lastIndex {
-				sb.WriteString(fieldSeparator)
+			if columnWidths[i] < valueStrLen {
+				columnWidths[i] = valueStrLen
 			}
 		}
+		rows = append(rows, result)
+	}
+
+	g.resultOrderBy(query, rows)
+	return rows, columnWidths, nil
+}
 
-		r.resultStr = sb.String()
-		resultSlice = append(resultSlice, r)
+func (*GroupSet) resultSelect(query *Query, sc *selectCondition, set *AggregateSet,
+	result *result) (int, error) {
+
+	var valueStr string
+	var value float64
+
+	switch sc.Operation {
+	case Count:
+		value = set.FValues[sc.FieldStorage]
+		valueStr = fmt.Sprintf("%d", int(value))
+	case Len:
+		fallthrough
+	case Sum:
+		fallthrough
+	case Min:
+		fallthrough
+	case Max:
+		value = set.FValues[sc.FieldStorage]
+		valueStr = fmt.Sprintf("%f", value)
+	case Last:
+		valueStr = set.SValues[sc.FieldStorage]
+		value, _ = strconv.ParseFloat(valueStr, 64)
+	case Avg:
+		value = set.FValues[sc.FieldStorage] / float64(set.Samples)
+		valueStr = fmt.Sprintf("%f", value)
+	default:
+		return 0, fmt.Errorf("Unknown aggregation method '%v'", sc.Operation)
 	}
 
-	if query.OrderBy != "" {
-		if query.ReverseOrder {
-			sort.SliceStable(resultSlice, func(i, j int) bool {
-				return resultSlice[i].orderBy < resultSlice[j].orderBy
-			})
-		} else {
-			sort.SliceStable(resultSlice, func(i, j int) bool {
-				return resultSlice[i].orderBy > resultSlice[j].orderBy
-			})
-		}
+	if sc.FieldStorage == query.OrderBy {
+		result.orderBy = value
 	}
+	result.values = append(result.values, valueStr)
 
-	var sb strings.Builder
+	return len(valueStr), nil
+}
 
-	// Write header first
-	if addHeader {
-		lastIndex := len(query.Select) - 1
-		sb.WriteString(lineStarter)
-		for i, sc := range query.Select {
-			sb.WriteString(sc.FieldStorage)
-			if i != lastIndex {
-				sb.WriteString(fieldSeparator)
-			}
-		}
-		sb.WriteString("\n")
+func (*GroupSet) resultOrderBy(query *Query, rows []result) {
+	if query.OrderBy == "" {
+		return
 	}
-
-	// And now write the data
-	for i, r := range resultSlice {
-		if i == limit {
-			break
-		}
-		sb.WriteString(lineStarter)
-		sb.WriteString(r.resultStr)
-		sb.WriteString("\n")
+	if query.ReverseOrder {
+		sort.SliceStable(rows, func(i, j int) bool {
+			return rows[i].orderBy < rows[j].orderBy
+		})
+	} else {
+		sort.SliceStable(rows, func(i, j int) bool {
+			return rows[i].orderBy > rows[j].orderBy
+		})
 	}
-
-	return sb.String(), len(resultSlice), nil
 }
diff --git a/internal/mapr/groupsetresult.go b/internal/mapr/groupsetresult.go
new file mode 100644
index 0000000..6d0ac1f
--- /dev/null
+++ b/internal/mapr/groupsetresult.go
@@ -0,0 +1,237 @@
+package mapr
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/mimecast/dtail/internal/color"
+	"github.com/mimecast/dtail/internal/config"
+	"github.com/mimecast/dtail/internal/io/dlog"
+	"github.com/mimecast/dtail/internal/io/pool"
+	"github.com/mimecast/dtail/internal/protocol"
+)
+
+// Result returns a nicely formated result of the query from the group set.
+func (g *GroupSet) Result(query *Query, rowsLimit int) (string, int, error) {
+	rows, columnWidths, err := g.result(query, true)
+	if err != nil {
+		return "", 0, err
+	}
+	if query.Limit != -1 {
+		rowsLimit = query.Limit
+	}
+	lastColumn := len(query.Select) - 1
+
+	sb := pool.BuilderBuffer.Get().(*strings.Builder)
+	defer pool.RecycleBuilderBuffer(sb)
+
+	g.resultWriteFormattedHeader(query, sb, lastColumn, rowsLimit, columnWidths)
+	g.resultWriteFormattedHeaderRowSeparator(query, sb, lastColumn, columnWidths)
+	g.resultWriteFormattedData(query, sb, lastColumn, rowsLimit, columnWidths, rows)
+
+	return sb.String(), len(rows), nil
+}
+
+// Write a nicely formatted header for the result data.
+func (g *GroupSet) resultWriteFormattedHeader(query *Query, sb *strings.Builder,
+	lastColumn, rowsLimit int, columnWidths []int) {
+
+	for i, sc := range query.Select {
+		format := fmt.Sprintf(" %%%ds ", columnWidths[i])
+		str := fmt.Sprintf(format, sc.FieldStorage)
+
+		g.resultWriteFormattedHeaderEntry(query, sb, sc, str)
+		if i == lastColumn {
+			continue
+		}
+		g.resultWriteFormattedHeaderEntrySeparator(query, sb)
+
+	}
+	sb.WriteString("\n")
+}
+
+func (g *GroupSet) resultWriteFormattedHeaderEntry(query *Query, sb *strings.Builder,
+	sc selectCondition, str string) {
+
+	if config.Client.TermColorsEnable {
+		attrs := []color.Attribute{config.Client.TermColors.MaprTable.HeaderAttr}
+		if sc.FieldStorage == query.OrderBy {
+			attrs = append(attrs, config.Client.TermColors.MaprTable.HeaderSortKeyAttr)
+		}
+		for _, groupBy := range query.GroupBy {
+			if sc.FieldStorage == groupBy {
+				attrs = append(attrs, config.Client.TermColors.MaprTable.HeaderGroupKeyAttr)
+				break
+			}
+		}
+		color.PaintWithAttrs(sb, str,
+			config.Client.TermColors.MaprTable.HeaderFg,
+			config.Client.TermColors.MaprTable.HeaderBg,
+			attrs)
+
+	} else {
+		sb.WriteString(str)
+	}
+}
+
+func (g *GroupSet) resultWriteFormattedHeaderEntrySeparator(query *Query, sb *strings.Builder) {
+	if config.Client.TermColorsEnable {
+		color.PaintWithAttr(sb, protocol.FieldDelimiter,
+			config.Client.TermColors.MaprTable.HeaderDelimiterFg,
+			config.Client.TermColors.MaprTable.HeaderDelimiterBg,
+			config.Client.TermColors.MaprTable.HeaderDelimiterAttr)
+	} else {
+		sb.WriteString(protocol.FieldDelimiter)
+	}
+}
+
+// This writes a nicely formatted line separating the header and the data.
+func (g *GroupSet) resultWriteFormattedHeaderRowSeparator(query *Query, sb *strings.Builder,
+	lastColumn int, columnWidths []int) {
+
+	for i := 0; i < len(query.Select); i++ {
+		str := fmt.Sprintf("-%s-", strings.Repeat("-", columnWidths[i]))
+		if config.Client.TermColorsEnable {
+			color.PaintWithAttr(sb, str,
+				config.Client.TermColors.MaprTable.HeaderDelimiterFg,
+				config.Client.TermColors.MaprTable.HeaderDelimiterBg,
+				config.Client.TermColors.MaprTable.HeaderDelimiterAttr)
+		} else {
+			sb.WriteString(str)
+		}
+		if i == lastColumn {
+			continue
+		}
+		if config.Client.TermColorsEnable {
+			color.PaintWithAttr(sb, protocol.FieldDelimiter,
+				config.Client.TermColors.MaprTable.HeaderDelimiterFg,
+				config.Client.TermColors.MaprTable.HeaderDelimiterBg,
+				config.Client.TermColors.MaprTable.HeaderDelimiterAttr)
+		} else {
+			sb.WriteString(protocol.FieldDelimiter)
+		}
+	}
+	sb.WriteString("\n")
+}
+
+// Write the result data nicely formatted.
+func (g *GroupSet) resultWriteFormattedData(query *Query, sb *strings.Builder,
+	lastColumn, rowsLimit int, columnWidths []int, rows []result) {
+
+	for i, r := range rows {
+		if i == rowsLimit {
+			break
+		}
+		for j, value := range r.values {
+			g.resultWriteFormattedDataEntry(query, sb, columnWidths, j, value)
+			if j == lastColumn {
+				continue
+			}
+			// Now, write the data entry separator.
+			if config.Client.TermColorsEnable {
+				color.PaintWithAttr(sb, protocol.FieldDelimiter,
+					config.Client.TermColors.MaprTable.DelimiterFg,
+					config.Client.TermColors.MaprTable.DelimiterBg,
+					config.Client.TermColors.MaprTable.DelimiterAttr)
+			} else {
+				sb.WriteString(protocol.FieldDelimiter)
+			}
+		}
+		sb.WriteString("\n")
+	}
+}
+
+func (g *GroupSet) resultWriteFormattedDataEntry(query *Query, sb *strings.Builder,
+	columnWidths []int, j int, value string) {
+
+	format := fmt.Sprintf(" %%%ds ", columnWidths[j])
+	str := fmt.Sprintf(format, value)
+	if config.Client.TermColorsEnable {
+		color.PaintWithAttr(sb, str,
+			config.Client.TermColors.MaprTable.DataFg,
+			config.Client.TermColors.MaprTable.DataBg,
+			config.Client.TermColors.MaprTable.DataAttr)
+	} else {
+		sb.WriteString(str)
+	}
+}
+
+func (*GroupSet) writeQueryFile(query *Query) error {
+	queryFile := fmt.Sprintf("%s.query", query.Outfile)
+	tmpQueryFile := fmt.Sprintf("%s.tmp", queryFile)
+	dlog.Common.Debug("Writing query file", queryFile)
+
+	fd, err := os.Create(tmpQueryFile)
+	if err != nil {
+		return err
+	}
+	defer fd.Close()
+
+	fd.WriteString(query.RawQuery)
+	os.Rename(tmpQueryFile, queryFile)
+	return nil
+}
+
+// WriteResult writes the result to an CSV outfile.
+func (g *GroupSet) WriteResult(query *Query) error {
+	if !query.HasOutfile() {
+		return errors.New("No outfile specified")
+	}
+	if err := g.writeQueryFile(query); err != nil {
+		return err
+	}
+	rows, _, err := g.result(query, false)
+	if err != nil {
+		return err
+	}
+
+	dlog.Common.Info("Writing outfile", query.Outfile)
+	tmpOutfile := fmt.Sprintf("%s.tmp", query.Outfile)
+
+	fd, err := os.Create(tmpOutfile)
+	if err != nil {
+		return err
+	}
+	defer fd.Close()
+
+	return g.resultWriteUnformatted(query, rows, tmpOutfile, fd)
+}
+
+func (g *GroupSet) resultWriteUnformatted(query *Query, rows []result, tmpOutfile string,
+	fd *os.File) error {
+
+	// Generate header now
+	lastColumn := len(query.Select) - 1
+	for i, sc := range query.Select {
+		fd.WriteString(sc.FieldStorage)
+		if i == lastColumn {
+			continue
+		}
+		fd.WriteString(protocol.CSVDelimiter)
+	}
+	fd.WriteString("\n")
+
+	// And now write the data
+	for i, r := range rows {
+		if i == query.Limit {
+			break
+		}
+		for j, value := range r.values {
+			fd.WriteString(value)
+			if j == lastColumn {
+				continue
+			}
+			fd.WriteString(protocol.CSVDelimiter)
+		}
+		fd.WriteString("\n")
+	}
+
+	if err := os.Rename(tmpOutfile, query.Outfile); err != nil {
+		os.Remove(tmpOutfile)
+		return err
+	}
+
+	return nil
+}
diff --git a/internal/mapr/logformat/default.go b/internal/mapr/logformat/default.go
index 44bf558..f066f6e 100644
--- a/internal/mapr/logformat/default.go
+++ b/internal/mapr/logformat/default.go
@@ -1,26 +1,56 @@
 package logformat
 
 import (
-	"errors"
+	"fmt"
 	"strings"
+
+	"github.com/mimecast/dtail/internal/protocol"
 )
 
-// MakeFieldsDEFAULT is the default log file mapreduce parser.
+// MakeFieldsDEFAULT is the default DTail log file key-value parser.
 func (p *Parser) MakeFieldsDEFAULT(maprLine string) (map[string]string, error) {
-	fields := make(map[string]string, 20)
-	splitted := strings.Split(maprLine, "|")
+	splitted := strings.Split(maprLine, protocol.FieldDelimiter)
+
+	if len(splitted) < 11 || !strings.HasPrefix(splitted[9], "MAPREDUCE:") ||
+		!strings.HasPrefix(splitted[0], "INFO") {
+		// Not a DTail mapreduce log line.
+		return nil, ErrIgnoreFields
+	}
+
+	fields := make(map[string]string, len(splitted)+8)
 
 	fields["*"] = "*"
 	fields["$line"] = maprLine
 	fields["$empty"] = ""
 	fields["$hostname"] = p.hostname
+	fields["$server"] = p.hostname
 	fields["$timezone"] = p.timeZoneName
 	fields["$timeoffset"] = p.timeZoneOffset
 
-	for _, kv := range splitted {
+	fields["$severity"] = splitted[0]
+	fields["$loglevel"] = splitted[0]
+
+	time := splitted[1]
+	fields["$time"] = time
+	if len(time) == 15 {
+		// Example: 20211002-071209
+		fields["$date"] = time[0:8]
+		fields["$hour"] = time[9:11]
+		fields["$minute"] = time[11:13]
+		fields["$second"] = time[13:]
+	}
+	fields["$pid"] = splitted[2]
+	fields["$caller"] = splitted[3]
+	fields["$cpus"] = splitted[4]
+	fields["$goroutines"] = splitted[5]
+	fields["$cgocalls"] = splitted[6]
+	fields["$loadavg"] = splitted[7]
+	fields["$uptime"] = splitted[8]
+
+	for _, kv := range splitted[10:] {
 		keyAndValue := strings.SplitN(kv, "=", 2)
 		if len(keyAndValue) != 2 {
-			return fields, errors.New("Error parsing mapr token: " + kv)
+			return fields, fmt.Errorf("Unable to parse key-value token '%s'", kv)
 		}
 		fields[strings.ToLower(keyAndValue[0])] = keyAndValue[1]
 	}
diff --git a/internal/mapr/logformat/default_test.go b/internal/mapr/logformat/default_test.go
index 10ec8b7..28e1acc 100644
--- a/internal/mapr/logformat/default_test.go
+++ b/internal/mapr/logformat/default_test.go
@@ -1,6 +1,7 @@
 package logformat
 
 import (
+	"fmt"
 	"testing"
 )
 
@@ -10,26 +11,83 @@ func TestDefaultLogFormat(t *testing.T) {
 		t.Erro