src/procedure.c

   1 /* PSPP - computes sample statistics.
   2    Copyright (C) 1997-9, 2000, 2006 Free Software Foundation, Inc.
   3    Written by Ben Pfaff <blp@gnu.org>.
   4
   5    This program is free software; you can redistribute it and/or
   6    modify it under the terms of the GNU General Public License as
   7    published by the Free Software Foundation; either version 2 of the
   8    License, or (at your option) any later version.
   9
  10    This program is distributed in the hope that it will be useful, but
  11    WITHOUT ANY WARRANTY; without even the implied warranty of
  12    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13    General Public License for more details.
  14
  15    You should have received a copy of the GNU General Public License
  16    along with this program; if not, write to the Free Software
  17    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  18    02110-1301, USA. */
  19
  20 #include <config.h>
  21
  22 #include <procedure.h>
  23
  24 #include <errno.h>
  25 #include <stdio.h>
  26 #include <stdlib.h>
  27 #include <unistd.h>
  28
  29 #include "expressions/public.h"
  30 #include <data/case-source.h>
  31 #include <data/case-sink.h>
  32 #include <data/case.h>
  33 #include <data/casefile.h>
  34 #include <data/dictionary.h>
  35 #include <data/file-handle-def.h>
  36 #include <data/settings.h>
  37 #include <data/storage-stream.h>
  38 #include <data/value-labels.h>
  39 #include <data/variable.h>
  40 #include <language/control/control-stack.h>
  41 #include <libpspp/alloc.h>
  42 #include <libpspp/message.h>
  43 #include <libpspp/message.h>
  44 #include <libpspp/misc.h>
  45 #include <libpspp/str.h>
  46 #include <output/manager.h>
  47 #include <output/table.h>
  48
  49 #include "gettext.h"
  50 #define _(msgid) gettext (msgid)
  51
  52 /*
  53    Virtual File Manager (vfm):
  54
  55    vfm is used to process data files.  It uses the model that
  56    data is read from one stream (the data source), processed,
  57    then written to another (the data sink).  The data source is
  58    then deleted and the data sink becomes the data source for the
  59    next procedure. */
  60
  61 /* Procedure execution data. */
  62 struct write_case_data
  63   {
  64     /* Function to call for each case. */
  65     bool (*proc_func) (struct ccase *, void *); /* Function. */
  66     void *aux;                                 /* Auxiliary data. */
  67
  68     struct ccase trns_case;     /* Case used for transformations. */
  69     struct ccase sink_case;     /* Case written to sink, if
  70                                    compaction is necessary. */
  71     size_t cases_written;       /* Cases output so far. */
  72     size_t cases_analyzed;      /* Cases passed to procedure so far. */
  73   };
  74
  75 /* The current active file, from which cases are read. */
  76 struct case_source *vfm_source;
  77
  78 /* The replacement active file, to which cases are written. */
  79 struct case_sink *vfm_sink;
  80
  81 /* The compactor used to compact a compact, if necessary;
  82    otherwise a null pointer. */
  83 static struct dict_compactor *compactor;
  84
  85 /* Time at which vfm was last invoked. */
  86 static time_t last_vfm_invocation;
  87
  88 /* Lag queue. */
  89 int n_lag;                      /* Number of cases to lag. */
  90 static int lag_count;           /* Number of cases in lag_queue so far. */
  91 static int lag_head;            /* Index where next case will be added. */
  92 static struct ccase *lag_queue; /* Array of n_lag ccase * elements. */
  93
  94 /* Active transformations. */
  95 struct transformation *t_trns;
  96 size_t n_trns, m_trns, f_trns;
  97
  98 static bool internal_procedure (bool (*proc_func) (struct ccase *, void *),
  99                                 void *aux);
 100 static void update_last_vfm_invocation (void);
 101 static void create_trns_case (struct ccase *, struct dictionary *);
 102 static void open_active_file (void);
 103 static bool write_case (struct write_case_data *wc_data);
 104 static int execute_transformations (struct ccase *c,
 105                                     struct transformation *trns,
 106                                     int first_idx, int last_idx,
 107                                     int case_num);
 108 static int filter_case (const struct ccase *c, int case_num);
 109 static void lag_case (const struct ccase *c);
 110 static void clear_case (struct ccase *c);
 111 static bool close_active_file (void);
 112 \f
 113 /* Public functions. */
 114
 115 /* Returns the last time the data was read. */
 116 time_t
 117 time_of_last_procedure (void)
 118 {
 119   if (last_vfm_invocation == 0)
 120     update_last_vfm_invocation ();
 121   return last_vfm_invocation;
 122 }
 123
 124 /* Reads the data from the input program and writes it to a new
 125    active file.  For each case we read from the input program, we
 126    do the following
 127
 128    1. Execute permanent transformations.  If these drop the case,
 129       start the next case from step 1.
 130
 131    2. N OF CASES.  If we have already written N cases, start the
 132       next case from step 1.
 133
 134    3. Write case to replacement active file.
 135
 136    4. Execute temporary transformations.  If these drop the case,
 137       start the next case from step 1.
 138
 139    5. FILTER, PROCESS IF.  If these drop the case, start the next
 140       case from step 1.
 141
 142    6. Post-TEMPORARY N OF CASES.  If we have already analyzed N
 143       cases, start the next case from step 1.
 144
 145    7. Pass case to PROC_FUNC, passing AUX as auxiliary data.
 146
 147    Returns true if successful, false if an I/O error occurred. */
 148 bool
 149 procedure (bool (*proc_func) (struct ccase *, void *), void *aux)
 150 {
 151   if (proc_func == NULL
 152       && case_source_is_class (vfm_source, &storage_source_class)
 153       && vfm_sink == NULL
 154       && !temporary
 155       && n_trns == 0)
 156     {
 157       /* Nothing to do. */
 158       update_last_vfm_invocation ();
 159       return true;
 160     }
 161   else
 162     {
 163       bool ok;
 164
 165       open_active_file ();
 166       ok = internal_procedure (proc_func, aux);
 167       if (!close_active_file ())
 168         ok = false;
 169
 170       return ok;
 171     }
 172 }
 173
 174 /* Executes a procedure, as procedure(), except that the caller
 175    is responsible for calling open_active_file() and
 176    close_active_file().
 177    Returns true if successful, false if an I/O error occurred. */
 178 static bool
 179 internal_procedure (bool (*proc_func) (struct ccase *, void *), void *aux)
 180 {
 181   static int recursive_call;
 182   struct write_case_data wc_data;
 183   bool ok;
 184
 185   assert (++recursive_call == 1);
 186
 187   wc_data.proc_func = proc_func;
 188   wc_data.aux = aux;
 189   create_trns_case (&wc_data.trns_case, default_dict);
 190   case_create (&wc_data.sink_case, dict_get_next_value_idx (default_dict));
 191   wc_data.cases_written = 0;
 192
 193   update_last_vfm_invocation ();
 194
 195   ok = (vfm_source == NULL
 196         || vfm_source->class->read (vfm_source,
 197                                     &wc_data.trns_case,
 198                                     write_case, &wc_data));
 199
 200   case_destroy (&wc_data.sink_case);
 201   case_destroy (&wc_data.trns_case);
 202
 203   assert (--recursive_call == 0);
 204
 205   return ok;
 206 }
 207
 208 /* Updates last_vfm_invocation. */
 209 static void
 210 update_last_vfm_invocation (void)
 211 {
 212   last_vfm_invocation = time (NULL);
 213 }
 214
 215 /* Creates and returns a case, initializing it from the vectors
 216    that say which `value's need to be initialized just once, and
 217    which ones need to be re-initialized before every case. */
 218 static void
 219 create_trns_case (struct ccase *trns_case, struct dictionary *dict)
 220 {
 221   size_t var_cnt = dict_get_var_cnt (dict);
 222   size_t i;
 223
 224   case_create (trns_case, dict_get_next_value_idx (dict));
 225   for (i = 0; i < var_cnt; i++)
 226     {
 227       struct variable *v = dict_get_var (dict, i);
 228       union value *value = case_data_rw (trns_case, v->fv);
 229
 230       if (v->type == NUMERIC)
 231         value->f = v->reinit ? 0.0 : SYSMIS;
 232       else
 233         memset (value->s, ' ', v->width);
 234     }
 235 }
 236
 237 /* Makes all preparations for reading from the data source and writing
 238    to the data sink. */
 239 static void
 240 open_active_file (void)
 241 {
 242   /* Make temp_dict refer to the dictionary right before data
 243      reaches the sink */
 244   if (!temporary)
 245     {
 246       temp_trns = n_trns;
 247       temp_dict = default_dict;
 248     }
 249
 250   /* Figure out compaction. */
 251   compactor = (dict_needs_compaction (temp_dict)
 252                ? dict_make_compactor (temp_dict)
 253                : NULL);
 254
 255   /* Prepare sink. */
 256   if (vfm_sink == NULL)
 257     vfm_sink = create_case_sink (&storage_sink_class, temp_dict, NULL);
 258   if (vfm_sink->class->open != NULL)
 259     vfm_sink->class->open (vfm_sink);
 260
 261   /* Allocate memory for lag queue. */
 262   if (n_lag > 0)
 263     {
 264       int i;
 265
 266       lag_count = 0;
 267       lag_head = 0;
 268       lag_queue = xnmalloc (n_lag, sizeof *lag_queue);
 269       for (i = 0; i < n_lag; i++)
 270         case_nullify (&lag_queue[i]);
 271     }
 272
 273   /* Close any unclosed DO IF or LOOP constructs. */
 274   ctl_stack_clear ();
 275 }
 276
 277 /* Transforms trns_case and writes it to the replacement active
 278    file if advisable.  Returns true if more cases can be
 279    accepted, false otherwise.  Do not call this function again
 280    after it has returned false once.  */
 281 static bool
 282 write_case (struct write_case_data *wc_data)
 283 {
 284   int retval;
 285
 286   /* Execute permanent transformations.  */
 287   retval = execute_transformations (&wc_data->trns_case, t_trns, f_trns,
 288                                     temp_trns, wc_data->cases_written + 1);
 289   if (retval != 1)
 290     goto done;
 291
 292   /* N OF CASES. */
 293   if (dict_get_case_limit (default_dict)
 294       && wc_data->cases_written >= dict_get_case_limit (default_dict))
 295     goto done;
 296   wc_data->cases_written++;
 297
 298   /* Write case to LAG queue. */
 299   if (n_lag)
 300     lag_case (&wc_data->trns_case);
 301
 302   /* Write case to replacement active file. */
 303   if (vfm_sink->class->write != NULL)
 304     {
 305       if (compactor != NULL)
 306         {
 307           dict_compactor_compact (compactor, &wc_data->sink_case,
 308                                   &wc_data->trns_case);
 309           vfm_sink->class->write (vfm_sink, &wc_data->sink_case);
 310         }
 311       else
 312         vfm_sink->class->write (vfm_sink, &wc_data->trns_case);
 313     }
 314
 315   /* Execute temporary transformations. */
 316   retval = execute_transformations (&wc_data->trns_case, t_trns, temp_trns,
 317                                     n_trns, wc_data->cases_written);
 318   if (retval != 1)
 319     goto done;
 320
 321   /* FILTER, PROCESS IF, post-TEMPORARY N OF CASES. */
 322   if (filter_case (&wc_data->trns_case, wc_data->cases_written)
 323       || (dict_get_case_limit (temp_dict)
 324           && wc_data->cases_analyzed >= dict_get_case_limit (temp_dict)))
 325     goto done;
 326   wc_data->cases_analyzed++;
 327
 328   /* Pass case to procedure. */
 329   if (wc_data->proc_func != NULL)
 330     if (!wc_data->proc_func (&wc_data->trns_case, wc_data->aux))
 331       retval = -1;
 332
 333  done:
 334   clear_case (&wc_data->trns_case);
 335   return retval != -1;
 336 }
 337
 338 /* Transforms case C using the transformations in TRNS[] with
 339    indexes FIRST_IDX through LAST_IDX, exclusive.  Case C will
 340    become case CASE_NUM (1-based) in the output file.  Returns 1
 341    if the case was successfully transformed, 0 if it was filtered
 342    out by one of the transformations, or -1 if the procedure
 343    should be abandoned due to a fatal error. */
 344 static int
 345 execute_transformations (struct ccase *c,
 346                          struct transformation *trns,
 347                          int first_idx, int last_idx,
 348                          int case_num)
 349 {
 350   int idx;
 351
 352   for (idx = first_idx; idx != last_idx; )
 353     {
 354       struct transformation *t = &trns[idx];
 355       int retval = t->proc (t->private, c, case_num);
 356       switch (retval)
 357         {
 358         case TRNS_CONTINUE:
 359           idx++;
 360           break;
 361
 362         case TRNS_DROP_CASE:
 363           return 0;
 364
 365         case TRNS_ERROR:
 366           return -1;
 367
 368         case TRNS_NEXT_CASE:
 369           abort ();
 370
 371         case TRNS_END_FILE:
 372           abort ();
 373
 374         default:
 375           idx = retval;
 376           break;
 377         }
 378     }
 379
 380   return 1;
 381 }
 382
 383 /* Returns nonzero if case C with case number CASE_NUM should be
 384    excluded as specified on FILTER or PROCESS IF, otherwise
 385    zero. */
 386 static int
 387 filter_case (const struct ccase *c, int case_idx)
 388 {
 389   /* FILTER. */
 390   struct variable *filter_var = dict_get_filter (default_dict);
 391   if (filter_var != NULL)
 392     {
 393       double f = case_num (c, filter_var->fv);
 394       if (f == 0.0 || mv_is_num_missing (&filter_var->miss, f))
 395         return 1;
 396     }
 397
 398   /* PROCESS IF. */
 399   if (process_if_expr != NULL
 400       && expr_evaluate_num (process_if_expr, c, case_idx) != 1.0)
 401     return 1;
 402
 403   return 0;
 404 }
 405
 406 /* Add C to the lag queue. */
 407 static void
 408 lag_case (const struct ccase *c)
 409 {
 410   if (lag_count < n_lag)
 411     lag_count++;
 412   case_destroy (&lag_queue[lag_head]);
 413   case_clone (&lag_queue[lag_head], c);
 414   if (++lag_head >= n_lag)
 415     lag_head = 0;
 416 }
 417
 418 /* Clears the variables in C that need to be cleared between
 419    processing cases.  */
 420 static void
 421 clear_case (struct ccase *c)
 422 {
 423   size_t var_cnt = dict_get_var_cnt (default_dict);
 424   size_t i;
 425
 426   for (i = 0; i < var_cnt; i++)
 427     {
 428       struct variable *v = dict_get_var (default_dict, i);
 429       if (v->reinit)
 430         {
 431           if (v->type == NUMERIC)
 432             case_data_rw (c, v->fv)->f = SYSMIS;
 433           else
 434             memset (case_data_rw (c, v->fv)->s, ' ', v->width);
 435         }
 436     }
 437 }
 438
 439 /* Closes the active file. */
 440 static bool
 441 close_active_file (void)
 442 {
 443   /* Free memory for lag queue, and turn off lagging. */
 444   if (n_lag > 0)
 445     {
 446       int i;
 447
 448       for (i = 0; i < n_lag; i++)
 449         case_destroy (&lag_queue[i]);
 450       free (lag_queue);
 451       n_lag = 0;
 452     }
 453
 454   /* Dictionary from before TEMPORARY becomes permanent.. */
 455   if (temporary)
 456     {
 457       dict_destroy (default_dict);
 458       default_dict = temp_dict;
 459       temp_dict = NULL;
 460     }
 461
 462   /* Finish compaction. */
 463   if (compactor != NULL)
 464     {
 465       dict_compactor_destroy (compactor);
 466       dict_compact_values (default_dict);
 467     }
 468
 469   /* Free data source. */
 470   free_case_source (vfm_source);
 471   vfm_source = NULL;
 472
 473   /* Old data sink becomes new data source. */
 474   if (vfm_sink->class->make_source != NULL)
 475     vfm_source = vfm_sink->class->make_source (vfm_sink);
 476   free_case_sink (vfm_sink);
 477   vfm_sink = NULL;
 478
 479   /* Cancel TEMPORARY, PROCESS IF, FILTER, N OF CASES, vectors,
 480      and get rid of all the transformations. */
 481   cancel_temporary ();
 482   expr_free (process_if_expr);
 483   process_if_expr = NULL;
 484   dict_set_case_limit (default_dict, 0);
 485   dict_clear_vectors (default_dict);
 486   return cancel_transformations ();
 487 }
 488 \f
 489 /* Returns a pointer to the lagged case from N_BEFORE cases before the
 490    current one, or NULL if there haven't been that many cases yet. */
 491 struct ccase *
 492 lagged_case (int n_before)
 493 {
 494   assert (n_before >= 1 );
 495   assert (n_before <= n_lag);
 496
 497   if (n_before <= lag_count)
 498     {
 499       int index = lag_head - n_before;
 500       if (index < 0)
 501         index += n_lag;
 502       return &lag_queue[index];
 503     }
 504   else
 505     return NULL;
 506 }
 507
 508 /* Appends TRNS to t_trns[], the list of all transformations to be
 509    performed on data as it is read from the active file. */
 510 void
 511 add_transformation (trns_proc_func *proc, trns_free_func *free, void *private)
 512 {
 513   struct transformation *trns;
 514   if (n_trns >= m_trns)
 515     t_trns = x2nrealloc (t_trns, &m_trns, sizeof *t_trns);
 516   trns = &t_trns[n_trns++];
 517   trns->proc = proc;
 518   trns->free = free;
 519   trns->private = private;
 520 }
 521
 522 /* Returns the index number that the next transformation added by
 523    add_transformation() will receive.  A trns_proc_func that
 524    returns this index causes control flow to jump to it. */
 525 size_t
 526 next_transformation (void)
 527 {
 528   return n_trns;
 529 }
 530
 531 /* Cancels all active transformations, including any transformations
 532    created by the input program.
 533    Returns true if successful, false if an I/O error occurred. */
 534 bool
 535 cancel_transformations (void)
 536 {
 537   bool ok = true;
 538   size_t i;
 539   for (i = 0; i < n_trns; i++)
 540     {
 541       struct transformation *t = &t_trns[i];
 542       if (t->free != NULL)
 543         {
 544           if (!t->free (t->private))
 545             ok = false;
 546         }
 547     }
 548   n_trns = f_trns = 0;
 549   free (t_trns);
 550   t_trns = NULL;
 551   m_trns = 0;
 552   return ok;
 553 }
 554 \f
 555 /* Represents auxiliary data for handling SPLIT FILE. */
 556 struct split_aux_data
 557   {
 558     size_t case_count;          /* Number of cases so far. */
 559     struct ccase prev_case;     /* Data in previous case. */
 560
 561     /* Functions to call... */
 562     void (*begin_func) (void *);               /* ...before data. */
 563     bool (*proc_func) (struct ccase *, void *); /* ...with data. */
 564     void (*end_func) (void *);                 /* ...after data. */
 565     void *func_aux;                            /* Auxiliary data. */
 566   };
 567
 568 static int equal_splits (const struct ccase *, const struct ccase *);
 569 static bool procedure_with_splits_callback (struct ccase *, void *);
 570 static void dump_splits (struct ccase *);
 571
 572 /* Like procedure(), but it automatically breaks the case stream
 573    into SPLIT FILE break groups.  Before each group of cases with
 574    identical SPLIT FILE variable values, BEGIN_FUNC is called.
 575    Then PROC_FUNC is called with each case in the group.
 576    END_FUNC is called when the group is finished.  FUNC_AUX is
 577    passed to each of the functions as auxiliary data.
 578
 579    If the active file is empty, none of BEGIN_FUNC, PROC_FUNC,
 580    and END_FUNC will be called at all.
 581
 582    If SPLIT FILE is not in effect, then there is one break group
 583    (if the active file is nonempty), and BEGIN_FUNC and END_FUNC
 584    will be called once.
 585
 586    Returns true if successful, false if an I/O error occurred. */
 587 bool
 588 procedure_with_splits (void (*begin_func) (void *aux),
 589                        bool (*proc_func) (struct ccase *, void *aux),
 590                        void (*end_func) (void *aux),
 591                        void *func_aux)
 592 {
 593   struct split_aux_data split_aux;
 594   bool ok;
 595
 596   split_aux.case_count = 0;
 597   case_nullify (&split_aux.prev_case);
 598   split_aux.begin_func = begin_func;
 599   split_aux.proc_func = proc_func;
 600   split_aux.end_func = end_func;
 601   split_aux.func_aux = func_aux;
 602
 603   open_active_file ();
 604   ok = internal_procedure (procedure_with_splits_callback, &split_aux);
 605   if (split_aux.case_count > 0 && end_func != NULL)
 606     end_func (func_aux);
 607   if (!close_active_file ())
 608     ok = false;
 609
 610   case_destroy (&split_aux.prev_case);
 611
 612   return ok;
 613 }
 614
 615 /* procedure() callback used by procedure_with_splits(). */
 616 static bool
 617 procedure_with_splits_callback (struct ccase *c, void *split_aux_)
 618 {
 619   struct split_aux_data *split_aux = split_aux_;
 620
 621   /* Start a new series if needed. */
 622   if (split_aux->case_count == 0
 623       || !equal_splits (c, &split_aux->prev_case))
 624     {
 625       if (split_aux->case_count > 0 && split_aux->end_func != NULL)
 626         split_aux->end_func (split_aux->func_aux);
 627
 628       dump_splits (c);
 629       case_destroy (&split_aux->prev_case);
 630       case_clone (&split_aux->prev_case, c);
 631
 632       if (split_aux->begin_func != NULL)
 633         split_aux->begin_func (split_aux->func_aux);
 634     }
 635
 636   split_aux->case_count++;
 637   if (split_aux->proc_func != NULL)
 638     return split_aux->proc_func (c, split_aux->func_aux);
 639   else
 640     return true;
 641 }
 642
 643 /* Compares the SPLIT FILE variables in cases A and B and returns
 644    nonzero only if they differ. */
 645 static int
 646 equal_splits (const struct ccase *a, const struct ccase *b)
 647 {
 648   return case_compare (a, b,
 649                        dict_get_split_vars (default_dict),
 650                        dict_get_split_cnt (default_dict)) == 0;
 651 }
 652
 653 /* Dumps out the values of all the split variables for the case C. */
 654 static void
 655 dump_splits (struct ccase *c)
 656 {
 657   struct variable *const *split;
 658   struct tab_table *t;
 659   size_t split_cnt;
 660   int i;
 661
 662   split_cnt = dict_get_split_cnt (default_dict);
 663   if (split_cnt == 0)
 664     return;
 665
 666   t = tab_create (3, split_cnt + 1, 0);
 667   tab_dim (t, tab_natural_dimensions);
 668   tab_vline (t, TAL_GAP, 1, 0, split_cnt);
 669   tab_vline (t, TAL_GAP, 2, 0, split_cnt);
 670   tab_text (t, 0, 0, TAB_NONE, _("Variable"));
 671   tab_text (t, 1, 0, TAB_LEFT, _("Value"));
 672   tab_text (t, 2, 0, TAB_LEFT, _("Label"));
 673   split = dict_get_split_vars (default_dict);
 674   for (i = 0; i < split_cnt; i++)
 675     {
 676       struct variable *v = split[i];
 677       char temp_buf[80];
 678       const char *val_lab;
 679
 680       assert (v->type == NUMERIC || v->type == ALPHA);
 681       tab_text (t, 0, i + 1, TAB_LEFT | TAT_PRINTF, "%s", v->name);
 682
 683       data_out (temp_buf, &v->print, case_data (c, v->fv));
 684
 685       temp_buf[v->print.w] = 0;
 686       tab_text (t, 1, i + 1, TAT_PRINTF, "%.*s", v->print.w, temp_buf);
 687
 688       val_lab = val_labs_find (v->val_labs, *case_data (c, v->fv));
 689       if (val_lab)
 690         tab_text (t, 2, i + 1, TAB_LEFT, val_lab);
 691     }
 692   tab_flags (t, SOMF_NO_TITLE);
 693   tab_submit (t);
 694 }
 695 \f
 696 /* Represents auxiliary data for handling SPLIT FILE in a
 697    multipass procedure. */
 698 struct multipass_split_aux_data
 699   {
 700     struct ccase prev_case;     /* Data in previous case. */
 701     struct casefile *casefile;  /* Accumulates data for a split. */
 702
 703     /* Function to call with the accumulated data. */
 704     bool (*split_func) (const struct casefile *, void *);
 705     void *func_aux;                            /* Auxiliary data. */
 706   };
 707
 708 static bool multipass_split_callback (struct ccase *c, void *aux_);
 709 static void multipass_split_output (struct multipass_split_aux_data *);
 710
 711 /* Returns true if successful, false if an I/O error occurred. */
 712 bool
 713 multipass_procedure_with_splits (bool (*split_func) (const struct casefile *,
 714                                                      void *),
 715                                  void *func_aux)
 716 {
 717   struct multipass_split_aux_data aux;
 718   bool ok;
 719
 720   assert (split_func != NULL);
 721
 722   open_active_file ();
 723
 724   case_nullify (&aux.prev_case);
 725   aux.casefile = NULL;
 726   aux.split_func = split_func;
 727   aux.func_aux = func_aux;
 728
 729   ok = internal_procedure (multipass_split_callback, &aux);
 730   if (aux.casefile != NULL)
 731     multipass_split_output (&aux);
 732   case_destroy (&aux.prev_case);
 733
 734   if (!close_active_file ())
 735     ok = false;
 736
 737   return ok;
 738 }
 739
 740 /* procedure() callback used by multipass_procedure_with_splits(). */
 741 static bool
 742 multipass_split_callback (struct ccase *c, void *aux_)
 743 {
 744   struct multipass_split_aux_data *aux = aux_;
 745
 746   /* Start a new series if needed. */
 747   if (aux->casefile == NULL || !equal_splits (c, &aux->prev_case))
 748     {
 749       /* Pass any cases to split_func. */
 750       if (aux->casefile != NULL)
 751         multipass_split_output (aux);
 752
 753       /* Start a new casefile. */
 754       aux->casefile = casefile_create (dict_get_next_value_idx (default_dict));
 755
 756       /* Record split values. */
 757       dump_splits (c);
 758       case_destroy (&aux->prev_case);
 759       case_clone (&aux->prev_case, c);
 760     }
 761
 762   return casefile_append (aux->casefile, c);
 763 }
 764
 765 static void
 766 multipass_split_output (struct multipass_split_aux_data *aux)
 767 {
 768   assert (aux->casefile != NULL);
 769   aux->split_func (aux->casefile, aux->func_aux);
 770   casefile_destroy (aux->casefile);
 771   aux->casefile = NULL;
 772 }
 773
 774
 775 /* Discards all the current state in preparation for a data-input
 776    command like DATA LIST or GET. */
 777 void
 778 discard_variables (void)
 779 {
 780   dict_clear (default_dict);
 781   fh_set_default_handle (NULL);
 782
 783   n_lag = 0;
 784
 785   if (vfm_source != NULL)
 786     {
 787       free_case_source (vfm_source);
 788       vfm_source = NULL;
 789     }
 790
 791   cancel_transformations ();
 792
 793   ctl_stack_clear ();
 794
 795   expr_free (process_if_expr);
 796   process_if_expr = NULL;
 797
 798   cancel_temporary ();
 799 }