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

diff --git a/src/language/stats/aggregate.c b/src/language/stats/aggregate.c
index 4c85592b..0d181bd4 100644
--- a/src/language/stats/aggregate.c
+++ b/src/language/stats/aggregate.c
@@ -1,35 +1,34 @@
-/* PSPP - computes sample statistics.
-   Copyright (C) 1997-9, 2000, 2006 Free Software Foundation, Inc.
+/* PSPP - a program for statistical analysis.
+   Copyright (C) 1997-9, 2000, 2006, 2008, 2009 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.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/subcase.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>
@@ -45,8 +43,11 @@
 #include <libpspp/str.h>
 #include <math/moments.h>
 #include <math/sort.h>
+#include <math/statistic.h>
+#include <math/percentiles.h>
 
 #include "minmax.h"
+#include "xalloc.h"
 
 #include "gettext.h"
 #define _(msgid) gettext (msgid)
@@ -64,24 +65,29 @@ 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;
+    double cc;
+
+    struct variable *subject;
+    struct variable *weight;
+    struct casewriter *writer;
   };
 
 /* Aggregation functions. */
 enum
   {
-    NONE, SUM, MEAN, SD, MAX, MIN, PGT, PLT, PIN, POUT, FGT, FLT, FIN,
+    NONE, SUM, MEAN, MEDIAN, SD, MAX, MIN, PGT, PLT, PIN, POUT, FGT, FLT, FIN,
     FOUT, N, NU, NMISS, NUMISS, FIRST, LAST,
     N_AGR_FUNCS, N_NO_VARS, NU_NO_VARS,
     FUNC = 0x1f, /* Function mask. */
@@ -93,36 +99,37 @@ struct agr_func
   {
     const char *name;		/* Aggregation function name. */
     size_t n_args;              /* Number of arguments. */
-    enum var_type alpha_type;   /* When given ALPHA arguments, output type. */
+    enum val_type alpha_type;   /* When given ALPHA arguments, output type. */
     struct fmt_spec format;	/* Format spec if alpha_type != ALPHA. */
   };
 
 /* 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}},
+    {"MEDIAN",	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}}, 
-    {"LAST",    0, VAR_STRING,  {-1, -1, -1}},
+    {"MAX",     0, VAL_STRING,  {-1, -1, -1}},
+    {"MIN",     0, VAL_STRING,  {-1, -1, -1}},
+    {"PGT",     1, VAL_NUMERIC, {FMT_F, 5, 1}},
+    {"PLT",     1, VAL_NUMERIC, {FMT_F, 5, 1}},
+    {"PIN",     2, VAL_NUMERIC, {FMT_F, 5, 1}},
+    {"POUT",    2, VAL_NUMERIC, {FMT_F, 5, 1}},
+    {"FGT",     1, VAL_NUMERIC, {FMT_F, 5, 3}},
+    {"FLT",     1, VAL_NUMERIC, {FMT_F, 5, 3}},
+    {"FIN",     2, VAL_NUMERIC, {FMT_F, 5, 3}},
+    {"FOUT",    2, VAL_NUMERIC, {FMT_F, 5, 3}},
+    {"N",       0, VAL_NUMERIC, {FMT_F, 7, 0}},
+    {"NU",      0, VAL_NUMERIC, {FMT_F, 7, 0}},
+    {"NMISS",   0, VAL_NUMERIC, {FMT_F, 7, 0}},
+    {"NUMISS",  0, VAL_NUMERIC, {FMT_F, 7, 0}},
+    {"FIRST",   0, VAL_STRING,  {-1, -1, -1}},
+    {"LAST",    0, VAL_STRING,  {-1, -1, -1}},
     {NULL,      0, -1,          {-1, -1, -1}},
-    {"N",       0, VAR_NUMERIC, {FMT_F, 7, 0}},
-    {"NU",      0, VAR_NUMERIC, {FMT_F, 7, 0}},
+    {"N",       0, VAL_NUMERIC, {FMT_F, 7, 0}},
+    {"NU",      0, VAL_NUMERIC, {FMT_F, 7, 0}},
   };
 
 /* Missing value types. */
@@ -133,43 +140,32 @@ 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 subcase sort;                /* Sort criteria (break variables). */
+    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. */
+    struct ccase *break_case;           /* Last values of break variables. */
 
     enum missing_treatment missing;     /* How to treat missing values. */
     struct agr_var *agr_vars;           /* First aggregate variable. */
     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,17 +176,22 @@ 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;
-  case_nullify (&agr.break_case);
+  agr.break_case = NULL;
 
   agr.dict = dict_create ();
   agr.src_dict = dict;
+  subcase_init_empty (&agr.sort);
   dict_set_label (agr.dict, dict_get_label (dict));
   dict_set_documents (agr.dict, dict_get_documents (dict));
 
@@ -204,12 +205,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 +230,11 @@ 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);
-          if (agr.sort == NULL)
+          if (!parse_sort_criteria (lexer, dict, &agr.sort, &agr.break_vars,
+                                    &saw_direction))
             goto error;
