X-Git-Url: https://pintos-os.org/cgi-bin/gitweb.cgi?a=blobdiff_plain;f=src%2Flanguage%2Fstats%2Faggregate.c;h=c9d68d35da70867fccfaf92f0b7f5a1b08322f86;hb=9f4661992f4b481c6dafa6fd53c94ecfe7b3af8c;hp=4c85592bc58973baf0ef7ddd6aaf19e8d350455c;hpb=72147c12357446b304264b6944b8a1d92c45f53c;p=pspp-builds.git

diff --git a/src/language/stats/aggregate.c b/src/language/stats/aggregate.c
index 4c85592b..c9d68d35 100644
--- a/src/language/stats/aggregate.c
+++ b/src/language/stats/aggregate.c
@@ -1,35 +1,34 @@
-/* PSPP - computes sample statistics.
+/* PSPP - a program for statistical analysis.
    Copyright (C) 1997-9, 2000, 2006 Free Software Foundation, Inc.
 
-   This program is free software; you can redistribute it and/or
-   modify it under the terms of the GNU General Public License as
-   published by the Free Software Foundation; either version 2 of the
-   License, or (at your option) any later version.
+   This program is free software: you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation, either version 3 of the License, or
+   (at your option) any later version.
 
-   This program is distributed in the hope that it will be useful, but
-   WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
-   General Public License for more details.
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
 
    You should have received a copy of the GNU General Public License
-   along with this program; if not, write to the Free Software
-   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
-   02110-1301, USA. */
+   along with this program.  If not, see <http://www.gnu.org/licenses/>. */
 
 #include <config.h>
 
 #include <stdlib.h>
 
 #include <data/any-writer.h>
-#include <data/case-sink.h>
+#include <data/case-ordering.h>
 #include <data/case.h>
-#include <data/casefile.h>
+#include <data/casegrouper.h>
+#include <data/casereader.h>
+#include <data/casewriter.h>
 #include <data/dictionary.h>
 #include <data/file-handle-def.h>
 #include <data/format.h>
 #include <data/procedure.h>
 #include <data/settings.h>
-#include <data/storage-stream.h>
 #include <data/sys-file-writer.h>
 #include <data/variable.h>
 #include <language/command.h>
@@ -37,7 +36,6 @@
 #include <language/lexer/lexer.h>
 #include <language/lexer/variable-parser.h>
 #include <language/stats/sort-criteria.h>
-#include <libpspp/alloc.h>
 #include <libpspp/assertion.h>
 #include <libpspp/message.h>
 #include <libpspp/misc.h>
@@ -47,6 +45,7 @@
 #include <math/sort.h>
 
 #include "minmax.h"
+#include "xalloc.h"
 
 #include "gettext.h"
 #define _(msgid) gettext (msgid)
@@ -64,17 +63,17 @@ struct agr_var
     struct agr_var *next;		/* Next in list. */
 
     /* Collected during parsing. */
-    struct variable *src;	/* Source variable. */
+    const struct variable *src;	/* Source variable. */
     struct variable *dest;	/* Target variable. */
     int function;		/* Function. */
-    int include_missing;	/* 1=Include user-missing values. */
+    enum mv_class exclude;      /* Classes of missing values to exclude. */
     union agr_argument arg[2];	/* Arguments. */
 
     /* Accumulated during AGGREGATE execution. */
     double dbl[3];
     int int1, int2;
     char *string;
-    int missing;
+    bool saw_missing;
     struct moments1 *moments;
   };
 
@@ -98,27 +97,27 @@ struct agr_func
   };
 
 /* Attributes of aggregation functions. */
