Revise.

[pspp] / src / vfm.c

/* PSPP - computes sample statistics.
   Copyright (C) 1997-9, 2000 Free Software Foundation, Inc.
   Written by Ben Pfaff <blp@gnu.org>.

   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 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. */

#include <config.h>
#include "vfm.h"
#include "vfmP.h"
#include "error.h"
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#if HAVE_UNISTD_H
#include <unistd.h>     /* Required by SunOS4. */
#endif
#include "alloc.h"
#include "case.h"
#include "casefile.h"
#include "dictionary.h"
#include "do-ifP.h"
#include "error.h"
#include "expressions/public.h"
#include "misc.h"
#include "settings.h"
#include "som.h"
#include "str.h"
#include "tab.h"
#include "var.h"
#include "value-labels.h"

#include "gettext.h"
#define _(msgid) gettext (msgid)

/*
   Virtual File Manager (vfm):

   vfm is used to process data files.  It uses the model that
   data is read from one stream (the data source), processed,
   then written to another (the data sink).  The data source is
   then deleted and the data sink becomes the data source for the
   next procedure. */

/* Procedure execution data. */
struct write_case_data
  {
    /* Function to call for each case. */
    int (*proc_func) (struct ccase *, void *); /* Function. */
    void *aux;                                 /* Auxiliary data. */ 

    struct ccase trns_case;     /* Case used for transformations. */
    struct ccase sink_case;     /* Case written to sink, if
                                   compaction is necessary. */
    size_t cases_written;       /* Cases output so far. */
    size_t cases_analyzed;      /* Cases passed to procedure so far. */
  };

/* The current active file, from which cases are read. */
struct case_source *vfm_source;

/* The replacement active file, to which cases are written. */
struct case_sink *vfm_sink;

/* Nonzero if the case needs to have values deleted before being
   stored, zero otherwise. */
static int compaction_necessary;

/* Time at which vfm was last invoked. */
time_t last_vfm_invocation;

/* Lag queue. */
int n_lag;                      /* Number of cases to lag. */
static int lag_count;           /* Number of cases in lag_queue so far. */
static int lag_head;            /* Index where next case will be added. */
static struct ccase *lag_queue; /* Array of n_lag ccase * elements. */

static void internal_procedure (int (*proc_func) (struct ccase *, void *),
                                void *aux);
static void create_trns_case (struct ccase *, struct dictionary *);
static void open_active_file (void);
static int write_case (struct write_case_data *wc_data);
static int execute_transformations (struct ccase *c,
                                    struct trns_header **trns,
                                    int first_idx, int last_idx,
                                    int case_num);
static int filter_case (const struct ccase *c, int case_num);
static void lag_case (const struct ccase *c);
static void clear_case (struct ccase *c);
static void close_active_file (void);
\f
/* Public functions. */

/* Reads the data from the input program and writes it to a new
   active file.  For each case we read from the input program, we
   do the following

   1. Execute permanent transformations.  If these drop the case,
      start the next case from step 1.

   2. N OF CASES.  If we have already written N cases, start the
      next case from step 1.
   
   3. Write case to replacement active file.
   
   4. Execute temporary transformations.  If these drop the case,
      start the next case from step 1.
      
   5. FILTER, PROCESS IF.  If these drop the case, start the next
      case from step 1.
   
   6. Post-TEMPORARY N OF CASES.  If we have already analyzed N
      cases, start the next case from step 1.
      
   7. Pass case to PROC_FUNC, passing AUX as auxiliary data. */
void
procedure (int (*proc_func) (struct ccase *, void *), void *aux)
{
  if (proc_func == NULL
      && case_source_is_class (vfm_source, &storage_source_class)
      && vfm_sink == NULL
      && !temporary
      && n_trns == 0)
    {
      /* Nothing to do. */
      return;
    }

  open_active_file ();
  internal_procedure (proc_func, aux);
  close_active_file ();
}

/* Executes a procedure, as procedure(), except that the caller
   is responsible for calling open_active_file() and
   close_active_file(). */