-	  
+          agr.break_var_cnt = subcase_get_n_fields (&agr.sort);
+
           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 +252,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,108 +260,98 @@ 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_proto (agr.dict));
+    }
+  else
+    {
+      output = any_writer_open (out_file, agr.dict);
+      if (output == NULL)
+        goto error;
+    }
+
+  input = proc_open (ds);
+  if (!subcase_is_empty (&agr.sort) && !presorted)
+    {
+      input = sort_execute (input, &agr.sort);
+      subcase_clear (&agr.sort);
+    }
 
-      if (agr.sort != NULL && !presorted) 
+  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 = casereader_peek (group, 0);
+      if (c == NULL)
         {
-          if (!sort_active_file_in_place (ds, agr.sort))
-            goto error;
+          casereader_destroy (group);
+          continue;
         }
+      initialize_aggregate_info (&agr, c);
+      case_unref (c);
 
-      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))
+      for (; (c = casereader_read (group)) != NULL; case_unref (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;
+
+  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);
+  fh_unref (out_file);
   return CMD_SUCCESS;
 
 error:
+  if (input != NULL)
+    proc_commit (ds);
+  casewriter_destroy (output);
   agr_destroy (&agr);
+  fh_unref (out_file);
   return CMD_CASCADING_FAILURE;
 }
 
 /* Parse all the aggregate functions. */
 static bool
-parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, struct agr_proc *agr)
+parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict,
+			   struct agr_proc *agr)
 {
   struct agr_var *tail; /* Tail of linked list starting at agr->vars. */
 
@@ -374,13 +364,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 +383,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 +404,7 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 	  }
 
 
-	  
+
 	  if (lex_token (lexer) == T_STRING)
 	    {
 	      struct string label;
@@ -433,21 +424,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 +470,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,25 +480,25 @@ 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)
 		  {
 		    arg[i].c = ds_xstrdup (lex_tokstr (lexer));
-		    type = VAR_STRING;
+		    type = VAL_STRING;
 		  }
 		else if (lex_is_number (lexer))
 		  {
 		    arg[i].f = lex_tokval (lexer);
-		    type = VAR_NUMERIC;
+		    type = VAL_NUMERIC;
 		  }
                 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 +516,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 +525,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,19 +540,19 @@ 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++)
 	{
-	  struct agr_var *v = xmalloc (sizeof *v);
+	  struct agr_var *v = xzalloc (sizeof *v);
 
 	  /* Add variable to chain. */
 	  if (agr->agr_vars != NULL)
@@ -571,37 +562,37 @@ 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;
 		    v->string = xmalloc (var_get_width (src[i]));
 		  }
 
-		if (function->alpha_type == VAR_STRING)
+		if (function->alpha_type == VAL_STRING)
 		  destvar = dict_clone_var (agr->dict, v->src, dest[i]);
 		else
                   {
                     assert (var_is_numeric (v->src)
-                            || function->alpha_type == VAR_NUMERIC);
+                            || function->alpha_type == VAL_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 +603,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 +625,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 +640,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 +661,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 +680,7 @@ parse_aggregate_functions (struct lexer *lexer, const struct dictionary *dict, s
 	    arg[i].c = NULL;
 	  }
       free (src);
-	
+
       return false;
     }
 }
@@ -700,11 +691,9 @@ 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);
+  subcase_destroy (&agr->sort);
   free (agr->break_vars);
-  case_destroy (&agr->break_case);
+  case_unref (agr->break_case);
   for (iter = agr->agr_vars; iter; iter = next)
     {
       next = iter->next;
@@ -721,48 +710,21 @@ agr_destroy (struct agr_proc *agr)
 	}
       else if (iter->function == SD)
         moments1_destroy (iter->moments);
