Merge branch 'master' of git@github.com:klbostee/dumbo

dumbo/core.py

@@ -277,7 +277,7 @@ def run(self):
                      ' -m dumbo encodepipe -file ' + ' -file '.join(inputs)
         if addedopts['inputformat'] and addedopts['inputformat'][0] == 'code':
             encodepipe += ' -alreadycoded yes'
-        if addedopts['addpath'] and addedopts['addpath'][0] == 'yes':
+        if addedopts['addpath'] and addedopts['addpath'][0] != 'no':
             encodepipe += ' -addpath yes'
         if addedopts['numreducetasks'] and addedopts['numreducetasks'][0] == '0':
             retval = execute("%s | %s %s %s %s > '%s'" % (encodepipe,
@@ -409,7 +409,7 @@ def run(self):
         if outputformat_shortcuts.has_key(outputformat.lower()):
             outputformat = outputformat_shortcuts[outputformat.lower()]
         self.opts.append(('outputformat', outputformat))
-        if addedopts['addpath'] and addedopts['addpath'][0] == 'yes':
+        if addedopts['addpath'] and addedopts['addpath'][0] != 'no':
             self.opts.append(('cmdenv', 'dumbo_addpath=true'))
         pyenv = envdef('PYTHONPATH',
                        addedopts['libegg'],
@@ -603,12 +603,14 @@ def run(mapper,
             print >> sys.stderr, 'ERROR: no output path given'
             sys.exit(1)
         preoutputsopt = getopt(opts, 'preoutputs')
+        addpathopt = getopt(opts, 'addpath', delete=False)
         if iter != 0:
             newopts['input'] = outputopt[0] + "_pre" + str(iter)
             if not (preoutputsopt and preoutputsopt[0] == 'yes'):
                 newopts['delinputs'] = 'yes'
             newopts['inputformat'] = 'code'
-            newopts['addpath'] = 'no'
+            if addpathopt and addpathopt[0] == 'yes':  # not when == 'iter'
+                newopts['addpath'] = 'no'
         if iter < itercnt - 1:
             newopts['output'] = outputopt[0] + "_pre" + str(iter + 1)
             newopts['outputformat'] = 'code'

dumbo/lib.py

@@ -99,7 +99,7 @@ def __new__(cls):
 
     def __init__(self):
         self._mappers = []
-        self.opts = [("addpath", "yes")]
+        self.opts = [("addpath", "iter")]
 
     def itermappers(self):
         for pattern, mapper in self._mappers:
@@ -143,9 +143,23 @@ class JoinReducer(object):
 
     def __call__(self, key, values):
         if key.isprimary:
-            self.primary(key.body, values)
+            output = self.primary(key.body, values)
+            if output:
+                for k, v in output:
+                    jk = copy(key)
+                    jk.body = k
+                    yield jk, v
         else:
             for k, v in self.secondary(key.body, values):
                 jk = copy(key)
                 jk.body = k
                 yield jk, v
+
+    def primary(self, key, values):
+        for value in values:
+            yield key, value
+
+    def secondary(self, key, values):
+        for value in values:
+            yield key, value
+

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name = 'dumbo',
-    version = '0.21.13',
+    version = '0.21.14',
     author = 'Klaas Bosteels',
     author_email = 'klaas@last.fm',
     license = 'Apache Software License (ASF)',