static void
internal_procedure (int (*proc_func) (struct ccase *, void *), void *aux) 
{
  static int recursive_call;

  struct write_case_data wc_data;

  assert (++recursive_call == 1);

  wc_data.proc_func = proc_func;
  wc_data.aux = aux;
  create_trns_case (&wc_data.trns_case, default_dict);
  case_create (&wc_data.sink_case, dict_get_next_value_idx (default_dict));
  wc_data.cases_written = 0;

  last_vfm_invocation = time (NULL);

  if (vfm_source != NULL) 
    vfm_source->class->read (vfm_source,
                             &wc_data.trns_case,
                             write_case, &wc_data);

  case_destroy (&wc_data.sink_case);
  case_destroy (&wc_data.trns_case);

  assert (--recursive_call == 0);
}

/* Creates and returns a case, initializing it from the vectors
   that say which `value's need to be initialized just once, and
   which ones need to be re-initialized before every case. */
static void
create_trns_case (struct ccase *trns_case, struct dictionary *dict)
{
  size_t var_cnt = dict_get_var_cnt (dict);
  size_t i;

  case_create (trns_case, dict_get_next_value_idx (dict));
  for (i = 0; i < var_cnt; i++) 
    {
      struct variable *v = dict_get_var (dict, i);
      union value *value = case_data_rw (trns_case, v->fv);

      if (v->type == NUMERIC)
        value->f = v->reinit ? 0.0 : SYSMIS;
      else
        memset (value->s, ' ', v->width);
    }
}

/* Makes all preparations for reading from the data source and writing
   to the data sink. */
static void
open_active_file (void)
{
  /* Make temp_dict refer to the dictionary right before data
     reaches the sink */
  if (!temporary)
    {
      temp_trns = n_trns;
      temp_dict = default_dict;
    }

  /* Figure out compaction. */
  compaction_necessary = (dict_get_next_value_idx (temp_dict)
                          != dict_get_compacted_value_cnt (temp_dict));

  /* Prepare sink. */
  if (vfm_sink == NULL)
    vfm_sink = create_case_sink (&storage_sink_class, temp_dict, NULL);
  if (vfm_sink->class->open != NULL)
    vfm_sink->class->open (vfm_sink);

  /* Allocate memory for lag queue. */
  if (n_lag > 0)
    {
      int i;
  
      lag_count = 0;
      lag_head = 0;
      lag_queue = xmalloc (n_lag * sizeof *lag_queue);
      for (i = 0; i < n_lag; i++)
        case_nullify (&lag_queue[i]);
    }

  /* Close any unclosed DO IF or LOOP constructs. */
  discard_ctl_stack ();
}

/* Transforms trns_case and writes it to the replacement active
   file if advisable.  Returns nonzero if more cases can be
   accepted, zero otherwise.  Do not call this function again
   after it has returned zero once.  */
static int
write_case (struct write_case_data *wc_data)
{
  /* Execute permanent transformations.  */
  if (!execute_transformations (&wc_data->trns_case, t_trns, f_trns, temp_trns,
                                wc_data->cases_written + 1))
    goto done;

  /* N OF CASES. */
  if (dict_get_case_limit (default_dict)
      && wc_data->cases_written >= dict_get_case_limit (default_dict))
    goto done;
  wc_data->cases_written++;

  /* Write case to LAG queue. */
  if (n_lag)
    lag_case (&wc_data->trns_case);

  /* Write case to replacement active file. */
  if (vfm_sink->class->write != NULL) 
    {
      if (compaction_necessary) 
        {
          dict_compact_case (temp_dict, &wc_data->sink_case,
                             &wc_data->trns_case);
          vfm_sink->class->write (vfm_sink, &wc_data->sink_case);
        }
      else
        vfm_sink->class->write (vfm_sink, &wc_data->trns_case);
    }
  
  /* Execute temporary transformations. */
  if (!execute_transformations (&wc_data->trns_case, t_trns, temp_trns, n_trns,
                                wc_data->cases_written))
    goto done;
  
  /* FILTER, PROCESS IF, post-TEMPORARY N OF CASES. */
  if (filter_case (&wc_data->trns_case, wc_data->cases_written)
      || (dict_get_case_limit (temp_dict)
          && wc_data->cases_analyzed >= dict_get_case_limit (temp_dict)))
    goto done;
  wc_data->cases_analyzed++;

  /* Pass case to procedure. */
  if (wc_data->proc_func != NULL)
    wc_data->proc_func (&wc_data->trns_case, wc_data->aux);

 done:
  clear_case (&wc_data->trns_case);
  return 1;
}

