1 /* PSPP - a program for statistical analysis.
2 Copyright (C) 2007, 2009-11, 14 Free Software Foundation, Inc.
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.
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.
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/>. */
17 /* FIXME: error checking. */
18 /* FIXME: merge pattern should be improved, this one causes a
19 performance regression. */
22 #include "math/merge.h"
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"
32 #include "gl/xalloc.h"
34 #define MAX_MERGE_ORDER 7
38 struct casereader *reader;
44 struct subcase ordering;
45 struct merge_input inputs[MAX_MERGE_ORDER];
47 struct caseproto *proto;
49 merge_distinct_combine_func *combine;
53 static void do_merge (struct merge *m);
56 merge_create (const struct subcase *ordering, const struct caseproto *proto,
57 merge_distinct_combine_func *combine, void *aux)
59 struct merge *m = xmalloc (sizeof *m);
60 subcase_clone (&m->ordering, ordering);
62 m->proto = caseproto_ref (proto);
69 merge_destroy (struct merge *m)
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);
84 merge_append (struct merge *m, struct casereader *r)
86 r = casereader_rename (r);
87 m->inputs[m->input_cnt++].reader = r;
88 if (m->input_cnt >= MAX_MERGE_ORDER)
93 merge_make_reader (struct merge *m)
100 if (m->input_cnt == 1)
102 r = m->inputs[0].reader;
105 else if (m->input_cnt == 0)
107 struct casewriter *writer = mem_writer_create (m->proto);
108 r = casewriter_make_reader (writer);
117 read_input_case (struct merge *m, size_t idx)
119 struct merge_input *i = &m->inputs[idx];
121 i->c = casereader_read (i->reader);
126 casereader_destroy (i->reader);
127 remove_element (m->inputs, m->input_cnt, sizeof *m->inputs, idx);
134 do_merge (struct merge *m)
136 struct casewriter *w;
137 struct ccase *prev_case;
140 assert (m->input_cnt > 1);
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));
147 for (i = 0; i < m->input_cnt; )
148 if (read_input_case (m, i))
152 while (m->input_cnt > 0)
154 struct ccase *min_case;
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)
163 min_case = m->inputs[min].c;
164 if (m->combine != NULL)
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);
173 casewriter_write (w, prev_case);
174 prev_case = min_case;
178 casewriter_write (w, min_case);
180 read_input_case (m, min);
182 if (prev_case != NULL)
183 casewriter_write (w, prev_case);
186 m->inputs[0].reader = casewriter_make_reader (w);