diff --git a/extract_dataset_parts.sh b/extract_dataset_parts.sh
new file mode 100644
index 00000000000..26f915c0c3d
--- /dev/null
+++ b/extract_dataset_parts.sh
@@ -0,0 +1,8 @@
+#!/bin/sh
+
+cd `dirname $0`
+for file in $1/split_info*.json
+do
+    # echo processing $file
+    python ./scripts/extract_dataset_part.py $file
+done
diff --git a/lib/galaxy/datatypes/data.py b/lib/galaxy/datatypes/data.py
index b26670c41a0..6172713a54a 100644
--- a/lib/galaxy/datatypes/data.py
+++ b/lib/galaxy/datatypes/data.py
@@ -360,9 +360,12 @@ class Data( object ):
         but might be brittle. Need to revisit this.
         """
         if len(split_files) == 1:
-            os.system( 'mv -f %s %s' % ( split_files[0], output_file ) )
+            cmd = 'mv -f %s %s' % ( split_files[0], output_file ) 
         else:
-            os.system( 'cat %s > %s' % ( ' '.join(split_files), output_file ) )
+            cmd = 'cat %s > %s' % ( ' '.join(split_files), output_file ) 
+        result = os.system(cmd)
+        if result != 0:
+            raise Exception('Result %s from %s' % (result, cmd))
     merge = staticmethod(merge)
 
 class Text( Data ):
@@ -533,6 +536,13 @@ class Text( Data ):
         f.close()
     split = staticmethod(split)
 
+class LineCount( Text ):
+    """
+    Dataset contains a single line with a single integer that denotes the
+    line count for a related dataset. Used for custom builds.
+    """
+    pass
+
 class Newick( Text ):
     pass
 
diff --git a/lib/galaxy/datatypes/sequence.py b/lib/galaxy/datatypes/sequence.py
index 80e1324965e..feedebeeaa3 100644
--- a/lib/galaxy/datatypes/sequence.py
+++ b/lib/galaxy/datatypes/sequence.py
@@ -7,6 +7,7 @@ import data
 import logging
 import re
 import string
+import os
 from cgi import escape
 from galaxy.datatypes.metadata import MetadataElement
 from galaxy.datatypes import metadata
@@ -14,8 +15,52 @@ import galaxy.model
 from galaxy import util
 from sniff import *
 
+import pkg_resources
+pkg_resources.require("simplejson")
+import simplejson
+
 log = logging.getLogger(__name__)
 
+class SequenceSplitLocations( data.Text ):
+    """
+    Class storing information about a sequence file composed of multiple gzip files concatenated as
+    one OR an uncompressed file. In the GZIP case, each sub-file's location is stored in start and end. 
+    The format of the file is JSON:
+    { "sections" : [
+            { "start" : "x", "end" : "y", "clusters" : "z" },
+            ...
+    ]}
+    """
+    def set_peek( self, dataset, is_multi_byte=False ):
+        if not dataset.dataset.purged:
+            try:
+                parsed_data = simplejson.load(open(dataset.file_name))
+                # dataset.peek = simplejson.dumps(data, sort_keys=True, indent=4)
+                dataset.peek = data.get_file_peek( dataset.file_name, is_multi_byte=is_multi_byte )
+                dataset.blurb = '%d sections' % len(parsed_data['sections'])
+            except Exception, e:
+                dataset.peek = 'Not FQTOC file'
+                dataset.blurb = 'Not FQTOC file'
+        else:
+            dataset.peek = 'file does not exist'
+            dataset.blurb = 'file purged from disk'
+
+    file_ext = "fqtoc"
+
+    def sniff( self, filename ):
+        if os.path.getsize(filename) < 50000:
+            try:
+                data = simplejson.load(open(filename))
+                sections = data['sections']
+                for section in sections:
+                    if 'start' not in section or 'end' not in section or 'clusters' not in section:
+                        return False
+                return True
+            except:
+                pass
+        return False
+
+
 class Sequence( data.Text ):
     """Class describing a sequence"""
 
@@ -50,143 +95,237 @@ class Sequence( data.Text ):
         else:
             dataset.peek = 'file does not exist'
             dataset.blurb = 'file purged from disk'
+
+    def get_sequences_per_file(total_clusters, split_params):
+        if split_params['split_mode'] == 'number_of_parts':
+            # legacy basic mode - split into a specified number of parts
+            parts = int(split_params['split_size'])
+            sequences_per_file = [total_clusters/parts for i in range(parts)]
+            for i in range(total_clusters % parts):
+                sequences_per_file[i] += 1
+        elif split_params['split_mode'] == 'to_size':
+            # loop through the sections and calculate the number of clusters
+            chunk_size = long(split_params['split_size'])
+
+            chunks = total_clusters / chunk_size
+            rem = total_clusters % chunk_size
+            sequences_per_file = [chunk_size for i in range(total_clusters / chunk_size)]
+            # TODO: Should we invest the time in a better way to handle small remainders?
+            if rem > 0:
+                sequences_per_file.append(rem)
+        else:
+            raise Exception('Unsupported split mode %s' % split_params['split_mode'])
+        return sequences_per_file
+    get_sequences_per_file = staticmethod(get_sequences_per_file)
+
+    def do_slow_split( cls, input_datasets, subdir_generator_function, split_params):
+        # count the clusters so we can split
+        # TODO: if metadata is present, take the number of lines / 4
+        if input_datasets[0].metadata is not None and input_datasets[0].metadata.sequences is not None:
+            total_clusters = input_datasets[0].metadata.sequences
+        else:
+            input_file = input_datasets[0].file_name
+            compress = is_gzip(input_file)
+            if compress:
+                # gzip is really slow before python 2.7!
+                in_file = gzip.GzipFile(input_file, 'r')
+            else:
+                # TODO
+                # if a file is not compressed, seek locations can be calculated and stored
+                # ideally, this would be done in metadata
+                # TODO
+                # Add BufferedReader if python 2.7?
+                in_file = open(input_file, 'rt')
+            total_clusters = long(0)
+            for i, line in enumerate(in_file):
+                total_clusters += 1
+            in_file.close()
+            total_clusters /= 4
+
+        sequences_per_file = cls.get_sequences_per_file(total_clusters, split_params)
+        return cls.write_split_files(input_datasets, None, subdir_generator_function, sequences_per_file)
+    do_slow_split = classmethod(do_slow_split)
+
+    def do_fast_split( cls, input_datasets, toc_file_datasets, subdir_generator_function, split_params):
+        data = simplejson.load(open(toc_file_datasets[0].file_name))
+        sections = data['sections']
+        total_clusters = long(0)
+        for section in sections:
+            total_clusters += long(section['clusters'])
+        sequences_per_file = cls.get_sequences_per_file(total_clusters, split_params)
+        return cls.write_split_files(input_datasets, toc_file_datasets, subdir_generator_function, sequences_per_file)
+    do_fast_split = classmethod(do_fast_split)
+
+    def write_split_files(cls, input_datasets, toc_file_datasets, subdir_generator_function, sequences_per_file):
+        directories = []
+        def get_subdir(idx):
+            if idx < len(directories):
+                return directories[idx]
+            dir = subdir_generator_function()
+            directories.append(dir)
+            return dir
+
+        # we know how many splits and how many clusters in each. What remains is to write out instructions for the 
+        # splitting of all the input files. To decouple the format of those instructions from this code, the exact format of
+        # those instructions is delegated to scripts
+        start_sequence=0
+        for part_no in range(len(sequences_per_file)):
+            dir = get_subdir(part_no)
+            for ds_no in range(len(input_datasets)):
+                ds = input_datasets[ds_no]
+                base_name = os.path.basename(ds.file_name)
+                part_path = os.path.join(dir, base_name)
+                split_data = dict(class_name='%s.%s' % (cls.__module__, cls.__name__),
+                                  output_name=part_path,
+                                  input_name=ds.file_name,
+                                  args=dict(start_sequence=start_sequence, num_sequences=sequences_per_file[part_no]))
+                if toc_file_datasets is not None:
+                    toc = toc_file_datasets[ds_no]
+                    split_data['args']['toc_file'] = toc.file_name
+                f = open(os.path.join(dir, 'split_info_%s.json' % base_name), 'w')
+                simplejson.dump(split_data, f)
+                f.close()
+            start_sequence += sequences_per_file[part_no]
+        return directories
+    write_split_files = classmethod(write_split_files)
     
-    def split( input_files, subdir_generator_function, split_params):
+    def split( cls, input_datasets, subdir_generator_function, split_params):
         """
         FASTQ files are split on cluster boundaries, in increments of 4 lines
         """
         if split_params is None:
-            return
-        
-        def split_calculate_clusters( input_file, get_dir, default_clusters):
-            """
-            Split the 0th file into even sized chunks, and return the number of clusters in each
-            """
-            compress = is_gzip(input_file)
-            if compress:
-# TODO: Python 2.4, 2.5 don't have io.BufferedReader!!!
-# add a buffered reader because gzip is really slow before python 2.7
-                in_file = gzip.GzipFile(input_file, 'r')
-            else:
-                in_file = open(input_file, 'rt')
-            part_file = None
-            part = 0
-            local_clusters_per_file = []
-            for i, line in enumerate(in_file):
-                cluster_number,  line_in_cluster = divmod(i, 4)
-                current_part, remainder = divmod(cluster_number, default_clusters)
-                
-                if (current_part != part or part_file is None):
-                    if (part_file):
-                        part_file.close()
-                    part = current_part
-                    part_dir = get_dir()
-                    part_path = os.path.join(part_dir, os.path.basename(input_file))
-# TODO: If the input was compressed, compress the output?
-                    part_file = open(part_path, 'w')
-                    local_clusters_per_file.append(default_clusters)
-                part_file.write(line)
-            if (part_file):
-                part_file.close()
-            in_file.close()
-            local_clusters_per_file[part] = remainder + 1
-            return local_clusters_per_file
-    
-        def split_to_size(input_file, get_dir, clusters_per_file):
-            """
-            Split the files beyond the 0th to the same number of clusters as the 0th.
-            This is used to split in a variety of ways, so these are both legal for
-            clusters_per_file:
-                [ 10000, 10000, 10000, 10000, 2 ] # to_size=10000, 40002 total
-                [ 10001, 10001, 10000, 10000 ] # number_of_parts = 4, 40002 total
+            return None
 