/* Transforms case C using the transformations in TRNS[] with
   indexes FIRST_IDX through LAST_IDX, exclusive.  Case C will
   become case CASE_NUM (1-based) in the output file.  Returns
   zero if the case was filtered out by one of the
   transformations, nonzero otherwise. */
static int
execute_transformations (struct ccase *c,
                         struct trns_header **trns,
                         int first_idx, int last_idx,
                         int case_num) 
{
  int idx;

  for (idx = first_idx; idx != last_idx; )
    {
      int retval = trns[idx]->proc (trns[idx], c, case_num);
      switch (retval)
        {
        case -1:
          idx++;
          break;
          
        case -2:
          return 0;
          
        default:
          idx = retval;
          break;
        }
    }

  return 1;
}

/* Returns nonzero if case C with case number CASE_NUM should be
   exclude as specified on FILTER or PROCESS IF, otherwise
   zero. */
static int
filter_case (const struct ccase *c, int case_idx)
{
  /* FILTER. */
  struct variable *filter_var = dict_get_filter (default_dict);
  if (filter_var != NULL) 
    {
      double f = case_num (c, filter_var->fv);
      if (f == 0.0 || f == SYSMIS || is_num_user_missing (f, filter_var))
        return 1;
    }

  /* PROCESS IF. */
  if (process_if_expr != NULL
      && expr_evaluate_num (process_if_expr, c, case_idx) != 1.0)
    return 1;

  return 0;
}

/* Add C to the lag queue. */
static void
lag_case (const struct ccase *c)
{
  if (lag_count < n_lag)
    lag_count++;
  case_destroy (&lag_queue[lag_head]);
  case_clone (&lag_queue[lag_head], c);
  if (++lag_head >= n_lag)
    lag_head = 0;
}

/* Clears the variables in C that need to be cleared between
   processing cases.  */
static void
clear_case (struct ccase *c)
{
  size_t var_cnt = dict_get_var_cnt (default_dict);
  size_t i;
  
  for (i = 0; i < var_cnt; i++) 
    {
      struct variable *v = dict_get_var (default_dict, i);
      if (v->init && v->reinit) 
        {
          if (v->type == NUMERIC)
            case_data_rw (c, v->fv)->f = SYSMIS;
          else
            memset (case_data_rw (c, v->fv)->s, ' ', v->width);
        } 
    }
}

/* Closes the active file. */
static void
close_active_file (void)
{
  /* Free memory for lag queue, and turn off lagging. */
  if (n_lag > 0)
    {
      int i;
      
      for (i = 0; i < n_lag; i++)
        case_destroy (&lag_queue[i]);
      free (lag_queue);
      n_lag = 0;
    }
  
  /* Dictionary from before TEMPORARY becomes permanent.. */
  if (temporary)
    {
      dict_destroy (default_dict);
      default_dict = temp_dict;
      temp_dict = NULL;
    }

  /* Finish compaction. */
  if (compaction_necessary)
    dict_compact_values (default_dict);
    
  /* Free data source. */
  free_case_source (vfm_source);
  vfm_source = NULL;

  /* Old data sink becomes new data source. */
  if (vfm_sink->class->make_source != NULL)
    vfm_source = vfm_sink->class->make_source (vfm_sink);
  free_case_sink (vfm_sink);
  vfm_sink = NULL;

  /* Cancel TEMPORARY, PROCESS IF, FILTER, N OF CASES, vectors,
     and get rid of all the transformations. */
  cancel_temporary ();
  expr_free (process_if_expr);
  process_if_expr = NULL;
  if (dict_get_filter (default_dict) != NULL && !FILTER_before_TEMPORARY)
    dict_set_filter (default_dict, NULL);
  dict_set_case_limit (default_dict, 0);
  dict_clear_vectors (default_dict);
  cancel_transformations ();
}
\f
/* Storage case stream. */

/* Information about storage sink or source. */
struct storage_stream_info 
  {
    struct casefile *casefile;  /* Storage. */
  };

/* Initializes a storage sink. */
static void
storage_sink_open (struct case_sink *sink)
{
  struct storage_stream_info *info;

  sink->aux = info = xmalloc (sizeof *info);
  info->casefile = casefile_create (sink->value_cnt);
}

/* Destroys storage stream represented by INFO. */
static void
destroy_storage_stream_info (struct storage_stream_info *info) 
{
  if (info != NULL) 
    {
      casefile_destroy (info->casefile);
      free (info); 
    }
}