-static const struct agr_func agr_func_tab[] = 
+static const struct agr_func agr_func_tab[] =
   {
     {"<NONE>",  0, -1,          {0, 0, 0}},
     {"SUM",     0, -1,          {FMT_F, 8, 2}},
     {"MEAN",	0, -1,          {FMT_F, 8, 2}},
     {"SD",      0, -1,          {FMT_F, 8, 2}},
-    {"MAX",     0, VAR_STRING,  {-1, -1, -1}}, 
-    {"MIN",     0, VAR_STRING,  {-1, -1, -1}}, 
-    {"PGT",     1, VAR_NUMERIC, {FMT_F, 5, 1}},      
-    {"PLT",     1, VAR_NUMERIC, {FMT_F, 5, 1}},       
-    {"PIN",     2, VAR_NUMERIC, {FMT_F, 5, 1}},       
-    {"POUT",    2, VAR_NUMERIC, {FMT_F, 5, 1}},       
-    {"FGT",     1, VAR_NUMERIC, {FMT_F, 5, 3}},       
-    {"FLT",     1, VAR_NUMERIC, {FMT_F, 5, 3}},       
-    {"FIN",     2, VAR_NUMERIC, {FMT_F, 5, 3}},       
-    {"FOUT",    2, VAR_NUMERIC, {FMT_F, 5, 3}},       
-    {"N",       0, VAR_NUMERIC, {FMT_F, 7, 0}},       
-    {"NU",      0, VAR_NUMERIC, {FMT_F, 7, 0}},       
-    {"NMISS",   0, VAR_NUMERIC, {FMT_F, 7, 0}},       
-    {"NUMISS",  0, VAR_NUMERIC, {FMT_F, 7, 0}},       
-    {"FIRST",   0, VAR_STRING,  {-1, -1, -1}}, 
+    {"MAX",     0, VAR_STRING,  {-1, -1, -1}},
+    {"MIN",     0, VAR_STRING,  {-1, -1, -1}},
+    {"PGT",     1, VAR_NUMERIC, {FMT_F, 5, 1}},
+    {"PLT",     1, VAR_NUMERIC, {FMT_F, 5, 1}},
+    {"PIN",     2, VAR_NUMERIC, {FMT_F, 5, 1}},
+    {"POUT",    2, VAR_NUMERIC, {FMT_F, 5, 1}},
+    {"FGT",     1, VAR_NUMERIC, {FMT_F, 5, 3}},
+    {"FLT",     1, VAR_NUMERIC, {FMT_F, 5, 3}},
+    {"FIN",     2, VAR_NUMERIC, {FMT_F, 5, 3}},
+    {"FOUT",    2, VAR_NUMERIC, {FMT_F, 5, 3}},
+    {"N",       0, VAR_NUMERIC, {FMT_F, 7, 0}},
+    {"NU",      0, VAR_NUMERIC, {FMT_F, 7, 0}},
+    {"NMISS",   0, VAR_NUMERIC, {FMT_F, 7, 0}},
+    {"NUMISS",  0, VAR_NUMERIC, {FMT_F, 7, 0}},
+    {"FIRST",   0, VAR_STRING,  {-1, -1, -1}},
     {"LAST",    0, VAR_STRING,  {-1, -1, -1}},
     {NULL,      0, -1,          {-1, -1, -1}},
     {"N",       0, VAR_NUMERIC, {FMT_F, 7, 0}},
@@ -133,15 +132,11 @@ enum missing_treatment
   };
 
 /* An entire AGGREGATE procedure. */
-struct agr_proc 
+struct agr_proc
   {
-    /* We have either an output file or a sink. */
-    struct any_writer *writer;          /* Output file, or null if none. */
-    struct case_sink *sink;             /* Sink, or null if none. */
-
     /* Break variables. */
-    struct sort_criteria *sort;         /* Sort criteria. */
-    struct variable **break_vars;       /* Break variables. */
+    struct case_ordering *sort;         /* Sort criteria. */
+    const struct variable **break_vars;       /* Break variables. */
     size_t break_var_cnt;               /* Number of break variables. */
     struct ccase break_case;            /* Last values of break variables. */
 
@@ -150,26 +145,18 @@ struct agr_proc
     struct dictionary *dict;            /* Aggregate dictionary. */
     const struct dictionary *src_dict;  /* Dict of the source */
     int case_cnt;                       /* Counts aggregated cases. */
-    struct ccase agr_case;              /* Aggregate case for output. */
   };
 
 static void initialize_aggregate_info (struct agr_proc *,
                                        const struct ccase *);
-
+static void accumulate_aggregate_info (struct agr_proc *,
+                                       const struct ccase *);
 /* Prototypes. */
 static bool parse_aggregate_functions (struct lexer *, const struct dictionary *,
 				       struct agr_proc *);
 static void agr_destroy (struct agr_proc *);