-            """
-            compress = is_gzip(input_file)
-            if compress:
-# TODO: Python 2.4, 2.5 don't have io.BufferedReader!!!
-# add a buffered reader because gzip is really slow before python 2.7
-                in_file = gzip.GzipFile(input_file, 'r')
-            else:
-                in_file = open(input_file, 'rt')
-            part_file = None
-            part = 0
-            clusters_this_part = 0
-            for i, line in enumerate(in_file):
-                cluster_number,  line_in_cluster = divmod(i, 4)
-                if clusters_this_part  == clusters_per_file[part]:
-                    current_part = part + 1
-                else:
-                    current_part = part
-                
-                if (current_part != part or part_file is None):
-                    if (part_file):
-                        part_file.close()
-                    part = current_part
-                    clusters_this_part = 0
-                    part_dir = get_dir()
-                    part_path = os.path.join(part_dir, os.path.basename(input_file))
-# TODO: If the input was compressed, compress the output?
-                    part_file = open(part_path, 'w')
-                    if clusters_per_file is None and part > 0:
-                        local_clusters_per_file.append(default_clusters)
-                part_file.write(line)
-                if line_in_cluster == 3:
-                    clusters_this_part += 1
-            if (part_file):
-                part_file.close()
-            in_file.close()
+        # first, see if there are any associated FQTOC files that will give us the split locations
+        # if so, we don't need to read the files to do the splitting
+        toc_file_datasets = []
+        for ds in input_datasets:
+            tmp_ds = ds
+            fqtoc_file = None
+            while fqtoc_file is None and tmp_ds is not None:
+                fqtoc_file = tmp_ds.get_converted_files_by_type('fqtoc')
+                tmp_ds = tmp_ds.copied_from_library_dataset_dataset_association
 
-        directories = []
-        def create_subdir():
-            dir = subdir_generator_function()
-            directories.append(dir)
-            return dir
+            if fqtoc_file is not None:
+                toc_file_datasets.append(fqtoc_file)
+
+        if len(toc_file_datasets) == len(input_datasets):
+            return cls.do_fast_split(input_datasets, toc_file_datasets, subdir_generator_function, split_params)
+        return cls.do_slow_split(input_datasets, subdir_generator_function, split_params)
+    split = classmethod(split)
+
+    def process_split_file(data):
+        """
+        This is called in the context of an external process launched by a Task (possibly not on the Galaxy machine)
+        to create the input files for the Task. The parameters:
+        data - a dict containing the contents of the split file
+        """
+        args = data['args']
+        input_name = data['input_name']
+        output_name = data['output_name']
+        start_sequence = long(args['start_sequence'])
+        sequence_count = long(args['num_sequences'])
         
-        clusters_per_file = None
-        if split_params['split_mode'] == 'number_of_parts':
-            # legacy splitting. To keep things simple, just scan the 0th file and count the clusters,
-            # then split it
-            clusters_per_file = []
-            in_file = open(input_files[0], 'rt')
-            for i, line in enumerate(in_file):
-                pass
-            in_file.close()
-            length = (i+1)/4
-            
-            if length <= 0:
-                raise Exception('Invalid sequence file %s' % input_files[0])
-            parts = int(split_params['split_size'])
-            if length < parts:
-                parts = length
-            len_each, remainder = divmod(length, parts)
-            while length > 0:
-                chunk = len_each
-                if remainder > 0:
-                    chunk += 1
-                clusters_per_file.append(chunk)
-                remainder=- 1
-                length -= chunk
-            split_to_size(input_files[0],  create_subdir,  clusters_per_file)
-        elif split_params['split_mode'] == 'to_size':
-            # split one file and see what the cluster sizes turn out to be
-            clusters_per_file = split_calculate_clusters(input_files[0],  create_subdir,  
-                int(split_params['split_size']))
+        if 'toc_file' in args:
+            toc_file = simplejson.load(open(args['toc_file'], 'r'))
+            commands = Sequence.get_split_commands_with_toc(input_name, output_name, toc_file, start_sequence, sequence_count)
         else:
-            raise Exception('Unsupported split mode %s' % split_params['split_mode'])
+            commands = Sequence.get_split_commands_sequential(is_gzip(input_name), input_name, output_name, start_sequence, sequence_count)
+        for cmd in commands:
+            if 0 != os.system(cmd):
+                raise Exception("Executing '%s' failed" % cmd)
+        return True
+    process_split_file = staticmethod(process_split_file)
+    
+    def get_split_commands_with_toc(input_name, output_name, toc_file, start_sequence, sequence_count):
+        """
+        Uses a Table of Contents dict, parsed from an FQTOC file, to come up with a set of
+        shell commands that will extract the parts necessary
+        >>> three_sections=[dict(start=0, end=74, clusters=10), dict(start=74, end=148, clusters=10), dict(start=148, end=148+76, clusters=10)]
+        >>> Sequence.get_split_commands_with_toc('./input.gz', './output.gz', dict(sections=three_sections), start_sequence=0, sequence_count=10)
+        ['dd bs=1 skip=0 count=74 if=./input.gz 2> /dev/null >> ./output.gz']
+        >>> Sequence.get_split_commands_with_toc('./input.gz', './output.gz', dict(sections=three_sections), start_sequence=1, sequence_count=5)
+        ['(dd bs=1 skip=0 count=74 if=./input.gz 2> /dev/null )| zcat | ( tail -n +5 2> /dev/null) | head -20 | gzip -c >> ./output.gz']
+        >>> Sequence.get_split_commands_with_toc('./input.gz', './output.gz', dict(sections=three_sections), start_sequence=0, sequence_count=20)
+        ['dd bs=1 skip=0 count=148 if=./input.gz 2> /dev/null >> ./output.gz']
+        >>> Sequence.get_split_commands_with_toc('./input.gz', './output.gz', dict(sections=three_sections), start_sequence=5, sequence_count=10)
+        ['(dd bs=1 skip=0 count=74 if=./input.gz 2> /dev/null )| zcat | ( tail -n +21 2> /dev/null) | head -20 | gzip -c >> ./output.gz', '(dd bs=1 skip=74 count=74 if=./input.gz 2> /dev/null )| zcat | ( tail -n +1 2> /dev/null) | head -20 | gzip -c >> ./output.gz']
+        >>> Sequence.get_split_commands_with_toc('./input.gz', './output.gz', dict(sections=three_sections), start_sequence=10, sequence_count=10)
+        ['dd bs=1 skip=74 count=74 if=./input.gz 2> /dev/null >> ./output.gz']
+        >>> Sequence.get_split_commands_with_toc('./input.gz', './output.gz', dict(sections=three_sections), start_sequence=5, sequence_count=20)
+        ['(dd bs=1 skip=0 count=74 if=./input.gz 2> /dev/null )| zcat | ( tail -n +21 2> /dev/null) | head -20 | gzip -c >> ./output.gz', 'dd bs=1 skip=74 count=74 if=./input.gz 2> /dev/null >> ./output.gz', '(dd bs=1 skip=148 count=76 if=./input.gz 2> /dev/null )| zcat | ( tail -n +1 2> /dev/null) | head -20 | gzip -c >> ./output.gz']
+        """
+        sections = toc_file['sections']
+        result = []
+
+        current_sequence = long(0)
+        i=0
+        # skip to the section that contains my starting sequence
+        while i < len(sections) and start_sequence >= current_sequence + long(sections[i]['clusters']):
+            current_sequence += long(sections[i]['clusters'])
+            i += 1
+        if i == len(sections): # bad input data!
+            raise Exception('No FQTOC section contains starting sequence %s' % start_sequence)
+    
+        # These two variables act as an accumulator for consecutive entire blocks that
+        # can be copied verbatim (without decompressing)
+        start_chunk = long(-1)
+        end_chunk = long(-1)
+        copy_chunk_cmd = 'dd bs=1 skip=%s count=%s if=%s 2> /dev/null >> %s'
         
-        # split the rest, using the same number of clusters for each file
-        current_dir_idx = [0] # use a list to get around Python 2.x lame closure support
-        def get_subdir():
-            if len(directories) <= current_dir_idx[0]:
-                raise Exception('FASTQ files do not have the same number of clusters - splitting failed')
-            result = directories[current_dir_idx[0]]
-            current_dir_idx[0] = current_dir_idx[0] + 1
-            return result
-            
-        for i in range(1,  len(input_files)):
-            current_dir_idx[0] = 0
-            split_to_size(input_files[i],  get_subdir,  clusters_per_file)
-    split = staticmethod(split)
+        while sequence_count > 0 and i < len(sections):
+            # we need to extract partial data. So, find the byte offsets of the chunks that contain the data we need
+            # use a combination of dd (to pull just the right sections out) tail (to skip lines) and head (to get the 
+            # right number of lines
+            sequences = long(sections[i]['clusters'])
+            skip_sequences = start_sequence-current_sequence
+            sequences_to_extract = min(sequence_count, sequences-skip_sequences)
+            start_copy = long(sections[i]['start'])
+            end_copy = long(sections[i]['end'])
+            if sequences_to_extract < sequences:
+                if start_chunk > -1:
+                    result.append(copy_chunk_cmd % (start_chunk, end_chunk-start_chunk, input_name, output_name))
+                    start_chunk = -1
+                # extract, unzip, trim, recompress
+                result.append('(dd bs=1 skip=%s count=%s if=%s 2> /dev/null )| zcat | ( tail -n +%s 2> /dev/null) | head -%s | gzip -c >> %s' % 
+                    (start_copy, end_copy-start_copy, input_name, skip_sequences*4+1, sequences_to_extract*4, output_name))
+            else: # whole section - add it to the start_chunk/end_chunk accumulator
+                if start_chunk == -1:
+                    start_chunk = start_copy
+                end_chunk = end_copy
+            sequence_count -= sequences_to_extract
+            start_sequence += sequences_to_extract
+            current_sequence += sequences
+            i += 1
+        if start_chunk > -1:
+            result.append(copy_chunk_cmd % (start_chunk, end_chunk-start_chunk, input_name, output_name))
+                
+        if sequence_count > 0:
+            raise Exception('%s sequences not found in file' % sequence_count)
+             
+        return result
+    get_split_commands_with_toc = staticmethod(get_split_commands_with_toc)
+
+
+    def get_split_commands_sequential(is_compressed, input_name, output_name, start_sequence, sequence_count):
+        """
+        Does a brain-dead sequential scan & extract of certain sequences
+        >>> Sequence.get_split_commands_sequential(True, './input.gz', './output.gz', start_sequence=0, sequence_count=10)
+        ['zcat "./input.gz" | ( tail -n +1 2> /dev/null) | head -40 | gzip -c > "./output.gz"']
+        >>> Sequence.get_split_commands_sequential(False, './input.fastq', './output.fastq', start_sequence=10, sequence_count=10)
+        ['tail -n +41 "./input.fastq" 2> /dev/null | head -40 > "./output.fastq"']
+        """
+        start_line = start_sequence * 4
+        line_count = sequence_count * 4
+        # TODO: verify that tail can handle 64-bit numbers
+        if is_compressed:
+            cmd = 'zcat "%s" | ( tail -n +%s 2> /dev/null) | head -%s | gzip -c' % (input_name, start_line+1, line_count)
+        else:
+            cmd = 'tail -n +%s "%s" 2> /dev/null | head -%s'  % (start_line+1, input_name, line_count)
+        cmd += ' > "%s"' % output_name
+        
+        return [cmd]
+    get_split_commands_sequential = staticmethod(get_split_commands_sequential)
 
 
 
@@ -690,3 +829,7 @@ class Lav( data.Text ):
                 return False
         except:
             return False
+
+if __name__ == '__main__':
+    import doctest, sys
+    doctest.testmod(sys.modules[__name__])
\ No newline at end of file
diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index 65283dcf132..ca97c24c7d7 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -32,9 +32,9 @@ TOOL_PROVIDED_JOB_METADATA_FILE = 'galaxy.json'
 class JobManager( object ):
     """
     Highest level interface to job management.