+
+      var_destroy (iter->subject);
+      var_destroy (iter->weight);
+
       free (iter);
     }
   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 +738,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 +751,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)
 	  {
@@ -806,6 +766,25 @@ accumulate_aggregate_info (struct agr_proc *agr,
             iter->dbl[0] += v->f * weight;
             iter->dbl[1] += weight;
             break;
+	  case MEDIAN:
+	    {
+	      double wv ;
+	      struct ccase *cout;
+
+              cout = case_create (casewriter_get_proto (iter->writer));
+
+	      case_data_rw (cout, iter->subject)->f
+                = case_data (input, iter->src)->f;
+
+	      wv = dict_get_case_weight (agr->src_dict, input, NULL);
+
+	      case_data_rw (cout, iter->weight)->f = wv;
+
+	      iter->cc += wv;
+
+	      casewriter_write (iter->writer, cout);
+	    }
+	    break;
 	  case SD:
             moments1_add (iter->moments, v->f, weight);
             break;
@@ -814,8 +793,8 @@ accumulate_aggregate_info (struct agr_proc *agr,
 	    iter->int1 = 1;
 	    break;
 	  case MAX | FSTRING:
-	    if (memcmp (iter->string, v->s, src_width) < 0)
-	      memcpy (iter->string, v->s, src_width);
+	    if (memcmp (iter->string, value_str (v, src_width), src_width) < 0)
+	      memcpy (iter->string, value_str (v, src_width), src_width);
 	    iter->int1 = 1;
 	    break;
 	  case MIN:
@@ -823,8 +802,8 @@ accumulate_aggregate_info (struct agr_proc *agr,
 	    iter->int1 = 1;
 	    break;
 	  case MIN | FSTRING:
-	    if (memcmp (iter->string, v->s, src_width) > 0)
-	      memcpy (iter->string, v->s, src_width);
+	    if (memcmp (iter->string, value_str (v, src_width), src_width) > 0)
+	      memcpy (iter->string, value_str (v, src_width), src_width);
 	    iter->int1 = 1;
 	    break;
 	  case FGT:
@@ -835,7 +814,8 @@ accumulate_aggregate_info (struct agr_proc *agr,
             break;
 	  case FGT | FSTRING:
 	  case PGT | FSTRING:
-            if (memcmp (iter->arg[0].c, v->s, src_width) < 0)
+            if (memcmp (iter->arg[0].c,
+                        value_str (v, src_width), src_width) < 0)
               iter->dbl[0] += weight;
             iter->dbl[1] += weight;
             break;
@@ -847,7 +827,8 @@ accumulate_aggregate_info (struct agr_proc *agr,
             break;
 	  case FLT | FSTRING:
 	  case PLT | FSTRING:
-            if (memcmp (iter->arg[0].c, v->s, src_width) > 0)
+            if (memcmp (iter->arg[0].c,
+                        value_str (v, src_width), src_width) > 0)
               iter->dbl[0] += weight;
             iter->dbl[1] += weight;
             break;
@@ -859,8 +840,10 @@ accumulate_aggregate_info (struct agr_proc *agr,
             break;
 	  case FIN | FSTRING:
 	  case PIN | FSTRING:
-            if (memcmp (iter->arg[0].c, v->s, src_width) <= 0
-                && memcmp (iter->arg[1].c, v->s, src_width) >= 0)
+            if (memcmp (iter->arg[0].c,
+                        value_str (v, src_width), src_width) <= 0
+                && memcmp (iter->arg[1].c,
+                           value_str (v, src_width), src_width) >= 0)
               iter->dbl[0] += weight;
             iter->dbl[1] += weight;
             break;
@@ -872,8 +855,10 @@ accumulate_aggregate_info (struct agr_proc *agr,
             break;
 	  case FOUT | FSTRING:
 	  case POUT | FSTRING:
-            if (memcmp (iter->arg[0].c, v->s, src_width) > 0
-                || memcmp (iter->arg[1].c, v->s, src_width) < 0)
+            if (memcmp (iter->arg[0].c,
+                        value_str (v, src_width), src_width) > 0
+                || memcmp (iter->arg[1].c,
+                           value_str (v, src_width), src_width) < 0)
               iter->dbl[0] += weight;
             iter->dbl[1] += weight;
             break;
@@ -895,7 +880,7 @@ accumulate_aggregate_info (struct agr_proc *agr,
 	  case FIRST | FSTRING:
 	    if (iter->int1 == 0)
 	      {
-		memcpy (iter->string, v->s, src_width);
+		memcpy (iter->string, value_str (v, src_width), src_width);
 		iter->int1 = 1;
 	      }
 	    break;
@@ -904,7 +889,7 @@ accumulate_aggregate_info (struct agr_proc *agr,
 	    iter->int1 = 1;
 	    break;
 	  case LAST | FSTRING:
-	    memcpy (iter->string, v->s, src_width);
+	    memcpy (iter->string, value_str (v, src_width), src_width);
 	    iter->int1 = 1;
 	    break;
           case NMISS:
@@ -932,45 +917,43 @@ 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 (dict_get_proto (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];
-        size_t value_cnt = var_get_value_cnt (v);
-        memcpy (case_data_rw_idx (output, value_idx),
-                case_data (&agr->break_case, v),
-                sizeof (union value) * value_cnt);
-        value_idx += value_cnt; 
+        const struct variable *v = agr->break_vars[i];
+        value_copy (case_data_rw_idx (c, value_idx),
+                    case_data (agr->break_case, v),
+                    var_get_width (v));
+        value_idx++;
       }
   }
-  
+
   {
     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);
+        int width = var_get_width (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)
 	  {
-	    if (var_is_alpha (i->dest))
-	      memset (v->s, ' ', var_get_width (i->dest));
-	    else
-	      v->f = SYSMIS;
+            value_set_missing (v, width);
+	    casewriter_destroy (i->writer);
 	    continue;
 	  }
-	
+
 	switch (i->function)
 	  {
 	  case SUM:
@@ -979,6 +962,25 @@ dump_aggregate_info (struct agr_proc *agr, struct ccase *output)
 	  case MEAN:
 	    v->f = i->dbl[1] != 0.0 ? i->dbl[0] / i->dbl[1] : SYSMIS;
 	    break;
+	  case MEDIAN:
+	    {
+	      struct casereader *sorted_reader;
+	      struct order_stats *median = percentile_create (0.5, i->cc);
+
+	      sorted_reader = casewriter_make_reader (i->writer);
+
+	      order_stats_accumulate (&median, 1,
+				      sorted_reader,
+				      i->weight,
+				      i->subject,
+				      i->exclude);
+
+	      v->f = percentile_calculate ((struct percentile *) median,
+					   PC_HAVERAGE);
+
+	      statistic_destroy ((struct statistic *) median);
+	    }
+	    break;
 	  case SD:
             {
               double variance;
@@ -989,7 +991,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:
@@ -999,9 +1001,9 @@ dump_aggregate_info (struct agr_proc *agr, struct ccase *output)
 	  case MAX | FSTRING:
 	  case MIN | FSTRING:
 	    if (i->int1)
-	      memcpy (v->s, i->string, var_get_width (i->dest));
+	      memcpy (value_str_rw (v, width), i->string, width);
 	    else
-	      memset (v->s, ' ', var_get_width (i->dest));
+              value_set_missing (v, width);
 	    break;
 	  case FGT:
 	  case FGT | FSTRING:
@@ -1038,9 +1040,9 @@ dump_aggregate_info (struct agr_proc *agr, struct ccase *output)
 	  case FIRST | FSTRING:
 	  case LAST | FSTRING:
 	    if (i->int1)
-	      memcpy (v->s, i->string, var_get_width (i->dest));
+	      memcpy (value_str_rw (v, width), i->string, width);
 	    else
-	      memset (v->s, ' ', var_get_width (i->dest));
+              value_set_missing (v, width);
 	    break;
 	  case N_NO_VARS:
 	    v->f = i->dbl[0];
@@ -1061,6 +1063,8 @@ dump_aggregate_info (struct agr_proc *agr, struct ccase *output)
 	  }
       }
   }
+
+  casewriter_write (output, c);
 }
 
 /* Resets the state for all the aggregate functions. */
@@ -1069,12 +1073,12 @@ initialize_aggregate_info (struct agr_proc *agr, const struct ccase *input)
 {
   struct agr_var *iter;
 
-  case_destroy (&agr->break_case);
-  case_clone (&agr->break_case, input);
+  case_unref (agr->break_case);
+  agr->break_case = case_ref (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)
@@ -1091,6 +1095,29 @@ initialize_aggregate_info (struct agr_proc *agr, const struct ccase *input)
 	case MAX | FSTRING:
 	  memset (iter->string, 0, var_get_width (iter->src));
 	  break;
+	case MEDIAN:
+	  {
+            struct caseproto *proto;
+            struct subcase ordering;
+
+            proto = caseproto_create ();
+            proto = caseproto_add_width (proto, 0);
+            proto = caseproto_add_width (proto, 0);
+
+	    if ( ! iter->subject)
+	      iter->subject = var_create_internal (0);
+
+	    if ( ! iter->weight)
+	      iter->weight = var_create_internal (1);
+
+            subcase_init_var (&ordering, iter->subject, SC_ASCEND);
+	    iter->writer = sort_create_writer (&ordering, proto);
+            subcase_destroy (&ordering);
+            caseproto_unref (proto);
+
+	    iter->cc = 0;
+	  }
+	  break;
         case SD:
           if (iter->moments == NULL)
             iter->moments = moments1_create (MOMENT_VARIANCE);
@@ -1102,31 +1129,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;
-}