-static bool aggregate_single_case (struct agr_proc *agr,
-				   const struct ccase *input,
-				   struct ccase *output);
-static void dump_aggregate_info (struct agr_proc *agr, struct ccase *output);
-
-/* Aggregating to the active file. */
-static bool agr_to_active_file (const struct ccase *, void *aux, const struct dataset *);
-
-/* Aggregating to a system file. */
-static bool presorted_agr_to_sysfile (const struct ccase *, void *aux, const struct dataset *);
+static void dump_aggregate_info (struct agr_proc *agr,
+                                 struct casewriter *output);
 
 /* Parsing. */
 
@@ -180,10 +167,14 @@ cmd_aggregate (struct lexer *lexer, struct dataset *ds)
   struct dictionary *dict = dataset_dict (ds);
   struct agr_proc agr;
   struct file_handle *out_file = NULL;
+  struct casereader *input = NULL, *group;
+  struct casegrouper *grouper;
+  struct casewriter *output = NULL;
 
   bool copy_documents = false;
   bool presorted = false;
   bool saw_direction;
+  bool ok;
 
   memset(&agr, 0 , sizeof (agr));
   agr.missing = ITEMWISE;
@@ -204,12 +195,12 @@ cmd_aggregate (struct lexer *lexer, struct dataset *ds)
       if (out_file == NULL)
         goto error;
     }