-
+    
     TODO: Currently the app accesses "job_queue" and "job_stop_queue" directly.
-          This should be decoupled.
+          This should be decoupled. 
     """
     def __init__( self, app ):
         self.app = app
@@ -71,7 +71,7 @@ class Sleeper( object ):
 
 class JobQueue( object ):
     """
-    Job manager, waits for jobs to be runnable and then dispatches to
+    Job manager, waits for jobs to be runnable and then dispatches to 
     a JobRunner.
     """
     STOP_SIGNAL = object()
@@ -95,7 +95,7 @@ class JobQueue( object ):
         self.running = True
         self.dispatcher = dispatcher
         self.monitor_thread = threading.Thread( target=self.__monitor )
-        self.monitor_thread.start()
+        self.monitor_thread.start()        
         log.info( "job manager started" )
         if app.config.get_bool( 'enable_job_recovery', True ):
             self.__check_jobs_at_startup()
@@ -132,7 +132,7 @@ class JobQueue( object ):
 
     def __monitor( self ):
         """
-        Continually iterate the waiting jobs, checking is each is ready to
+        Continually iterate the waiting jobs, checking is each is ready to 
         run and dispatching if so.
         """
         # HACK: Delay until after forking, we need a way to do post fork notification!!!
@@ -180,12 +180,12 @@ class JobQueue( object ):
                     jobs_to_check.append( self.sa_session.query( model.Job ).get( job_id ) )
             except Empty:
                 pass
-        # Iterate over new and waiting jobs and look for any that are
+        # Iterate over new and waiting jobs and look for any that are 
         # ready to run
         new_waiting_jobs = []
         for job in jobs_to_check:
             try:
-                # Check the job's dependencies, requeue if they're not done
+                # Check the job's dependencies, requeue if they're not done                    
                 job_state = self.__check_if_ready_to_run( job )
                 if job_state == JOB_WAIT:
                     if not self.track_jobs_in_database:
@@ -216,7 +216,7 @@ class JobQueue( object ):
         self.waiting_jobs = new_waiting_jobs
         # Done with the session
         self.sa_session.remove()
-
+        
     def __check_if_ready_to_run( self, job ):
         """
         Check if a job is ready to run by verifying that each of its input
