pintos-os.org Git - pspp/blob - src/math/merge.c

   1 /* PSPP - a program for statistical analysis.
   2    Copyright (C) 2007, 2009-11, 14 Free Software Foundation, Inc.
   3
   4    This program is free software: you can redistribute it and/or modify
   5    it under the terms of the GNU General Public License as published by
   6    the Free Software Foundation, either version 3 of the License, or
   7    (at your option) any later version.
   8
   9    This program is distributed in the hope that it will be useful,
  10    but WITHOUT ANY WARRANTY; without even the implied warranty of
  11    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12    GNU General Public License for more details.
  13
  14    You should have received a copy of the GNU General Public License
  15    along with this program.  If not, see <http://www.gnu.org/licenses/>. */
  16
  17 /* FIXME: error checking. */
  18 /* FIXME: merge pattern should be improved, this one causes a
  19    performance regression. */
  20 #include <config.h>
  21
  22 #include "math/merge.h"
  23
  24 #include "data/case.h"
  25 #include "data/casereader.h"
  26 #include "data/casewriter.h"
  27 #include "data/subcase.h"
  28 #include "libpspp/array.h"
  29 #include "libpspp/assertion.h"
  30 #include "libpspp/taint.h"
  31
  32 #include "gl/xalloc.h"
  33
  34 #define MAX_MERGE_ORDER 7
  35
  36 struct merge_input
  37   {
  38     struct casereader *reader;
  39     struct ccase *c;
  40   };
  41
  42 struct merge
  43   {
  44     struct subcase ordering;
  45     struct merge_input inputs[MAX_MERGE_ORDER];
  46     size_t input_cnt;
  47     struct caseproto *proto;
  48
  49     merge_distinct_combine_func *combine;
  50     void *aux;
  51   };
  52
  53 static void do_merge (struct merge *m);
  54
  55 struct merge *
  56 merge_create (const struct subcase *ordering, const struct caseproto *proto,
  57               merge_distinct_combine_func *combine, void *aux)
  58 {
  59   struct merge *m = xmalloc (sizeof *m);
  60   subcase_clone (&m->ordering, ordering);
  61   m->input_cnt = 0;
  62   m->proto = caseproto_ref (proto);
  63   m->combine = combine;
  64   m->aux = aux;
  65   return m;
  66 }
  67
  68 void
  69 merge_destroy (struct merge *m)
  70 {
  71   if (m != NULL)
  72     {
  73       size_t i;
  74
  75       subcase_destroy (&m->ordering);
  76       for (i = 0; i < m->input_cnt; i++)
  77         casereader_destroy (m->inputs[i].reader);
  78       caseproto_unref (m->proto);
  79       free (m);
  80     }
  81 }
  82
  83 void
  84 merge_append (struct merge *m, struct casereader *r)
  85 {
  86   r = casereader_rename (r);
  87   m->inputs[m->input_cnt++].reader = r;
  88   if (m->input_cnt >= MAX_MERGE_ORDER)
  89     do_merge (m);
  90 }
  91
  92 struct casereader *
  93 merge_make_reader (struct merge *m)
  94 {
  95   struct casereader *r;
  96
  97   if (m->input_cnt > 1)
  98     do_merge (m);
  99
 100   if (m->input_cnt == 1)
 101     {
 102       r = m->inputs[0].reader;
 103       m->input_cnt = 0;
 104     }
 105   else if (m->input_cnt == 0)
 106     {
 107       struct casewriter *writer = mem_writer_create (m->proto);
 108       r = casewriter_make_reader (writer);
 109     }
 110   else
 111     NOT_REACHED ();
 112
 113   return r;
 114 }
 115
 116 static bool
 117 read_input_case (struct merge *m, size_t idx)
 118 {
 119   struct merge_input *i = &m->inputs[idx];
 120
 121   i->c = casereader_read (i->reader);
 122   if (i->c)
 123     return true;
 124   else
 125     {
 126       casereader_destroy (i->reader);
 127       remove_element (m->inputs, m->input_cnt, sizeof *m->inputs, idx);
 128       m->input_cnt--;
 129       return false;
 130     }
 131 }
 132
 133 static void
 134 do_merge (struct merge *m)
 135 {
 136   struct casewriter *w;
 137   struct ccase *prev_case;
 138   size_t i;
 139
 140   assert (m->input_cnt > 1);
 141
 142   w = tmpfile_writer_create (m->proto);
 143   for (i = 0; i < m->input_cnt; i++)
 144     taint_propagate (casereader_get_taint (m->inputs[i].reader),
 145                      casewriter_get_taint (w));
 146
 147   for (i = 0; i < m->input_cnt; )
 148     if (read_input_case (m, i))
 149       i++;
 150
 151   prev_case = NULL;
 152   while (m->input_cnt > 0)
 153     {
 154       struct ccase *min_case;
 155       size_t min;
 156
 157       min = 0;
 158       for (i = 1; i < m->input_cnt; i++)
 159         if (subcase_compare_3way (&m->ordering, m->inputs[i].c,
 160                                   &m->ordering, m->inputs[min].c) < 0)
 161           min = i;
 162
 163       min_case = m->inputs[min].c;
 164       if (m->combine != NULL)
 165         {
 166           if (prev_case == NULL)
 167             prev_case = min_case;
 168           else if (subcase_equal (&m->ordering, min_case,
 169                                   &m->ordering, prev_case))
 170             prev_case = m->combine (prev_case, min_case, m->aux);
 171           else
 172             {
 173               casewriter_write (w, prev_case);
 174               prev_case = min_case;
 175             }
 176         }
 177       else
 178         casewriter_write (w, min_case);
 179
 180       read_input_case (m, min);
 181     }
 182   if (prev_case != NULL)
 183     casewriter_write (w, prev_case);
 184
 185   m->input_cnt = 1;
 186   m->inputs[0].reader = casewriter_make_reader (w);
 187 }