/* Writes case C to the storage sink SINK. */
static void
storage_sink_write (struct case_sink *sink, const struct ccase *c)
{
  struct storage_stream_info *info = sink->aux;

  casefile_append (info->casefile, c);
}

/* Destroys internal data in SINK. */
static void
storage_sink_destroy (struct case_sink *sink)
{
  destroy_storage_stream_info (sink->aux);
}

/* Closes the sink and returns a storage source to read back the
   written data. */
static struct case_source *
storage_sink_make_source (struct case_sink *sink) 
{
  struct case_source *source
    = create_case_source (&storage_source_class, sink->aux);
  sink->aux = NULL;
  return source;
}

/* Storage sink. */
const struct case_sink_class storage_sink_class = 
  {
    "storage",
    storage_sink_open,
    storage_sink_write,
    storage_sink_destroy,
    storage_sink_make_source,
  };
\f
/* Storage source. */

/* Returns the number of cases that will be read by
   storage_source_read(). */
static int
storage_source_count (const struct case_source *source) 
{
  struct storage_stream_info *info = source->aux;

  return casefile_get_case_cnt (info->casefile);
}

/* Reads all cases from the storage source and passes them one by one to
   write_case(). */
static void
storage_source_read (struct case_source *source,
                     struct ccase *output_case,
                     write_case_func *write_case, write_case_data wc_data)
{
  struct storage_stream_info *info = source->aux;
  struct ccase casefile_case;
  struct casereader *reader;

  for (reader = casefile_get_reader (info->casefile);
       casereader_read (reader, &casefile_case);
       case_destroy (&casefile_case))
    {
      case_copy (output_case, 0,
                 &casefile_case, 0,
                 casefile_get_value_cnt (info->casefile));
      write_case (wc_data);
    }
  casereader_destroy (reader);
}

/* Destroys the source's internal data. */
static void
storage_source_destroy (struct case_source *source)
{
  destroy_storage_stream_info (source->aux);
}

/* Storage source. */
const struct case_source_class storage_source_class = 
  {
    "storage",
    storage_source_count,
    storage_source_read,
    storage_source_destroy,
  };

struct casefile *
storage_source_get_casefile (struct case_source *source) 
{
  struct storage_stream_info *info = source->aux;

  assert (source->class == &storage_source_class);
  return info->casefile;
}

struct case_source *
storage_source_create (struct casefile *cf)
{
  struct storage_stream_info *info;

  info = xmalloc (sizeof *info);
  info->casefile = cf;

  return create_case_source (&storage_source_class, info);
}
\f
/* Null sink.  Used by a few procedures that keep track of output
   themselves and would throw away anything that the sink
   contained anyway. */

const struct case_sink_class null_sink_class = 
  {
    "null",
    NULL,
    NULL,
    NULL,
    NULL,
  };
\f
/* Returns a pointer to the lagged case from N_BEFORE cases before the
   current one, or NULL if there haven't been that many cases yet. */
struct ccase *
lagged_case (int n_before)
{
  assert (n_before >= 1 );
  assert (n_before <= n_lag);

  if (n_before <= lag_count)
    {
      int index = lag_head - n_before;
      if (index < 0)
        index += n_lag;
      return &lag_queue[index];
    }
  else
    return NULL;
}
   
/* Appends TRNS to t_trns[], the list of all transformations to be
   performed on data as it is read from the active file. */
void
add_transformation (struct trns_header * trns)
{
  if (n_trns >= m_trns)
    {
      m_trns += 16;
      t_trns = xrealloc (t_trns, sizeof *t_trns * m_trns);
    }
  t_trns[n_trns] = trns;
  trns->index = n_trns++;
}

/* Cancels all active transformations, including any transformations
   created by the input program. */
void
cancel_transformations (void)
{
  int i;
  for (i = 0; i < n_trns; i++)
    {
      if (t_trns[i]->free)
        t_trns[i]->free (t_trns[i]);
      free (t_trns[i]);
    }
  n_trns = f_trns = 0;
  free (t_trns);
  t_trns = NULL;
  m_trns = 0;
}
\f
/* Creates a case source with class CLASS and auxiliary data AUX
   and based on dictionary DICT. */