@@ -281,13 +281,13 @@ class JobQueue( object ):
         if len( user_jobs ) >= self.app.config.user_job_limit:
             return JOB_WAIT
         return JOB_READY
-
+            
     def put( self, job_id, tool ):
         """Add a job to the queue (by job identifier)"""
         if not self.track_jobs_in_database:
             self.queue.put( ( job_id, tool.id ) )
             self.sleeper.wake()
-
+    
     def shutdown( self ):
         """Attempts to gracefully shut down the worker thread"""
         if self.parent_pid != os.getpid():
@@ -304,7 +304,7 @@ class JobQueue( object ):
 
 class JobWrapper( object ):
     """
-    Wraps a 'model.Job' with convenience methods for running processes and
+    Wraps a 'model.Job' with convenience methods for running processes and 
     state management.
     """
     def __init__( self, job, queue ):
@@ -330,15 +330,15 @@ class JobWrapper( object ):
         self.output_dataset_paths = None
         self.tool_provided_job_metadata = None
         # Wrapper holding the info required to restore and clean up from files used for setting metadata externally
-        self.external_output_metadata = metadata.JobExternalOutputMetadataWrapper( job )
-
+        self.external_output_metadata = metadata.JobExternalOutputMetadataWrapper( job )        
+    
     def get_job( self ):
         return self.sa_session.query( model.Job ).get( self.job_id )
-
+    
     def get_id_tag(self):
         # For compatability with drmaa, which uses job_id right now, and TaskWrapper
         return str(self.job_id)
-
+    
     def get_param_dict( self ):
         """
         Restore the dictionary of parameters from the database.
@@ -347,10 +347,10 @@ class JobWrapper( object ):
         param_dict = dict( [ ( p.name, p.value ) for p in job.parameters ] )
         param_dict = self.tool.params_from_strings( param_dict, self.app )
         return param_dict
-
+        
     def get_version_string_path( self ):
         return os.path.abspath(os.path.join(self.app.config.new_file_path, "GALAXY_VERSION_STRING_%s" % self.job_id))
-
+        
     def prepare( self ):
         """
         Prepare the job to run by creating the working directory and the
@@ -372,9 +372,9 @@ class JobWrapper( object ):
         out_data = dict( [ ( da.name, da.dataset ) for da in job.output_datasets ] )
         inp_data.update( [ ( da.name, da.dataset ) for da in job.input_library_datasets ] )
         out_data.update( [ ( da.name, da.dataset ) for da in job.output_library_datasets ] )
-
-        # Set up output dataset association for export history jobs. Because job
-        # uses a Dataset rather than an HDA or LDA, it's necessary to set up a
+        
+        # Set up output dataset association for export history jobs. Because job 
+        # uses a Dataset rather than an HDA or LDA, it's necessary to set up a 
         # fake dataset association that provides the needed attributes for
         # preparing a job.
         class FakeDatasetAssociation ( object ):
@@ -401,7 +401,7 @@ class JobWrapper( object ):
         # ( this used to be performed in the "exec_before_job" hook, but hooks are deprecated ).
         self.tool.exec_before_job( self.queue.app, inp_data, out_data, param_dict )
         # Run the before queue ("exec_before_job") hook
