nodes: 0, module; 1, function_definition; 2, function_name:_merge_sorted_items; 3, parameters; 4, block; 5, identifier:self; 6, identifier:index; 7, expression_statement; 8, function_definition; 9, expression_statement; 10, if_statement; 11, return_statement; 12, comment:""" load a partition from disk, then sort and group by key """; 13, function_name:load_partition; 14, parameters; 15, block; 16, assignment; 17, attribute; 18, comment:# all the partitions are already sorted; 19, block; 20, else_clause; 21, generator_expression; 22, identifier:j; 23, expression_statement; 24, expression_statement; 25, with_statement; 26, identifier:disk_items; 27, list_comprehension; 28, identifier:self; 29, identifier:_sorted; 30, expression_statement; 31, comment:# Flatten the combined values, so it will not consume huge; 32, comment:# memory during merging sort.; 33, block; 34, tuple; 35, for_in_clause; 36, assignment; 37, assignment; 38, with_clause; 39, block; 40, call; 41, for_in_clause; 42, assignment; 43, expression_statement; 44, expression_statement; 45, expression_statement; 46, identifier:k; 47, identifier:vs; 48, pattern_list; 49, call; 50, identifier:path; 51, call; 52, identifier:p; 53, call; 54, with_item; 55, for_statement; 56, identifier:load_partition; 57, argument_list; 58, identifier:j; 59, call; 60, identifier:sorted_items; 61, call; 62, assignment; 63, assignment; 64, assignment; 65, identifier:k; 66, identifier:vs; 67, identifier:GroupByKey; 68, argument_list; 69, attribute; 70, argument_list; 71, attribute; 72, argument_list; 73, as_pattern; 74, identifier:v; 75, call; 76, block; 77, identifier:j; 78, identifier:range; 79, argument_list; 80, attribute; 81, argument_list; 82, identifier:ser; 83, call; 84, identifier:sorter; 85, call; 86, identifier:sorted_items; 87, call; 88, identifier:sorted_items; 89, identifier:self; 90, identifier:_get_spill_dir; 91, identifier:j; 92, attribute; 93, identifier:join; 94, identifier:path; 95, call; 96, call; 97, as_pattern_target; 98, attribute; 99, argument_list; 100, expression_statement; 101, attribute; 102, identifier:heapq; 103, identifier:merge; 104, identifier:disk_items; 105, keyword_argument; 106, attribute; 107, argument_list; 108, identifier:ExternalSorter; 109, argument_list; 110, attribute; 111, argument_list; 112, identifier:os; 113, identifier:path; 114, identifier:str; 115, argument_list; 116, identifier:open; 117, argument_list; 118, identifier:f; 119, attribute; 120, identifier:load_stream; 121, identifier:f; 122, yield; 123, identifier:self; 124, identifier:spills; 125, identifier:key; 126, call; 127, identifier:self; 128, identifier:flattened_serializer; 129, attribute; 130, identifier:ser; 131, identifier:sorter; 132, identifier:sorted; 133, call; 134, keyword_argument; 135, identifier:index; 136, identifier:p; 137, string; 138, integer:65536; 139, identifier:self; 140, identifier:serializer; 141, identifier:v; 142, attribute; 143, argument_list; 144, identifier:self; 145, identifier:memory_limit; 146, attribute; 147, argument_list; 148, identifier:key; 149, call; 150, string_content:rb; 151, identifier:operator; 152, identifier:itemgetter; 153, integer:0; 154, identifier:itertools; 155, identifier:chain; 156, list_splat; 157, attribute; 158, argument_list; 159, identifier:disk_items; 160, identifier:operator; 161, identifier:itemgetter; 162, integer:0
code: def _merge_sorted_items(self, index):
          """ load a partition from disk, then sort and group by key """
          def load_partition(j):
              path = self._get_spill_dir(j)
              p = os.path.join(path, str(index))
              with open(p, 'rb', 65536) as f:
                  for v in self.serializer.load_stream(f):
                      yield v
  
          disk_items = [load_partition(j) for j in range(self.spills)]
  
          if self._sorted:
              # all the partitions are already sorted
              sorted_items = heapq.merge(disk_items, key=operator.itemgetter(0))
  
          else:
              # Flatten the combined values, so it will not consume huge
              # memory during merging sort.
              ser = self.flattened_serializer()
              sorter = ExternalSorter(self.memory_limit, ser)
              sorted_items = sorter.sorted(itertools.chain(*disk_items),
                                           key=operator.itemgetter(0))
          return ((k, vs) for k, vs in GroupByKey(sorted_items))