src/data/casegrouper.c

   1 /* PSPP - a program for statistical analysis.
   2    Copyright (C) 2007 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 #include <config.h>
  18
  19 #include <data/casegrouper.h>
  20
  21 #include <stdlib.h>
  22
  23 #include <data/case-ordering.h>
  24 #include <data/casereader.h>
  25 #include <data/casewriter.h>
  26 #include <data/dictionary.h>
  27 #include <libpspp/taint.h>
  28
  29 #include "xalloc.h"
  30
  31 /* A casegrouper. */
  32 struct casegrouper
  33   {
  34     struct casereader *reader;  /* Source of input cases. */
  35     struct taint *taint;        /* Error status for casegrouper. */
  36
  37     /* Functions for grouping cases. */
  38     bool (*same_group) (const struct ccase *, const struct ccase *, void *aux);
  39     void (*destroy) (void *aux);
  40     void *aux;
  41   };
  42
  43 /* Creates and returns a new casegrouper that takes its input
  44    from READER.  SAME_GROUP is used to decide which cases are in
  45    a group: it returns true if the pair of cases provided are in
  46    the same group, false otherwise.  DESTROY will be called when
  47    the casegrouper is destroyed and should free any storage
  48    needed by SAME_GROUP.
  49
  50    SAME_GROUP may be a null pointer.  If so, READER's entire
  51    contents is considered to be a single group. */
  52 struct casegrouper *
  53 casegrouper_create_func (struct casereader *reader,
  54                          bool (*same_group) (const struct ccase *,
  55                                              const struct ccase *,
  56                                              void *aux),
  57                          void (*destroy) (void *aux),
  58                          void *aux)
  59 {
  60   struct casegrouper *grouper = xmalloc (sizeof *grouper);
  61   grouper->reader = casereader_rename (reader);
  62   grouper->taint = taint_clone (casereader_get_taint (grouper->reader));
  63   grouper->same_group = same_group;
  64   grouper->destroy = destroy;
  65   grouper->aux = aux;
  66   return grouper;
  67 }
  68
  69 /* Obtains the next group of cases from GROUPER.  Returns true if
  70    successful, false if no groups remain.  If successful, *READER
  71    is set to the casereader for the new group; otherwise, it is
  72    set to NULL. */
  73 bool
  74 casegrouper_get_next_group (struct casegrouper *grouper,
  75                             struct casereader **reader)
  76 {
  77   /* FIXME: we really shouldn't need a temporary casewriter for
  78      the common case where we read an entire group's data before
  79      going on to the next. */
  80   if (grouper->same_group != NULL)
  81     {
  82       struct casewriter *writer;
  83       struct ccase group_case, tmp;
  84
  85       if (!casereader_read (grouper->reader, &group_case))
  86         {
  87           *reader = NULL;
  88           return false;
  89         }
  90
  91       writer = autopaging_writer_create (casereader_get_value_cnt (grouper->reader));
  92       case_clone (&tmp, &group_case);
  93       casewriter_write (writer, &tmp);
  94
  95       while (casereader_peek (grouper->reader, 0, &tmp)
  96              && grouper->same_group (&group_case, &tmp, grouper->aux))
  97         {
  98           struct ccase discard;
  99           casereader_read (grouper->reader, &discard);
 100           case_destroy (&discard);
 101           casewriter_write (writer, &tmp);
 102         }
 103       case_destroy (&tmp);
 104       case_destroy (&group_case);
 105
 106       *reader = casewriter_make_reader (writer);
 107       return true;
 108     }
 109   else
 110     {
 111       if (grouper->reader != NULL)
 112         {
 113           if (!casereader_is_empty (grouper->reader))
 114             {
 115               *reader = grouper->reader;
 116               grouper->reader = NULL;
 117               return true;
 118             }
 119           else
 120             {
 121               casereader_destroy (grouper->reader);
 122               grouper->reader = NULL;
 123               return false;
 124             }
 125         }
 126       else
 127         {
 128           *reader = NULL;
 129           return false;
 130         }
 131     }
 132 }
 133
 134 /* Destroys GROUPER.  Returns false if GROUPER's input casereader
 135    or any state derived from it had become tainted, which means
 136    that an I/O error or other serious error occurred in
 137    processing data derived from GROUPER; otherwise, return true. */
 138 bool
 139 casegrouper_destroy (struct casegrouper *grouper)
 140 {
 141   if (grouper != NULL)
 142     {
 143       struct taint *taint = grouper->taint;
 144       bool ok;
 145
 146       casereader_destroy (grouper->reader);
 147       if (grouper->destroy != NULL)
 148         grouper->destroy (grouper->aux);
 149       free (grouper);
 150
 151       ok = !taint_has_tainted_successor (taint);
 152       taint_destroy (taint);
 153       return ok;
 154     }
 155   else
 156     return true;
 157 }
 158 \f
 159 /* Casegrouper based on equal values of variables from case to
 160    case. */
 161
 162 /* Casegrouper based on equal variables. */
 163 struct casegrouper_vars
 164   {
 165     const struct variable **vars; /* Variables to compare. */
 166     size_t var_cnt;               /* Number of variables. */
 167   };
 168
 169 static bool casegrouper_vars_same_group (const struct ccase *,
 170                                          const struct ccase *,
 171                                          void *);
 172 static void casegrouper_vars_destroy (void *);
 173
 174 /* Creates and returns a casegrouper that reads data from READER
 175    and breaks it into contiguous groups of cases that have equal
 176    values for the VAR_CNT variables in VARS.  If VAR_CNT is 0,
 177    then all the cases will be put in a single group. */
 178 struct casegrouper *
 179 casegrouper_create_vars (struct casereader *reader,
 180                          const struct variable *const *vars,
 181                          size_t var_cnt)
 182 {
 183   if (var_cnt > 0)
 184     {
 185       struct casegrouper_vars *cv = xmalloc (sizeof *cv);
 186       cv->vars = xmemdup (vars, sizeof *vars * var_cnt);
 187       cv->var_cnt = var_cnt;
 188       return casegrouper_create_func (reader,
 189                                       casegrouper_vars_same_group,
 190                                       casegrouper_vars_destroy,
 191                                       cv);
 192     }
 193   else
 194     return casegrouper_create_func (reader, NULL, NULL, NULL);
 195 }
 196
 197 /* Creates and returns a casegrouper that reads data from READER
 198    and breaks it into contiguous groups of cases that have equal
 199    values for the SPLIT FILE variables in DICT.  If DICT has no
 200    SPLIT FILE variables, then all the cases will be put into a
 201    single group. */
 202 struct casegrouper *
 203 casegrouper_create_splits (struct casereader *reader,
 204                            const struct dictionary *dict)
 205 {
 206   return casegrouper_create_vars (reader,
 207                                   dict_get_split_vars (dict),
 208                                   dict_get_split_cnt (dict));
 209 }
 210
 211 /* Creates and returns a casegrouper that reads data from READER
 212    and breaks it into contiguous groups of cases that have equal
 213    values for the variables used for sorting in CO.  If CO is
 214    empty (contains no sort keys), then all the cases will be put
 215    into a single group. */
 216 struct casegrouper *
 217 casegrouper_create_case_ordering (struct casereader *reader,
 218                                   const struct case_ordering *co)
 219 {
 220   const struct variable **vars;
 221   size_t var_cnt;
 222   struct casegrouper *grouper;
 223
 224   case_ordering_get_vars (co, &vars, &var_cnt);
 225   grouper = casegrouper_create_vars (reader, vars, var_cnt);
 226   free (vars);
 227
 228   return grouper;
 229 }
 230
 231 /* "same_group" function for an equal-variables casegrouper. */
 232 static bool
 233 casegrouper_vars_same_group (const struct ccase *a, const struct ccase *b,
 234                              void *cv_)
 235 {
 236   struct casegrouper_vars *cv = cv_;
 237   return case_compare (a, b, cv->vars, cv->var_cnt) == 0;
 238 }
 239
 240 /* "destroy" for an equal-variables casegrouper. */
 241 static void
 242 casegrouper_vars_destroy (void *cv_)
 243 {
 244   struct casegrouper_vars *cv = cv_;
 245   free (cv->vars);
 246   free (cv);
 247 }
 248