diff options
Diffstat (limited to 'internal/mapr')
| -rw-r--r-- | internal/mapr/globalgroupset.go | 4 | ||||
| -rw-r--r-- | internal/mapr/groupsetresult.go | 8 |
2 files changed, 6 insertions, 6 deletions
diff --git a/internal/mapr/globalgroupset.go b/internal/mapr/globalgroupset.go index 2d7f10b..2b12898 100644 --- a/internal/mapr/globalgroupset.go +++ b/internal/mapr/globalgroupset.go @@ -79,10 +79,10 @@ func (g *GlobalGroupSet) SwapOut() *GroupSet { } // WriteResult writes the result of a mapreduce aggregation to an outfile. -func (g *GlobalGroupSet) WriteResult(query *Query) error { +func (g *GlobalGroupSet) WriteResult(query *Query, finalResult bool) error { g.semaphore <- struct{}{} defer func() { <-g.semaphore }() - return g.GroupSet.WriteResult(query) + return g.GroupSet.WriteResult(query, finalResult) } // Result returns the result of the mapreduce aggregation as a string. diff --git a/internal/mapr/groupsetresult.go b/internal/mapr/groupsetresult.go index 58663b8..47bdab8 100644 --- a/internal/mapr/groupsetresult.go +++ b/internal/mapr/groupsetresult.go @@ -176,7 +176,7 @@ func (*GroupSet) writeQueryFile(query *Query) error { } // WriteResult writes the result to an CSV outfile. -func (g *GroupSet) WriteResult(query *Query) error { +func (g *GroupSet) WriteResult(query *Query, finalResult bool) error { if !query.HasOutfile() { return errors.New("No outfile specified") } @@ -204,7 +204,7 @@ func (g *GroupSet) WriteResult(query *Query) error { } defer fd.Close() - return g.resultWriteUnformatted(query, rows, fd, writeHeader) + return g.resultWriteUnformatted(query, rows, fd, writeHeader, finalResult) } func (g *GroupSet) getOutfileFD(query *Query) (*os.File, error) { @@ -218,7 +218,7 @@ func (g *GroupSet) getOutfileFD(query *Query) (*os.File, error) { return os.OpenFile(query.Outfile.FilePath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666) } -func (g *GroupSet) resultWriteUnformatted(query *Query, rows []result, fd *os.File, writeHeader bool) error { +func (g *GroupSet) resultWriteUnformatted(query *Query, rows []result, fd *os.File, writeHeader, finalResult bool) error { lastColumn := len(query.Select) - 1 if writeHeader { @@ -248,7 +248,7 @@ func (g *GroupSet) resultWriteUnformatted(query *Query, rows []result, fd *os.Fi } } - if !query.Outfile.AppendMode { + if !query.Outfile.AppendMode && finalResult { tmpOutfile := fmt.Sprintf("%s.tmp", query.Outfile.FilePath) if err := os.Rename(tmpOutfile, query.Outfile.FilePath); err != nil { os.Remove(tmpOutfile) |