struct case_source *
create_case_source (const struct case_source_class *class,
                    void *aux) 
{
  struct case_source *source = xmalloc (sizeof *source);
  source->class = class;
  source->aux = aux;
  return source;
}

/* Destroys case source SOURCE.  It is the caller's responsible to
   call the source's destroy function, if any. */
void
free_case_source (struct case_source *source) 
{
  if (source != NULL) 
    {
      if (source->class->destroy != NULL)
        source->class->destroy (source);
      free (source);
    }
}

/* Returns nonzero if a case source is "complex". */
int
case_source_is_complex (const struct case_source *source) 
{
  return source != NULL && (source->class == &input_program_source_class
                            || source->class == &file_type_source_class);
}

/* Returns nonzero if CLASS is the class of SOURCE. */
int
case_source_is_class (const struct case_source *source,
                      const struct case_source_class *class) 
{
  return source != NULL && source->class == class;
}

/* Creates a case sink to accept cases from the given DICT with
   class CLASS and auxiliary data AUX. */
struct case_sink *
create_case_sink (const struct case_sink_class *class,
                  const struct dictionary *dict,
                  void *aux) 
{
  struct case_sink *sink = xmalloc (sizeof *sink);
  sink->class = class;
  sink->value_cnt = dict_get_compacted_value_cnt (dict);
  sink->aux = aux;
  return sink;
}

/* Destroys case sink SINK.  */
void
free_case_sink (struct case_sink *sink) 
{
  if (sink != NULL) 
    {
      if (sink->class->destroy != NULL)
        sink->class->destroy (sink);
      free (sink); 
    }
}
\f
/* Represents auxiliary data for handling SPLIT FILE. */
struct split_aux_data 
  {
    size_t case_count;          /* Number of cases so far. */
    struct ccase prev_case;     /* Data in previous case. */

    /* Functions to call... */
    void (*begin_func) (void *);               /* ...before data. */
    int (*proc_func) (struct ccase *, void *); /* ...with data. */
    void (*end_func) (void *);                 /* ...after data. */
    void *func_aux;                            /* Auxiliary data. */ 
  };

static int equal_splits (const struct ccase *, const struct ccase *);
static int procedure_with_splits_callback (struct ccase *, void *);
static void dump_splits (struct ccase *);

/* Like procedure(), but it automatically breaks the case stream
   into SPLIT FILE break groups.  Before each group of cases with
   identical SPLIT FILE variable values, BEGIN_FUNC is called.
   Then PROC_FUNC is called with each case in the group.  
   END_FUNC is called when the group is finished.  FUNC_AUX is
   passed to each of the functions as auxiliary data.

   If the active file is empty, none of BEGIN_FUNC, PROC_FUNC,
   and END_FUNC will be called at all. 

   If SPLIT FILE is not in effect, then there is one break group
   (if the active file is nonempty), and BEGIN_FUNC and END_FUNC
   will be called once. */
void
procedure_with_splits (void (*begin_func) (void *aux),
                       int (*proc_func) (struct ccase *, void *aux),
                       void (*end_func) (void *aux),
                       void *func_aux) 
{
  struct split_aux_data split_aux;

  split_aux.case_count = 0;
  case_nullify (&split_aux.prev_case);
  split_aux.begin_func = begin_func;
  split_aux.proc_func = proc_func;
  split_aux.end_func = end_func;
  split_aux.func_aux = func_aux;

  open_active_file ();
  internal_procedure (procedure_with_splits_callback, &split_aux);
  if (split_aux.case_count > 0 && end_func != NULL)
    end_func (func_aux);
  close_active_file ();

  case_destroy (&split_aux.prev_case);
}

/* procedure() callback used by procedure_with_splits(). */
static int
procedure_with_splits_callback (struct ccase *c, void *split_aux_) 
{
  struct split_aux_data *split_aux = split_aux_;

  /* Start a new series if needed. */
  if (split_aux->case_count == 0
      || !equal_splits (c, &split_aux->prev_case))
    {
      if (split_aux->case_count > 0 && split_aux->end_func != NULL)
        split_aux->end_func (split_aux->func_aux);

      dump_splits (c);
      case_destroy (&split_aux->prev_case);
      case_clone (&split_aux->prev_case, c);

      if (split_aux->begin_func != NULL)
        split_aux->begin_func (split_aux->func_aux);
    }

  split_aux->case_count++;
  if (split_aux->proc_func != NULL)
    return split_aux->proc_func (c, split_aux->func_aux);
  else
    return 1;
}