-  
+
   /* Read most of the subcommands. */
   for (;;)
     {
       lex_match (lexer, '/');
-      
+
       if (lex_match_id (lexer, "MISSING"))
 	{
 	  lex_match (lexer, '=');
@@ -229,12 +220,14 @@ cmd_aggregate (struct lexer *lexer, struct dataset *ds)
           int i;
 
 	  lex_match (lexer, '=');
-          agr.sort = sort_parse_criteria (lexer, dict,
-                                          &agr.break_vars, &agr.break_var_cnt,
-                                          &saw_direction, NULL);
+          agr.sort = parse_case_ordering (lexer, dict,
+
+                                          &saw_direction);
           if (agr.sort == NULL)
             goto error;
-	  
+          case_ordering_get_vars (agr.sort,
+                                  &agr.break_vars, &agr.break_var_cnt);
+
           for (i = 0; i < agr.break_var_cnt; i++)
             dict_clone_var_assert (agr.dict, agr.break_vars[i],
                                    var_get_name (agr.break_vars[i]));
@@ -252,7 +245,7 @@ cmd_aggregate (struct lexer *lexer, struct dataset *ds)
     msg (SW, _("When PRESORTED is specified, specifying sorting directions "
                "with (A) or (D) has no effect.  Output data will be sorted "
                "the same way as the input data."));
-      
+
   /* Read in the aggregate functions. */
   lex_match (lexer, '/');
   if (!parse_aggregate_functions (lexer, dict, &agr))
@@ -260,101 +253,89 @@ cmd_aggregate (struct lexer *lexer, struct dataset *ds)
 
   /* Delete documents. */
   if (!copy_documents)
-    dict_set_documents (agr.dict, NULL);
+    dict_clear_documents (agr.dict);
 
   /* Cancel SPLIT FILE. */
   dict_set_split_vars (agr.dict, NULL, 0);
-  
+
   /* Initialize. */
   agr.case_cnt = 0;
-  case_create (&agr.agr_case, dict_get_next_value_idx (agr.dict));
 
-  /* Output to active file or external file? */
-  if (out_file == NULL) 
+  if (out_file == NULL)
     {
       /* The active file will be replaced by the aggregated data,
          so TEMPORARY is moot. */
       proc_cancel_temporary_transformations (ds);
+      proc_discard_output (ds);
+      output = autopaging_writer_create (dict_get_next_value_idx (agr.dict));
+    }
+  else
+    {
+      output = any_writer_open (out_file, agr.dict);
+      if (output == NULL)
+        goto error;
+    }
+
+  input = proc_open (ds);
+  if (agr.sort != NULL && !presorted)
+    {
+      input = sort_execute (input, agr.sort);
+      agr.sort = NULL;
+    }
+
+  for (grouper = casegrouper_create_vars (input, agr.break_vars,
+                                          agr.break_var_cnt);
+       casegrouper_get_next_group (grouper, &group);
+       casereader_destroy (group))
+    {
+      struct ccase c;
 
-      if (agr.sort != NULL && !presorted) 
+      if (!casereader_peek (group, 0, &c))
         {
-          if (!sort_active_file_in_place (ds, agr.sort))
-            goto error;
+          casereader_destroy (group);
+          continue;
         }
+      initialize_aggregate_info (&agr, &c);
+      case_destroy (&c);
+
+      for (; casereader_read (group, &c); case_destroy (&c))
+        accumulate_aggregate_info (&agr, &c);
+      dump_aggregate_info (&agr, output);
+    }
+  if (!casegrouper_destroy (grouper))
+    goto error;
+
+  if (!proc_commit (ds))
+    {
+      input = NULL;
+      goto error;
+    }
+  input = NULL;
 
-      agr.sink = create_case_sink (&storage_sink_class, agr.dict, NULL);
-      if (agr.sink->class->open != NULL)
-        agr.sink->class->open (agr.sink);
-      proc_set_sink (ds, 
-		     create_case_sink (&null_sink_class, 
-				       dict, NULL));
-      if (!procedure (ds, agr_to_active_file, &agr))
+  if (out_file == NULL)
+    {
+      struct casereader *next_input = casewriter_make_reader (output);
+      if (next_input == NULL)
         goto error;
-      if (agr.case_cnt > 0) 
-        {
-          dump_aggregate_info (&agr, &agr.agr_case);
-          if (!agr.sink->class->write (agr.sink, &agr.agr_case))
-            goto error;
-        }
-      discard_variables (ds);
-      dict_destroy (dict);
-      dataset_set_dict (ds, agr.dict);
+
+      proc_set_active_file (ds, next_input, agr.dict);
       agr.dict = NULL;
-      proc_set_source (ds, 
-		       agr.sink->class->make_source (agr.sink));
-      free_case_sink (agr.sink);
     }
   else
     {
-      agr.writer = any_writer_open (out_file, agr.dict);
-      if (agr.writer == NULL)
-        goto error;
-      
-      if (agr.sort != NULL && !presorted) 
-        {
-          /* Sorting is needed. */
-          struct casefile *dst;
-          struct casereader *reader;
-          struct ccase c;
-          bool ok = true;
-          
-          dst = sort_active_file_to_casefile (ds, agr.sort);
-          if (dst == NULL)
-            goto error;
-          reader = casefile_get_destructive_reader (dst);
-          while (ok && casereader_read_xfer (reader, &c)) 
-            {
-              if (aggregate_single_case (&agr, &c, &agr.agr_case)) 
-                ok = any_writer_write (agr.writer, &agr.agr_case);
-              case_destroy (&c);
-            }
-          casereader_destroy (reader);
-          if (ok)
-            ok = !casefile_error (dst);
-          casefile_destroy (dst);
-          if (!ok)
-            goto error;
-        }
-      else 
-        {
-          /* Active file is already sorted. */
-          if (!procedure (ds, presorted_agr_to_sysfile, &agr))
-            goto error;
-        }
-      
-      if (agr.case_cnt > 0) 
-        {
-          dump_aggregate_info (&agr, &agr.agr_case);
-          any_writer_write (agr.writer, &agr.agr_case);
-        }
-      if (any_writer_error (agr.writer))
+      ok = casewriter_destroy (output);
+      output = NULL;
+      if (!ok)
         goto error;
     }
-  
+
   agr_destroy (&agr);
   return CMD_SUCCESS;
 
 error:
+  if (input != NULL)
+    proc_commit (ds);
+  casewriter_destroy (output);
   agr_destroy (&agr);
   return CMD_CASCADING_FAILURE;
 }
@@ -374,13 +355,13 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
       size_t n_dest;
       struct string function_name;
 
-      int include_missing;
+      enum mv_class exclude;
       const struct agr_func *function;
       int func_index;
 
       union agr_argument arg[2];
 
-      struct variable **src;
+      const struct variable **src;
       size_t n_src;
 
       size_t i;
@@ -393,12 +374,13 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
       n_src = 0;
       arg[0].c = NULL;
       arg[1].c = NULL;
+      ds_init_empty (&function_name);
 
       /* Parse the list of target variables. */
       while (!lex_match (lexer, '='))
 	{
 	  size_t n_dest_prev = n_dest;
-	  
+
 	  if (!parse_DATA_LIST_vars (lexer, &dest, &n_dest,
                                      PV_APPEND | PV_SINGLE | PV_NO_SCRATCH))
 	    goto error;
@@ -413,7 +395,7 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 	  }
 
 
-	  
+
 	  if (lex_token (lexer) == T_STRING)
 	    {
 	      struct string label;
@@ -433,21 +415,21 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 	  goto error;
 	}
 
-      include_missing = 0;
+      exclude = MV_ANY;
 
-      ds_init_string (&function_name, lex_tokstr (lexer));
+      ds_assign_string (&function_name, lex_tokstr (lexer));
 
       ds_chomp (&function_name, '.');
 
       if (lex_tokid(lexer)[strlen (lex_tokid (lexer)) - 1] == '.')
-	  include_missing = 1;
+        exclude = MV_SYSTEM;
 
       for (function = agr_func_tab; function->name; function++)
 	if (!strcasecmp (function->name, ds_cstr (&function_name)))
 	  break;
       if (NULL == function->name)
 	{
-	  msg (SE, _("Unknown aggregation function %s."), 
+	  msg (SE, _("Unknown aggregation function %s."),
 	       ds_cstr (&function_name));
 	  goto error;
 	}
@@ -479,7 +461,7 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 	    else if (function->n_args)
 	      pv_opts |= PV_SAME_TYPE;
 
-	    if (!parse_variables (lexer, dict, &src, &n_src, pv_opts))
+	    if (!parse_variables_const (lexer, dict, &src, &n_src, pv_opts))
 	      goto error;
 	  }
 
@@ -489,7 +471,7 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 	    for (i = 0; i < function->n_args; i++)
 	      {
 		int type;
-	    
+
 		lex_match (lexer, ',');
 		if (lex_token (lexer) == T_STRING)
 		  {
@@ -503,11 +485,11 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 		  }
                 else
                   {
-		    msg (SE, _("Missing argument %d to %s."), i + 1,
-                         function->name);
+		    msg (SE, _("Missing argument %zu to %s."),
+                         i + 1, function->name);
 		    goto error;
 		  }
-	    
+
 		lex_get (lexer);
 
 		if (type != var_get_type (src[0]))
@@ -525,7 +507,7 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 	      lex_error (lexer, _("expecting `)'"));
 	      goto error;
 	    }
-	  
+
 	  /* Now check that the number of source variables match
 	     the number of target variables.  If we check earlier
 	     than this, the user can get very misleading error
@@ -534,14 +516,14 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 	     like `unknown variable t'. */
 	  if (n_src != n_dest)
 	    {
-	      msg (SE, _("Number of source variables (%u) does not match "
-			 "number of target variables (%u)."),
-		   (unsigned) n_src, (unsigned) n_dest);
+	      msg (SE, _("Number of source variables (%zu) does not match "
+			 "number of target variables (%zu)."),
+		    n_src, n_dest);
 	      goto error;
 	    }
 
           if ((func_index == PIN || func_index == POUT
-              || func_index == FIN || func_index == FOUT) 
+              || func_index == FIN || func_index == FOUT)
               && (var_is_numeric (src[0])
                   ? arg[0].f > arg[1].f
                   : str_compare_rpad (arg[0].c, arg[1].c) > 0))