-        self.tool.call_hook( 'exec_before_job', self.queue.app, inp_data=inp_data,
+        self.tool.call_hook( 'exec_before_job', self.queue.app, inp_data=inp_data, 
                              out_data=out_data, tool=self.tool, param_dict=incoming)
         self.sa_session.flush()
         # Build any required config files
@@ -434,7 +434,7 @@ class JobWrapper( object ):
 
     def fail( self, message, exception=False ):
         """
-        Indicate job failure by setting state and message on all output
+        Indicate job failure by setting state and message on all output 
         datasets.
         """
         job = self.get_job()
@@ -480,7 +480,7 @@ class JobWrapper( object ):
         if self.tool:
             self.tool.job_failed( self, message, exception )
         self.cleanup()
-
+        
     def change_state( self, state, info = False ):
         job = self.get_job()
         self.sa_session.refresh( job )
@@ -510,12 +510,12 @@ class JobWrapper( object ):
         job.job_runner_external_id = external_id
         self.sa_session.add( job )
         self.sa_session.flush()
-
+        
     def finish( self, stdout, stderr ):
         """
-        Called to indicate that the associated command has been run. Updates
+        Called to indicate that the associated command has been run. Updates 
         the output datasets based on stderr and stdout from the command, and
-        the contents of the output files.
+        the contents of the output files. 
         """
         # default post job setup
         self.sa_session.expunge_all()
@@ -537,7 +537,7 @@ class JobWrapper( object ):
             if os.path.exists(version_filename):
                 self.version_string = open(version_filename).read()
                 os.unlink(version_filename)
-
+            
         if self.app.config.outputs_to_working_directory:
             for dataset_path in self.get_output_fnames():
                 try:
@@ -585,7 +585,7 @@ class JobWrapper( object ):
                             else:
                                 # Security violation.
                                 log.exception( "from_work_dir specified a location not in the working directory: %s, %s" % ( source_file, self.working_directory ) )
-
+            
                 dataset.blurb = 'done'
                 dataset.peek  = 'no peek'
                 dataset.info  = context['stdout'] + context['stderr']
@@ -600,7 +600,7 @@ class JobWrapper( object ):
                         dataset.init_meta( copy_from=dataset )
                     #if a dataset was copied, it won't appear in our dictionary:
                     #either use the metadata from originating output dataset, or call set_meta on the copies
-                    #it would be quicker to just copy the metadata from the originating output dataset,
+                    #it would be quicker to just copy the metadata from the originating output dataset, 
                     #but somewhat trickier (need to recurse up the copied_from tree), for now we'll call set_meta()
                     if not self.app.config.set_metadata_externally or \
                      ( not self.external_output_metadata.external_metadata_set_successfully( dataset, self.sa_session ) \
@@ -612,7 +612,7 @@ class JobWrapper( object ):
                         #load metadata from file
                         #we need to no longer allow metadata to be edited while the job is still running,
                         #since if it is edited, the metadata changed on the running output will no longer match
-                        #the metadata that was stored to disk for use via the external process,
+                        #the metadata that was stored to disk for use via the external process, 
                         #and the changes made by the user will be lost, without warning or notice
                         dataset.metadata.from_JSON_dict( self.external_output_metadata.get_output_filenames_by_dataset( dataset, self.sa_session ).filename_out )
                     try:
@@ -653,13 +653,13 @@ class JobWrapper( object ):
         # Flush all the dataset and job changes above.  Dataset state changes
         # will now be seen by the user.
         self.sa_session.flush()
-        # Save stdout and stderr
+        # Save stdout and stderr    
         if len( stdout ) > 32768:
             log.error( "stdout for job %d is greater than 32K, only first part will be logged to database" % job.id )
         job.stdout = stdout[:32768]
         if len( stderr ) > 32768:
             log.error( "stderr for job %d is greater than 32K, only first part will be logged to database" % job.id )
-        job.stderr = stderr[:32768]
+        job.stderr = stderr[:32768]  
         # custom post process setup
         inp_data = dict( [ ( da.name, da.dataset ) for da in job.input_datasets ] )
         out_data = dict( [ ( da.name, da.dataset ) for da in job.output_datasets ] )
@@ -676,8 +676,8 @@ class JobWrapper( object ):
         # ( this used to be performed in the "exec_after_process" hook, but hooks are deprecated ).
         self.tool.exec_after_process( self.queue.app, inp_data, out_data, param_dict, job = job )
         # Call 'exec_after_process' hook
-        self.tool.call_hook( 'exec_after_process', self.queue.app, inp_data=inp_data,
-                             out_data=out_data, param_dict=param_dict,
+        self.tool.call_hook( 'exec_after_process', self.queue.app, inp_data=inp_data, 
+                             out_data=out_data, param_dict=param_dict, 
                              tool=self.tool, stdout=stdout, stderr=stderr )
         job.command_line = self.command_line
 
@@ -696,7 +696,7 @@ class JobWrapper( object ):
         self.sa_session.flush()
         log.debug( 'job %d ended' % self.job_id )
         self.cleanup()
-
+        
     def cleanup( self ):
         # remove temporary files
         try:
@@ -710,10 +710,10 @@ class JobWrapper( object ):
             galaxy.tools.imp_exp.JobImportHistoryArchiveWrapper( self.job_id ).cleanup_after_job( self.sa_session )
         except:
             log.exception( "Unable to cleanup job %d" % self.job_id )
-
+        
     def get_command_line( self ):
         return self.command_line
-
+    
     def get_session_id( self ):
         return self.session_id
 
@@ -884,13 +884,17 @@ class TaskWrapper(JobWrapper):
     Should be refactored into a generalized executable unit wrapper parent, then jobs and tasks.
     """
     # Abstract this to be more useful for running tasks that *don't* necessarily compose a job.
-
+    
     def __init__(self, task, queue):
         super(TaskWrapper, self).__init__(task.job, queue)
         self.task_id = task.id
         self.working_directory = task.working_directory
+        if task.prepare_input_files_cmd is not None:
+            self.prepare_input_files_cmds = [ task.prepare_input_files_cmd ]
+        else:
+            self.prepare_input_files_cmds = None
         self.status = task.states.NEW
-
+       
     def get_job( self ):
         if self.job_id:
             return self.sa_session.query( model.Job ).get( self.job_id )
@@ -953,7 +957,7 @@ class TaskWrapper(JobWrapper):
         # ( this used to be performed in the "exec_before_job" hook, but hooks are deprecated ).
         self.tool.exec_before_job( self.queue.app, inp_data, out_data, param_dict )
         # Run the before queue ("exec_before_job") hook
-        self.tool.call_hook( 'exec_before_job', self.queue.app, inp_data=inp_data,
+        self.tool.call_hook( 'exec_before_job', self.queue.app, inp_data=inp_data, 
                              out_data=out_data, tool=self.tool, param_dict=incoming)
         self.sa_session.flush()
         # Build any required config files
@@ -1000,12 +1004,12 @@ class TaskWrapper(JobWrapper):
         task.state = state
         self.sa_session.add( task )
         self.sa_session.flush()
-
+            
     def get_state( self ):
         task = self.get_task()
         self.sa_session.refresh( task )
         return task.state
-
+        
     def set_runner( self, runner_url, external_id ):
         task = self.get_task()
         self.sa_session.refresh( task )
@@ -1014,15 +1018,15 @@ class TaskWrapper(JobWrapper):
         # DBTODO Check task job_runner_stuff
         self.sa_session.add( task )
         self.sa_session.flush()
-
+        
     def finish( self, stdout, stderr ):
         # DBTODO integrate previous finish logic.
         # Simple finish for tasks.  Just set the flag OK.
         log.debug( 'task %s for job %d ended' % (self.task_id, self.job_id) )
         """
-        Called to indicate that the associated command has been run. Updates
+        Called to indicate that the associated command has been run. Updates 
         the output datasets based on stderr and stdout from the command, and
-        the contents of the output files.
+        the contents of the output files. 
         """
         # default post job setup_external_metadata
         self.sa_session.expunge_all()
@@ -1039,7 +1043,7 @@ class TaskWrapper(JobWrapper):
             task.state = task.states.ERROR
         else:
             task.state = task.states.OK
-        # Save stdout and stderr
+        # Save stdout and stderr    
         if len( stdout ) > 32768:
             log.error( "stdout for task %d is greater than 32K, only first part will be logged to database" % task.id )
         task.stdout = stdout[:32768]
@@ -1053,7 +1057,7 @@ class TaskWrapper(JobWrapper):
     def cleanup( self ):
         # There is no task cleanup.  The job cleans up for all tasks.
         pass
-
+        
     def get_command_line( self ):
         return self.command_line
 
@@ -1063,7 +1067,7 @@ class TaskWrapper(JobWrapper):
     def get_output_file_id( self, file ):
         # There is no permanent output file for tasks.
         return None
-
+        
     def get_tool_provided_job_metadata( self ):
         # DBTODO Handle this as applicable for tasks.
         return None
@@ -1085,7 +1089,7 @@ class TaskWrapper(JobWrapper):
     def setup_external_metadata( self, exec_dir = None, tmp_dir = None, dataset_files_path = None, config_root = None, datatypes_config = None, set_extension = True, **kwds ):
         # There is no metadata setting for tasks.  This is handled after the merge, at the job level.
         return ""
-
+        
 class DefaultJobDispatcher( object ):
     def __init__( self, app ):
         self.app = app
@@ -1115,7 +1119,7 @@ class DefaultJobDispatcher( object ):
             runner = getattr( module, obj )
             self.job_runners[name] = runner( self.app )
             log.debug( 'Loaded job runner: %s' % display_name )
-
+            
     def put( self, job_wrapper ):
         try:
             if self.app.config.use_tasked_jobs and job_wrapper.tool.parallelism is not None:
@@ -1126,8 +1130,8 @@ class DefaultJobDispatcher( object ):
                     self.job_runners[runner_name].put( job_wrapper )
                 else:
                     runner_name = "tasks"
-                    log.debug( "dispatching job %d to %s runner" %( job_wrapper.job_id, runner_name ) )
-                    self.job_runners[runner_name].put( job_wrapper )
+                log.debug( "dispatching job %d to %s runner" %( job_wrapper.job_id, runner_name ) )
+                self.job_runners[runner_name].put( job_wrapper )
             else:
                 runner_name = ( job_wrapper.tool.job_runner.split(":", 1) )[0]
                 log.debug( "dispatching job %d to %s runner" %( job_wrapper.job_id, runner_name ) )
@@ -1183,7 +1187,7 @@ class JobStopQueue( object ):
         self.sleeper = Sleeper()
         self.running = True
         self.monitor_thread = threading.Thread( target=self.monitor )
-        self.monitor_thread.start()
+        self.monitor_thread.start()        
         log.info( "job stopper started" )
 
     def monitor( self ):
@@ -1263,4 +1267,3 @@ class NoopQueue( object ):
         return
     def shutdown( self ):
         return
-
diff --git a/lib/galaxy/jobs/runners/__init__.py b/lib/galaxy/jobs/runners/__init__.py
index 597d517309f..51d714b1296 100644
--- a/lib/galaxy/jobs/runners/__init__.py
+++ b/lib/galaxy/jobs/runners/__init__.py
@@ -6,6 +6,7 @@ class BaseJobRunner( object ):
         Compose the sequence of commands necessary to execute a job. This will
         currently include:
             - environment settings corresponding to any requirement tags
+            - preparing input files
             - command line taken from job wrapper
             - commands to set metadata (if include_metadata is True)
         """
@@ -17,10 +18,13 @@ class BaseJobRunner( object ):
         # Prepend version string
         if job_wrapper.version_string_cmd:
             commands = "%s &> %s; " % ( job_wrapper.version_string_cmd, job_wrapper.get_version_string_path() ) + commands
+        # prepend getting input files (if defined)
+        if hasattr(job_wrapper, 'prepare_input_files_cmds') and job_wrapper.prepare_input_files_cmds is not None:
+            commands = "; ".join( job_wrapper.prepare_input_files_cmds + [ commands ] ) 
         # Prepend dependency injection
         if job_wrapper.dependency_shell_commands:
             commands = "; ".join( job_wrapper.dependency_shell_commands + [ commands ] ) 
-        
+
         # Append metadata setting commands, we don't want to overwrite metadata
         # that was copied over in init_meta(), as per established behavior
         if include_metadata and self.app.config.set_metadata_externally:
diff --git a/lib/galaxy/jobs/runners/lwr.py b/lib/galaxy/jobs/runners/lwr.py
index 56d099b505c..465f92ba6f6 100644
--- a/lib/galaxy/jobs/runners/lwr.py
+++ b/lib/galaxy/jobs/runners/lwr.py
@@ -249,6 +249,12 @@ class LwrJobRunner( BaseJobRunner ):
 
         try:
             job_wrapper.prepare()
+            if hasattr(job_wrapper, 'prepare_input_files_cmds') and job_wrapper.prepare_input_files_cmds is not None:
+                for cmd in job_wrapper.prepare_input_file_cmds: # run the commands to stage the input files
+                    #log.debug( 'executing: %s' % cmd )
+                    if 0 != os.system(cmd):
+                        raise Exception('Error running file staging command: %s' % cmd)
+                job_wrapper.prepare_input_files_cmds = None # prevent them from being used in-line
             command_line = self.build_command_line( job_wrapper, include_metadata=False )
         except:
             job_wrapper.fail( "failure preparing job", exception=True )
diff --git a/lib/galaxy/jobs/splitters/multi.py b/lib/galaxy/jobs/splitters/multi.py
index 71df3058e76..dac31b06264 100644
--- a/lib/galaxy/jobs/splitters/multi.py
+++ b/lib/galaxy/jobs/splitters/multi.py
@@ -1,5 +1,6 @@
 import os, logging,  shutil
-from galaxy import model
+from galaxy import model, util
+
 
 log = logging.getLogger( __name__ )
     
@@ -54,7 +55,7 @@ def do_split (job_wrapper):
         raise Exception(log_error)
     
     # split the first one to build up the task directories
-    input_files = []
+    input_datasets = []
     for input in parent_job.input_datasets:
         if input.name in split_inputs:
             this_input_files = job_wrapper.get_input_dataset_fnames(input.dataset)
@@ -62,13 +63,13 @@ def do_split (job_wrapper):
                 log_error = "The input '%s' is composed of multiple files - splitting is not allowed" % str(input.name)
                 log.error(log_error)
                 raise Exception(log_error)
-            input_files.extend(this_input_files)    
+            input_datasets.append(input.dataset)
     
     input_type = type_to_input_map.keys()[0]
     # DBTODO execute an external task to do the splitting, this should happen at refactor.
     # If the number of tasks is sufficiently high, we can use it to calculate job completion % and give a running status.
     try:
-        input_type.split(input_files, get_new_working_directory_name, parallel_settings)
+        input_type.split(input_datasets, get_new_working_directory_name, parallel_settings)
     except AttributeError:
         log_error = "The type '%s' does not define a method for splitting files" % str(input_type)
         log.error(log_error)
@@ -82,8 +83,9 @@ def do_split (job_wrapper):
                 for file in names:
                     os.symlink(file, os.path.join(dir,  os.path.basename(file)))
     tasks = []
+    prepare_files = os.path.join(util.galaxy_directory(), 'extract_dataset_parts.sh') + ' %s'
     for dir in task_dirs:
-        task = model.Task(parent_job, dir)
+        task = model.Task(parent_job, dir, prepare_files % dir)
         tasks.append(task)
     return tasks
                                
@@ -106,44 +108,51 @@ def do_merge( job_wrapper,  task_wrappers):
     
     illegal_outputs = [x for x in merge_outputs if x in pickone_outputs]
     if len(illegal_outputs) > 0:
-        raise Exception("Outputs have conflicting parallelism attributes: %s" % str( illegal_outputs ))
-
+        return ('Tool file error', 'Outputs have conflicting parallelism attributes: %s' % str( illegal_outputs ))
     
-    working_directory = job_wrapper.working_directory
-    task_dirs = [os.path.join(working_directory, x) for x in os.listdir(working_directory) if x.startswith('task_')]
-    # TODO: Output datasets can be very complex. This doesn't handle metadata files
-    outputs = job_wrapper.get_output_datasets_and_fnames()
-    pickone_done = []
-    task_dirs = [os.path.join(working_directory, x) for x in os.listdir(working_directory) if x.startswith('task_')]
-    for output in outputs:
-        output_file_name = str(outputs[output][1])
-        base_output_name = os.path.basename(output_file_name)
-        if output in merge_outputs:
-            output_type = outputs[output][0].datatype
-            output_files = [os.path.join(dir,base_output_name) for dir in task_dirs]
-            log.debug('files %s ' % output_files)
-            output_type.merge(output_files, output_file_name)
-            log.debug('merge finished: %s' % output_file_name)
-            pass # TODO: merge all the files
-        elif output in pickone_outputs:
-            # just pick one of them
-            if output not in pickone_done:
-                task_file_name = os.path.join(task_dirs[0], base_output_name)
-                shutil.move( task_file_name,  output_file_name )
-                pickone_done.append(output)
-        else:
-            log_error = "The output '%s' does not define a method for implementing parallelism" % output
-            log.error(log_error)
-            raise Exception(log_error)
-
     stdout = ''
-    stderr=''
+    stderr = ''
+    
+    try:
+        working_directory = job_wrapper.working_directory
+        task_dirs = [os.path.join(working_directory, x) for x in os.listdir(working_directory) if x.startswith('task_')]
+        # TODO: Output datasets can be very complex. This doesn't handle metadata files
+        outputs = job_wrapper.get_output_datasets_and_fnames()
+        pickone_done = []
+        task_dirs = [os.path.join(working_directory, x) for x in os.listdir(working_directory) if x.startswith('task_')]
+        for output in outputs:
+            output_file_name = str(outputs[output][1])
+            base_output_name = os.path.basename(output_file_name)
+            if output in merge_outputs:
+                output_type = outputs[output][0].datatype
+                output_files = [os.path.join(dir,base_output_name) for dir in task_dirs]
+                log.debug('files %s ' % output_files)
+                output_type.merge(output_files, output_file_name)
+                log.debug('merge finished: %s' % output_file_name)
+                pass # TODO: merge all the files
+            elif output in pickone_outputs:
+                # just pick one of them
+                if output not in pickone_done:
+                    task_file_name = os.path.join(task_dirs[0], base_output_name)
+                    shutil.move( task_file_name,  output_file_name )
+                    pickone_done.append(output)
+            else:
+                log_error = "The output '%s' does not define a method for implementing parallelism" % output
+                log.error(log_error)
+                raise Exception(log_error)
+    except Exception, e:
+        stdout = 'Error merging files';
+        stderr = str(e)
+        
+
     for tw in task_wrappers:
         # Prevent repetitive output, e.g. "Sequence File Aligned"x20
         # Eventually do a reduce for jobs that output "N reads mapped", combining all N for tasks.
-        if stdout.strip() != tw.get_task().stdout.strip():
-            stdout += tw.get_task().stdout
-        if stderr.strip() != tw.get_task().stderr.strip():                        
-            stderr += tw.get_task().stderr
+        out = tw.get_task().stdout.strip()
+        err = tw.get_task().stderr.strip()
+        if len(out) > 0:
+            stdout += tw.working_directory + ':\n' + out
+        if len(err) > 0:
+            stderr += tw.working_directory + ':\n' + err
     return (stdout,  stderr)
     
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 0b887c5137d..c87a94e3d88 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -204,29 +204,19 @@ class Task( object ):
                     ERROR = 'error',
                     DELETED = 'deleted' )
 
-    def __init__( self, job, part_file = None ):
+    def __init__( self, job, working_directory, prepare_files_cmd ):
         self.command_line = None
         self.parameters = []
         self.state = Task.states.NEW
         self.info = None
-        # TODO: Rename this to working_directory
-        # Does this necessitate a DB migration step?
-        self.part_file = part_file
+        self.working_directory = working_directory
         self.task_runner_name = None
         self.task_runner_external_id = None
         self.job = job
         self.stdout = None
         self.stderr = None
+        self.prepare_input_files_cmd = prepare_files_cmd
     
-    @property
-    def working_directory(self):
-        if self.part_file is not None:
-            if not os.path.isdir(self.part_file):
-                return os.path.dirname(self.part_file)
-            else:
-                return self.part_file
-        return None
-
     def set_state( self, state ):
         self.state = state
 
@@ -907,7 +897,9 @@ class DatasetInstance( object ):
     def get_converted_files_by_type( self, file_type ):
         for assoc in self.implicitly_converted_datasets:
             if not assoc.deleted and assoc.type == file_type:
-                return assoc.dataset
+                if assoc.dataset:
+                    return assoc.dataset
+                return assoc.dataset_ldda
         return None
     def get_converted_dataset_deps(self, trans, target_ext):
         """
@@ -1599,7 +1591,12 @@ class DatasetToValidationErrorAssociation( object ):
 class ImplicitlyConvertedDatasetAssociation( object ):
     def __init__( self, id = None, parent = None, dataset = None, file_type = None, deleted = False, purged = False, metadata_safe = True ):
         self.id = id
-        self.dataset = dataset
+        if isinstance(dataset, HistoryDatasetAssociation):
+            self.dataset = dataset
+        elif isinstance(dataset, LibraryDatasetDatasetAssociation):
+            self.dataset_ldda = dataset
+        else:
+            raise AttributeError, 'Unknown dataset type provided for dataset: %s' % type( dataset )
         if isinstance(parent, HistoryDatasetAssociation):
             self.parent_hda = parent
         elif isinstance(parent, LibraryDatasetDatasetAssociation):
diff --git a/lib/galaxy/model/mapping.py b/lib/galaxy/model/mapping.py
index f013186a978..6187af2e9ed 100644
--- a/lib/galaxy/model/mapping.py
+++ b/lib/galaxy/model/mapping.py
@@ -148,6 +148,7 @@ ImplicitlyConvertedDatasetAssociation.table = Table( "implicitly_converted_datas
     Column( "create_time", DateTime, default=now ),
     Column( "update_time", DateTime, default=now, onupdate=now ),
     Column( "hda_id", Integer, ForeignKey( "history_dataset_association.id" ), index=True, nullable=True ),
+    Column( "ldda_id", Integer, ForeignKey( "library_dataset_dataset_association.id" ), index=True, nullable=True ),
     Column( "hda_parent_id", Integer, ForeignKey( "history_dataset_association.id" ), index=True ),
     Column( "ldda_parent_id", Integer, ForeignKey( "library_dataset_dataset_association.id" ), index=True ),
     Column( "deleted", Boolean, index=True, default=False ),
@@ -469,9 +470,10 @@ Task.table = Table( "task", metadata,
     Column( "stderr", TEXT ),
     Column( "traceback", TEXT ),
     Column( "job_id", Integer, ForeignKey( "job.id" ), index=True, nullable=False ),
-    Column( "part_file", String(1024)),
+    Column( "working_directory", String(1024)),
     Column( "task_runner_name", String( 255 ) ),
-    Column( "task_runner_external_id", String( 255 ) ) )
+    Column( "task_runner_external_id", String( 255 ) ),
+    Column( "prepare_input_files_cmd", TEXT ) )
 
 PostJobAction.table = Table("post_job_action", metadata,
     Column("id", Integer, primary_key=True),
@@ -1211,6 +1213,9 @@ assign_mapper( context, ImplicitlyConvertedDatasetAssociation, ImplicitlyConvert
                         LibraryDatasetDatasetAssociation, 
                         primaryjoin=( ImplicitlyConvertedDatasetAssociation.table.c.ldda_parent_id == LibraryDatasetDatasetAssociation.table.c.id ) ),
                      
+                     dataset_ldda=relation( 
+                        LibraryDatasetDatasetAssociation, 
+                        primaryjoin=( ImplicitlyConvertedDatasetAssociation.table.c.ldda_id == LibraryDatasetDatasetAssociation.table.c.id ) ),
                      dataset=relation( 
                         HistoryDatasetAssociation, 
                         primaryjoin=( ImplicitlyConvertedDatasetAssociation.table.c.hda_id == HistoryDatasetAssociation.table.c.id ) ) ) )
@@ -1594,7 +1599,7 @@ assign_mapper( context, Page, Page.table,
                      annotations=relation( PageAnnotationAssociation, order_by=PageAnnotationAssociation.table.c.id, backref="pages" ),
                      ratings=relation( PageRatingAssociation, order_by=PageRatingAssociation.table.c.id, backref="pages" )  
                    ) )
-
+                   
 assign_mapper( context, ToolShedRepository, ToolShedRepository.table )
 
 # Set up proxy so that 
diff --git a/lib/galaxy/model/migrate/versions/0083_add_prepare_files_to_task.py b/lib/galaxy/model/migrate/versions/0083_add_prepare_files_to_task.py
new file mode 100644
index 00000000000..b2431c85710
--- /dev/null
+++ b/lib/galaxy/model/migrate/versions/0083_add_prepare_files_to_task.py
@@ -0,0 +1,63 @@
+"""
+Migration script to add 'prepare_input_files_cmd' column to the task table and to rename a column.
+"""
+
+from sqlalchemy import *
+from sqlalchemy.orm import *
+from migrate import *
+from migrate.changeset import *
+
+import logging
+log = logging.getLogger( __name__ )
+
+metadata = MetaData( migrate_engine )
+db_session = scoped_session( sessionmaker( bind=migrate_engine, autoflush=False, autocommit=True ) )
+
+def upgrade():
+    print __doc__
+    metadata.reflect()
+    try:
+        task_table = Table( "task", metadata, autoload=True )
+        c = Column( "prepare_input_files_cmd", TEXT, nullable=True )
+        c.create( task_table )
+        assert c is task_table.c.prepare_input_files_cmd
+    except Exception, e:
+        print "Adding prepare_input_files_cmd column to task table failed: %s" % str( e )
+        log.debug( "Adding prepare_input_files_cmd column to task table failed: %s" % str( e ) )
+    try:
+        task_table = Table( "task", metadata, autoload=True )
+        c = Column( "working_directory", String ( 1024 ), nullable=True )
+        c.create( task_table )
+        assert c is task_table.c.working_directory
+    except Exception, e:
+        print "Adding working_directory column to task table failed: %s" % str( e )
+        log.debug( "Adding working_directory column to task table failed: %s" % str( e ) )
+
+    # remove the 'part_file' column - nobody used tasks before this, so no data needs to be migrated
+    try:
+        task_table.c.part_file.drop()
+    except Exception, e:
+        log.debug( "Deleting column 'part_file' from the 'task' table failed: %s" % ( str( e ) ) )
+
+def downgrade():
+    metadata.reflect()
+    try:
+        task_table = Table( "task", metadata, autoload=True )
+        task_table.c.prepare_input_files_cmd.drop()
+    except Exception, e:
+        print "Dropping prepare_input_files_cmd column from task table failed: %s" % str( e )
+        log.debug( "Dropping prepare_input_files_cmd column from task table failed: %s" % str( e ) )
+    try:
+        task_table = Table( "task", metadata, autoload=True )
+        task_table.c.working_directory.drop()
+    except Exception, e:
+        print "Dropping working_directory column from task table failed: %s" % str( e )
+        log.debug( "Dropping working_directory column from task table failed: %s" % str( e ) )
+    try:
+        task_table = Table( "task", metadata, autoload=True )
+        c = Column( "part_file", String ( 1024 ), nullable=True )
+        c.create( task_table )
+        assert c is task_table.c.part_file
+    except Exception, e:
+        print "Adding part_file column to task table failed: %s" % str( e )
+        log.debug( "Adding part_file column to task table failed: %s" % str( e ) )
diff --git a/lib/galaxy/util/__init__.py b/lib/galaxy/util/__init__.py
index cad925ec778..76655450295 100644
--- a/lib/galaxy/util/__init__.py
+++ b/lib/galaxy/util/__init__.py
@@ -623,6 +623,9 @@ ucsc_build_sites = read_build_sites( os.path.join( galaxy_root_path, "tool-data"
 gbrowse_build_sites = read_build_sites( os.path.join( galaxy_root_path, "tool-data", "shared", "gbrowse", "gbrowse_build_sites.txt" ) )
 genetrack_sites = read_build_sites( os.path.join( galaxy_root_path, "tool-data", "shared", "genetrack", "genetrack_sites.txt" ), check_builds=False )
 
+def galaxy_directory():
+    return os.path.abspath(galaxy_root_path)
+
 if __name__ == '__main__':
     import doctest, sys
     doctest.testmod(sys.modules[__name__], verbose=False)
diff --git a/scripts/extract_dataset_part.py b/scripts/extract_dataset_part.py
new file mode 100644
index 00000000000..99fb3583595
--- /dev/null
+++ b/scripts/extract_dataset_part.py
@@ -0,0 +1,48 @@
+"""
+Reads a JSON file and uses it to call into a datatype class to extract
+a subset of a dataset for processing.
+
+Used by jobs that split large files into pieces to be processed concurrently
+on a gid in a scatter-gather mode. This does part of the scatter.
+
+"""
+import os
+import sys
+import logging
+logging.basicConfig()
+log = logging.getLogger( __name__ )
+
+new_path = [ os.path.join( os.getcwd(), "lib" ) ]
+new_path.extend( sys.path[1:] ) # remove scripts/ from the path
+sys.path = new_path
+
+from galaxy import eggs
+import pkg_resources
+pkg_resources.require("simplejson")
+import simplejson
+
+# This junk is here to prevent loading errors
+import galaxy.model.mapping #need to load this before we unpickle, in order to setup properties assigned by the mappers
+galaxy.model.Job() #this looks REAL stupid, but it is REQUIRED in order for SA to insert parameters into the classes defined by the mappers --> it appears that instantiating ANY mapper'ed class would suffice here
+galaxy.datatypes.metadata.DATABASE_CONNECTION_AVAILABLE = False #Let metadata know that there is no database connection, and to just assume object ids are valid
+
+def __main__():
+    """
+    Argument: a JSON file
+    """
+    file_path = sys.argv.pop( 1 )
+    data = simplejson.load(open(file_path, 'r'))
+    try:
+        class_name_parts = data['class_name'].split('.')
+        module_name = '.'.join(class_name_parts[:-1])
+        class_name = class_name_parts[-1]
+        mod = __import__(module_name, globals(),  locals(),  [class_name])
+        cls = getattr(mod,  class_name)
+        if not cls.process_split_file(data):
+            sys.stderr.write('Writing split file failed\n')
+            sys.exit(1)
+    except Exception, e:
+        sys.stderr.write(str(e))
+        sys.exit(1)
+        
+__main__()