/* Compares the SPLIT FILE variables in cases A and B and returns
   nonzero only if they differ. */
static int
equal_splits (const struct ccase *a, const struct ccase *b) 
{
  return case_compare (a, b,
                       dict_get_split_vars (default_dict),
                       dict_get_split_cnt (default_dict)) == 0;
}

/* Dumps out the values of all the split variables for the case C. */
static void
dump_splits (struct ccase *c)
{
  struct variable *const *split;
  struct tab_table *t;
  size_t split_cnt;
  int i;

  split_cnt = dict_get_split_cnt (default_dict);
  if (split_cnt == 0)
    return;

  t = tab_create (3, split_cnt + 1, 0);
  tab_dim (t, tab_natural_dimensions);
  tab_vline (t, TAL_1 | TAL_SPACING, 1, 0, split_cnt);
  tab_vline (t, TAL_1 | TAL_SPACING, 2, 0, split_cnt);
  tab_text (t, 0, 0, TAB_NONE, _("Variable"));
  tab_text (t, 1, 0, TAB_LEFT, _("Value"));
  tab_text (t, 2, 0, TAB_LEFT, _("Label"));
  split = dict_get_split_vars (default_dict);
  for (i = 0; i < split_cnt; i++)
    {
      struct variable *v = split[i];
      char temp_buf[80];
      const char *val_lab;

      assert (v->type == NUMERIC || v->type == ALPHA);
      tab_text (t, 0, i + 1, TAB_LEFT | TAT_PRINTF, "%s", v->name);
      
      data_out (temp_buf, &v->print, case_data (c, v->fv));
      
      temp_buf[v->print.w] = 0;
      tab_text (t, 1, i + 1, TAT_PRINTF, "%.*s", v->print.w, temp_buf);

      val_lab = val_labs_find (v->val_labs, *case_data (c, v->fv));
      if (val_lab)
        tab_text (t, 2, i + 1, TAB_LEFT, val_lab);
    }
  tab_flags (t, SOMF_NO_TITLE);
  tab_submit (t);
}
\f
/* Represents auxiliary data for handling SPLIT FILE in a
   multipass procedure. */
struct multipass_split_aux_data 
  {
    struct ccase prev_case;     /* Data in previous case. */
    struct casefile *casefile;  /* Accumulates data for a split. */

    /* Function to call with the accumulated data. */
    void (*split_func) (const struct casefile *, void *);
    void *func_aux;                            /* Auxiliary data. */ 
  };

static int multipass_split_callback (struct ccase *c, void *aux_);
static void multipass_split_output (struct multipass_split_aux_data *);

void
multipass_procedure_with_splits (void (*split_func) (const struct casefile *,
                                                     void *),
                                 void *func_aux) 
{
  struct multipass_split_aux_data aux;

  assert (split_func != NULL);

  open_active_file ();

  case_nullify (&aux.prev_case);
  aux.casefile = NULL;
  aux.split_func = split_func;
  aux.func_aux = func_aux;

  internal_procedure (multipass_split_callback, &aux);
  if (aux.casefile != NULL)
    multipass_split_output (&aux);
  case_destroy (&aux.prev_case);

  close_active_file ();
}

/* procedure() callback used by multipass_procedure_with_splits(). */
static int
multipass_split_callback (struct ccase *c, void *aux_)
{
  struct multipass_split_aux_data *aux = aux_;

  /* Start a new series if needed. */
  if (aux->casefile == NULL || !equal_splits (c, &aux->prev_case))
    {
      /* Pass any cases to split_func. */
      if (aux->casefile != NULL)
        multipass_split_output (aux);

      /* Start a new casefile. */
      aux->casefile = casefile_create (dict_get_next_value_idx (default_dict));

      /* Record split values. */
      dump_splits (c);
      case_destroy (&aux->prev_case);
      case_clone (&aux->prev_case, c);
    }

  casefile_append (aux->casefile, c);

  return 1;
}

static void
multipass_split_output (struct multipass_split_aux_data *aux)
{
  assert (aux->casefile != NULL);
  aux->split_func (aux->casefile, aux->func_aux);
  casefile_destroy (aux->casefile);
  aux->casefile = NULL;
}