@@ -549,14 +531,14 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
               union agr_argument t = arg[0];
               arg[0] = arg[1];
               arg[1] = t;
-                  
+
               msg (SW, _("The value arguments passed to the %s function "
                          "are out-of-order.  They will be treated as if "
                          "they had been specified in the correct order."),
                    function->name);
             }
 	}
-	
+
       /* Finally add these to the linked list of aggregation
          variables. */
       for (i = 0; i < n_dest; i++)
@@ -571,18 +553,18 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
           tail = v;
 	  tail->next = NULL;
           v->moments = NULL;
-	  
+
 	  /* Create the target variable in the aggregate
              dictionary. */
 	  {
 	    struct variable *destvar;
-	    
+
 	    v->function = func_index;
 
 	    if (src)
 	      {
 		v->src = src[i];
-		
+
 		if (var_is_alpha (src[i]))
 		  {
 		    v->function |= FSTRING;
@@ -596,12 +578,12 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
                     assert (var_is_numeric (v->src)
                             || function->alpha_type == VAR_NUMERIC);
                     destvar = dict_create_var (agr->dict, dest[i], 0);
-                    if (destvar != NULL) 
+                    if (destvar != NULL)
                       {
                         struct fmt_spec f;
                         if ((func_index == N || func_index == NMISS)
                             && dict_get_weight (dict) != NULL)
-                          f = fmt_for_output (FMT_F, 8, 2); 
+                          f = fmt_for_output (FMT_F, 8, 2);
                         else
                           f = function->format;
                         var_set_both_formats (destvar, &f);
@@ -612,12 +594,12 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 		v->src = NULL;
 		destvar = dict_create_var (agr->dict, dest[i], 0);
                 if (func_index == N_NO_VARS && dict_get_weight (dict) != NULL)
-                  f = fmt_for_output (FMT_F, 8, 2); 
+                  f = fmt_for_output (FMT_F, 8, 2);
                 else
                   f = function->format;
                 var_set_both_formats (destvar, &f);
 	      }
-	  
+
 	    if (!destvar)
 	      {
 		msg (SE, _("Variable name %s is not unique within the "
@@ -634,8 +616,8 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 
 	    v->dest = destvar;
 	  }
-	  
-	  v->include_missing = include_missing;
+
+	  v->exclude = exclude;
 
 	  if (v->src != NULL)
 	    {
@@ -649,7 +631,7 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 		  v->arg[j].c = xstrdup (arg[j].c);
 	    }
 	}
-      
+
       if (src != NULL && var_is_alpha (src[0]))
 	for (i = 0; i < function->n_args; i++)
 	  {
@@ -670,7 +652,7 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 	  return false;
 	}
       continue;
-      
+
     error:
       ds_destroy (&function_name);
       for (i = 0; i < n_dest; i++)
@@ -689,7 +671,7 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 	    arg[i].c = NULL;
 	  }
       free (src);
-	
+
       return false;
     }
 }
@@ -700,9 +682,7 @@ agr_destroy (struct agr_proc *agr)
 {
   struct agr_var *iter, *next;
 
-  any_writer_close (agr->writer);
-  if (agr->sort != NULL)
-    sort_destroy_criteria (agr->sort);
+  case_ordering_destroy (agr->sort);
   free (agr->break_vars);
   case_destroy (&agr->break_case);
   for (iter = agr->agr_vars; iter; iter = next)
@@ -725,44 +705,13 @@ agr_destroy (struct agr_proc *agr)
     }
   if (agr->dict != NULL)
     dict_destroy (agr->dict);
-
-  case_destroy (&agr->agr_case);
 }
 
 /* Execution. */
 
-static void accumulate_aggregate_info (struct agr_proc *,
-                                       const struct ccase *);
-static void dump_aggregate_info (struct agr_proc *, struct ccase *);
-
-/* Processes a single case INPUT for aggregation.  If output is
-   warranted, writes it to OUTPUT and returns true.
-   Otherwise, returns false and OUTPUT is unmodified. */
-static bool
-aggregate_single_case (struct agr_proc *agr,
-                       const struct ccase *input, struct ccase *output)
-{
-  bool finished_group = false;
-  
-  if (agr->case_cnt++ == 0)
-    initialize_aggregate_info (agr, input);
-  else if (case_compare (&agr->break_case, input,
-                         agr->break_vars, agr->break_var_cnt))
-    {
-      dump_aggregate_info (agr, output);
-      finished_group = true;
-
-      initialize_aggregate_info (agr, input);
-    }
-
-  accumulate_aggregate_info (agr, input);
-  return finished_group;
-}
-
 /* Accumulates aggregation data from the case INPUT. */
-static void 
-accumulate_aggregate_info (struct agr_proc *agr,
-                           const struct ccase *input)
+static void
+accumulate_aggregate_info (struct agr_proc *agr, const struct ccase *input)
 {
   struct agr_var *iter;
   double weight;
@@ -776,9 +725,7 @@ accumulate_aggregate_info (struct agr_proc *agr,
 	const union value *v = case_data (input, iter->src);
         int src_width = var_get_width (iter->src);
 
-	if (iter->include_missing
-            ? var_is_numeric (iter->src) && v->f == SYSMIS
-            : var_is_value_missing (iter->src, v))
+        if (var_is_value_missing (iter->src, v, iter->exclude))
 	  {
 	    switch (iter->function)
 	      {
@@ -791,10 +738,10 @@ accumulate_aggregate_info (struct agr_proc *agr,
 		iter->int1++;
 		break;
 	      }
-	    iter->missing = 1;
+	    iter->saw_missing = true;
 	    continue;
 	  }
-	
+
 	/* This is horrible.  There are too many possibilities. */
 	switch (iter->function)
 	  {
@@ -932,35 +879,37 @@ accumulate_aggregate_info (struct agr_proc *agr,
     }
 }
 
-/* We've come to a record that differs from the previous in one or
-   more of the break variables.  Make an output record from the
-   accumulated statistics in the OUTPUT case. */
-static void 
-dump_aggregate_info (struct agr_proc *agr, struct ccase *output)
+/* Writes an aggregated record to OUTPUT. */
+static void
+dump_aggregate_info (struct agr_proc *agr, struct casewriter *output)
 {
+  struct ccase c;
+
+  case_create (&c, dict_get_next_value_idx (agr->dict));
+
   {
     int value_idx = 0;
     int i;
 
-    for (i = 0; i < agr->break_var_cnt; i++) 
+    for (i = 0; i < agr->break_var_cnt; i++)
       {
-        struct variable *v = agr->break_vars[i];
+        const struct variable *v = agr->break_vars[i];
         size_t value_cnt = var_get_value_cnt (v);
-        memcpy (case_data_rw_idx (output, value_idx),
+        memcpy (case_data_rw_idx (&c, value_idx),
                 case_data (&agr->break_case, v),
                 sizeof (union value) * value_cnt);
-        value_idx += value_cnt; 
+        value_idx += value_cnt;
       }
   }
-  
+
   {
     struct agr_var *i;
-  
+
     for (i = agr->agr_vars; i; i = i->next)
       {
-	union value *v = case_data_rw (output, i->dest);
+	union value *v = case_data_rw (&c, i->dest);
 
-	if (agr->missing == COLUMNWISE && i->missing != 0
+	if (agr->missing == COLUMNWISE && i->saw_missing
 	    && (i->function & FUNC) != N && (i->function & FUNC) != NU
 	    && (i->function & FUNC) != NMISS && (i->function & FUNC) != NUMISS)
 	  {
@@ -970,7 +919,7 @@ dump_aggregate_info (struct agr_proc *agr, struct ccase *output)
 	      v->f = SYSMIS;
 	    continue;
 	  }
-	
+
 	switch (i->function)
 	  {
 	  case SUM:
@@ -989,7 +938,7 @@ dump_aggregate_info (struct agr_proc *agr, struct ccase *output)
               if (variance != SYSMIS)
                 v->f = sqrt (variance);
               else
-                v->f = SYSMIS; 
+                v->f = SYSMIS;
             }
 	    break;
 	  case MAX:
@@ -1061,6 +1010,8 @@ dump_aggregate_info (struct agr_proc *agr, struct ccase *output)
 	  }
       }
   }
+
+  casewriter_write (output, &c);
 }
 
 /* Resets the state for all the aggregate functions. */
@@ -1074,7 +1025,7 @@ initialize_aggregate_info (struct agr_proc *agr, const struct ccase *input)
 
   for (iter = agr->agr_vars; iter; iter = iter->next)
     {
-      iter->missing = 0;
+      iter->saw_missing = false;
       iter->dbl[0] = iter->dbl[1] = iter->dbl[2] = 0.0;
       iter->int1 = iter->int2 = 0;
       switch (iter->function)
@@ -1102,31 +1053,3 @@ initialize_aggregate_info (struct agr_proc *agr, const struct ccase *input)
 	}
     }
 }
-
-/* Aggregate each case as it comes through.  Cases which aren't needed
-   are dropped.
-   Returns true if successful, false if an I/O error occurred. */
-static bool
-agr_to_active_file (const struct ccase *c, void *agr_, const struct dataset *ds UNUSED)
-{
-  struct agr_proc *agr = agr_;
-
-  if (aggregate_single_case (agr, c, &agr->agr_case)) 
-    return agr->sink->class->write (agr->sink, &agr->agr_case);
-
-  return true;
-}
-
-/* Aggregate the current case and output it if we passed a
-   breakpoint. */
-static bool
-presorted_agr_to_sysfile (const struct ccase *c, void *agr_, 
-			  const struct dataset *ds UNUSED) 
-{
-  struct agr_proc *agr = agr_;
-
-  if (aggregate_single_case (agr, c, &agr->agr_case)) 
-    return any_writer_write (agr->writer, &agr->agr_case);
-  
-  return true;
-}