src/data/dataset.c

   1 /* PSPP - a program for statistical analysis.
   2    Copyright (C) 1997-9, 2000, 2006, 2007, 2009, 2010, 2011 Free Software Foundation, Inc.
   3
   4    This program is free software: you can redistribute it and/or modify
   5    it under the terms of the GNU General Public License as published by
   6    the Free Software Foundation, either version 3 of the License, or
   7    (at your option) any later version.
   8
   9    This program is distributed in the hope that it will be useful,
  10    but WITHOUT ANY WARRANTY; without even the implied warranty of
  11    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12    GNU General Public License for more details.
  13
  14    You should have received a copy of the GNU General Public License
  15    along with this program.  If not, see <http://www.gnu.org/licenses/>. */
  16
  17 #include <config.h>
  18
  19 #include "data/dataset.h"
  20
  21 #include <errno.h>
  22 #include <stdio.h>
  23 #include <stdlib.h>
  24 #include <unistd.h>
  25
  26 #include "data/case.h"
  27 #include "data/case-map.h"
  28 #include "data/caseinit.h"
  29 #include "data/casereader.h"
  30 #include "data/casereader-provider.h"
  31 #include "data/casereader-shim.h"
  32 #include "data/casewriter.h"
  33 #include "data/dictionary.h"
  34 #include "data/file-handle-def.h"
  35 #include "data/transformations.h"
  36 #include "data/variable.h"
  37 #include "libpspp/deque.h"
  38 #include "libpspp/misc.h"
  39 #include "libpspp/str.h"
  40 #include "libpspp/taint.h"
  41 #include "libpspp/i18n.h"
  42
  43 #include "gl/minmax.h"
  44 #include "gl/xalloc.h"
  45
  46 struct dataset {
  47   /* Cases are read from source,
  48      their transformation variables are initialized,
  49      pass through permanent_trns_chain (which transforms them into
  50      the format described by permanent_dict),
  51      are written to sink,
  52      pass through temporary_trns_chain (which transforms them into
  53      the format described by dict),
  54      and are finally passed to the procedure. */
  55   struct casereader *source;
  56   struct caseinit *caseinit;
  57   struct trns_chain *permanent_trns_chain;
  58   struct dictionary *permanent_dict;
  59   struct casewriter *sink;
  60   struct trns_chain *temporary_trns_chain;
  61   struct dictionary *dict;
  62
  63   /* Callback which occurs whenever the transformation chain(s) have
  64      been modified */
  65   transformation_change_callback_func *xform_callback;
  66   void *xform_callback_aux;
  67
  68   /* If true, cases are discarded instead of being written to
  69      sink. */
  70   bool discard_output;
  71
  72   /* The transformation chain that the next transformation will be
  73      added to. */
  74   struct trns_chain *cur_trns_chain;
  75
  76   /* The case map used to compact a case, if necessary;
  77      otherwise a null pointer. */
  78   struct case_map *compactor;
  79
  80   /* Time at which proc was last invoked. */
  81   time_t last_proc_invocation;
  82
  83   /* Cases just before ("lagging") the current one. */
  84   int n_lag;                    /* Number of cases to lag. */
  85   struct deque lag;             /* Deque of lagged cases. */
  86   struct ccase **lag_cases;     /* Lagged cases managed by deque. */
  87
  88   /* Procedure data. */
  89   enum
  90     {
  91       PROC_COMMITTED,           /* No procedure in progress. */
  92       PROC_OPEN,                /* proc_open called, casereader still open. */
  93       PROC_CLOSED               /* casereader from proc_open destroyed,
  94                                    but proc_commit not yet called. */
  95     }
  96   proc_state;
  97   casenumber cases_written;     /* Cases output so far. */
  98   bool ok;                      /* Error status. */
  99   struct casereader_shim *shim; /* Shim on proc_open() casereader. */
 100
 101   void (*callback) (void *); /* Callback for when the dataset changes */
 102   void *cb_data;
 103
 104   /* Default encoding for reading syntax files. */
 105   char *syntax_encoding;
 106 };
 107
 108 static void add_case_limit_trns (struct dataset *ds);
 109 static void add_filter_trns (struct dataset *ds);
 110
 111 static void update_last_proc_invocation (struct dataset *ds);
 112
 113 static void
 114 dataset_set_unsaved (const struct dataset *ds)
 115 {
 116   if (ds->callback) ds->callback (ds->cb_data);
 117 }
 118
 119 static void
 120 dict_callback (struct dictionary *d UNUSED, void *ds_)
 121 {
 122   struct dataset *ds = ds_;
 123   dataset_set_unsaved (ds);
 124 }
 125 \f
 126 /* Creates and returns a new dataset.  The dataset initially has an empty
 127    dictionary and no data source. */
 128 struct dataset *
 129 dataset_create (void)
 130 {
 131   struct dataset *ds;
 132
 133   ds = xzalloc (sizeof *ds);
 134   ds->dict = dict_create ();
 135   dict_set_change_callback (ds->dict, dict_callback, ds);
 136   dict_set_encoding (ds->dict, get_default_encoding ());
 137
 138   ds->caseinit = caseinit_create ();
 139   proc_cancel_all_transformations (ds);
 140   ds->syntax_encoding = xstrdup ("Auto");
 141   return ds;
 142 }
 143
 144 /* Destroys DS. */
 145 void
 146 dataset_destroy (struct dataset *ds)
 147 {
 148   if (ds != NULL)
 149     {
 150       dataset_clear (ds);
 151       dict_destroy (ds->dict);
 152       caseinit_destroy (ds->caseinit);
 153       trns_chain_destroy (ds->permanent_trns_chain);
 154
 155       if ( ds->xform_callback)
 156         ds->xform_callback (false, ds->xform_callback_aux);
 157       free (ds->syntax_encoding);
 158       free (ds);
 159     }
 160 }
 161
 162 /* Discards the active file dictionary, data, and transformations. */
 163 void
 164 dataset_clear (struct dataset *ds)
 165 {
 166   assert (ds->proc_state == PROC_COMMITTED);
 167
 168   dict_clear (ds->dict);
 169   fh_set_default_handle (NULL);
 170
 171   ds->n_lag = 0;
 172
 173   casereader_destroy (ds->source);
 174   ds->source = NULL;
 175
 176   proc_cancel_all_transformations (ds);
 177 }
 178
 179 /* Returns the dictionary within DS.  This is always nonnull, although it
 180    might not contain any variables. */
 181 struct dictionary *
 182 dataset_dict (const struct dataset *ds)
 183 {
 184   return ds->dict;
 185 }
 186
 187 /* Replaces DS's dictionary by DICT, discarding any source and
 188    transformations. */
 189 void
 190 dataset_set_dict (struct dataset *ds, struct dictionary *dict)
 191 {
 192   assert (ds->proc_state == PROC_COMMITTED);
 193   assert (ds->dict != dict);
 194
 195   dataset_clear (ds);
 196
 197   dict_destroy (ds->dict);
 198   ds->dict = dict;
 199   dict_set_change_callback (ds->dict, dict_callback, ds);
 200 }
 201
 202 /* Returns the casereader that will be read when a procedure is executed on
 203    DS.  This can be NULL if none has been set up yet. */
 204 const struct casereader *
 205 dataset_source (const struct dataset *ds)
 206 {
 207   return ds->source;
 208 }
 209
 210 /* Returns true if DS has a data source, false otherwise. */
 211 bool
 212 dataset_has_source (const struct dataset *ds)
 213 {
 214   return dataset_source (ds) != NULL;
 215 }
 216
 217 /* Replaces the active file's data by READER.  READER's cases must have an
 218    appropriate format for DS's dictionary. */
 219 bool
 220 dataset_set_source (struct dataset *ds, struct casereader *reader)
 221 {
 222   casereader_destroy (ds->source);
 223   ds->source = reader;
 224
 225   caseinit_clear (ds->caseinit);
 226   caseinit_mark_as_preinited (ds->caseinit, ds->dict);
 227
 228   return reader == NULL || !casereader_error (reader);
 229 }
 230
 231 /* Returns the data source from DS and removes it from DS.  Returns a null
 232    pointer if DS has no data source. */
 233 struct casereader *
 234 dataset_steal_source (struct dataset *ds)
 235 {
 236   struct casereader *reader = ds->source;
 237   ds->source = NULL;
 238
 239   return reader;
 240 }
 241 \f
 242 void
 243 dataset_set_callback (struct dataset *ds, void (*cb) (void *), void *cb_data)
 244 {
 245   ds->callback = cb;
 246   ds->cb_data = cb_data;
 247 }
 248
 249 void
 250 dataset_set_default_syntax_encoding (struct dataset *ds, const char *encoding)
 251 {
 252   free (ds->syntax_encoding);
 253   ds->syntax_encoding = xstrdup (encoding);
 254 }
 255
 256 const char *
 257 dataset_get_default_syntax_encoding (const struct dataset *ds)
 258 {
 259   return ds->syntax_encoding;
 260 }
 261
 262 /* Returns the last time the data was read. */
 263 time_t
 264 time_of_last_procedure (struct dataset *ds)
 265 {
 266   if (ds->last_proc_invocation == 0)
 267     update_last_proc_invocation (ds);
 268   return ds->last_proc_invocation;
 269 }
 270 \f
 271 /* Regular procedure. */
 272
 273 /* Executes any pending transformations, if necessary.
 274    This is not identical to the EXECUTE command in that it won't
 275    always read the source data.  This can be important when the
 276    source data is given inline within BEGIN DATA...END FILE. */
 277 bool
 278 proc_execute (struct dataset *ds)
 279 {
 280   bool ok;
 281
 282   if ((ds->temporary_trns_chain == NULL
 283        || trns_chain_is_empty (ds->temporary_trns_chain))
 284       && trns_chain_is_empty (ds->permanent_trns_chain))
 285     {
 286       ds->n_lag = 0;
 287       ds->discard_output = false;
 288       dict_set_case_limit (ds->dict, 0);
 289       dict_clear_vectors (ds->dict);
 290       return true;
 291     }
 292
 293   ok = casereader_destroy (proc_open (ds));
 294   return proc_commit (ds) && ok;
 295 }
 296
 297 static const struct casereader_class proc_casereader_class;
 298
 299 /* Opens dataset DS for reading cases with proc_read.  If FILTER is true, then
 300    cases filtered out with FILTER BY will not be included in the casereader
 301    (which is usually desirable).  If FILTER is false, all cases will be
 302    included regardless of FILTER BY settings.
 303
 304    proc_commit must be called when done. */
 305 struct casereader *
 306 proc_open_filtering (struct dataset *ds, bool filter)
 307 {
 308   struct casereader *reader;
 309
 310   assert (ds->source != NULL);
 311   assert (ds->proc_state == PROC_COMMITTED);
 312
 313   update_last_proc_invocation (ds);
 314
 315   caseinit_mark_for_init (ds->caseinit, ds->dict);
 316
 317   /* Finish up the collection of transformations. */
 318   add_case_limit_trns (ds);
 319   if (filter)
 320     add_filter_trns (ds);
 321   trns_chain_finalize (ds->cur_trns_chain);
 322
 323   /* Make permanent_dict refer to the dictionary right before
 324      data reaches the sink. */
 325   if (ds->permanent_dict == NULL)
 326     ds->permanent_dict = ds->dict;
 327
 328   /* Prepare sink. */
 329   if (!ds->discard_output)
 330     {
 331       struct dictionary *pd = ds->permanent_dict;
 332       size_t compacted_value_cnt = dict_count_values (pd, 1u << DC_SCRATCH);
 333       if (compacted_value_cnt < dict_get_next_value_idx (pd))
 334         {
 335           struct caseproto *compacted_proto;
 336           compacted_proto = dict_get_compacted_proto (pd, 1u << DC_SCRATCH);
 337           ds->compactor = case_map_to_compact_dict (pd, 1u << DC_SCRATCH);
 338           ds->sink = autopaging_writer_create (compacted_proto);
 339           caseproto_unref (compacted_proto);
 340         }
 341       else
 342         {
 343           ds->compactor = NULL;
 344           ds->sink = autopaging_writer_create (dict_get_proto (pd));
 345         }
 346     }
 347   else
 348     {
 349       ds->compactor = NULL;
 350       ds->sink = NULL;
 351     }
 352
 353   /* Allocate memory for lagged cases. */
 354   ds->lag_cases = deque_init (&ds->lag, ds->n_lag, sizeof *ds->lag_cases);
 355
 356   ds->proc_state = PROC_OPEN;
 357   ds->cases_written = 0;
 358   ds->ok = true;
 359
 360   /* FIXME: use taint in dataset in place of `ok'? */
 361   /* FIXME: for trivial cases we can just return a clone of
 362      ds->source? */
 363
 364   /* Create casereader and insert a shim on top.  The shim allows us to
 365      arbitrarily extend the casereader's lifetime, by slurping the cases into
 366      the shim's buffer in proc_commit().  That is especially useful when output
 367      table_items are generated directly from the procedure casereader (e.g. by
 368      the LIST procedure) when we are using an output driver that keeps a
 369      reference to the output items passed to it (e.g. the GUI output driver in
 370      PSPPIRE). */
 371   reader = casereader_create_sequential (NULL, dict_get_proto (ds->dict),
 372                                          CASENUMBER_MAX,
 373                                          &proc_casereader_class, ds);
 374   ds->shim = casereader_shim_insert (reader);
 375   return reader;
 376 }
 377
 378 /* Opens dataset DS for reading cases with proc_read.
 379    proc_commit must be called when done. */
 380 struct casereader *
 381 proc_open (struct dataset *ds)
 382 {
 383   return proc_open_filtering (ds, true);
 384 }
 385
 386 /* Returns true if a procedure is in progress, that is, if
 387    proc_open has been called but proc_commit has not. */
 388 bool
 389 proc_is_open (const struct dataset *ds)
 390 {
 391   return ds->proc_state != PROC_COMMITTED;
 392 }
 393
 394 /* "read" function for procedure casereader. */
 395 static struct ccase *
 396 proc_casereader_read (struct casereader *reader UNUSED, void *ds_)
 397 {
 398   struct dataset *ds = ds_;
 399   enum trns_result retval = TRNS_DROP_CASE;
 400   struct ccase *c;
 401
 402   assert (ds->proc_state == PROC_OPEN);
 403   for (; ; case_unref (c))
 404     {
 405       casenumber case_nr;
 406
 407       assert (retval == TRNS_DROP_CASE || retval == TRNS_ERROR);
 408       if (retval == TRNS_ERROR)
 409         ds->ok = false;
 410       if (!ds->ok)
 411         return NULL;
 412
 413       /* Read a case from source. */
 414       c = casereader_read (ds->source);
 415       if (c == NULL)
 416         return NULL;
 417       c = case_unshare_and_resize (c, dict_get_proto (ds->dict));
 418       caseinit_init_vars (ds->caseinit, c);
 419
 420       /* Execute permanent transformations.  */
 421       case_nr = ds->cases_written + 1;
 422       retval = trns_chain_execute (ds->permanent_trns_chain, TRNS_CONTINUE,
 423                                    &c, case_nr);
 424       caseinit_update_left_vars (ds->caseinit, c);
 425       if (retval != TRNS_CONTINUE)
 426         continue;
 427
 428       /* Write case to collection of lagged cases. */
 429       if (ds->n_lag > 0)
 430         {
 431           while (deque_count (&ds->lag) >= ds->n_lag)
 432             case_unref (ds->lag_cases[deque_pop_back (&ds->lag)]);
 433           ds->lag_cases[deque_push_front (&ds->lag)] = case_ref (c);
 434         }
 435
 436       /* Write case to replacement active file. */
 437       ds->cases_written++;
 438       if (ds->sink != NULL)
 439         casewriter_write (ds->sink,
 440                           case_map_execute (ds->compactor, case_ref (c)));
 441
 442       /* Execute temporary transformations. */
 443       if (ds->temporary_trns_chain != NULL)
 444         {
 445           retval = trns_chain_execute (ds->temporary_trns_chain, TRNS_CONTINUE,
 446                                        &c, ds->cases_written);
 447           if (retval != TRNS_CONTINUE)
 448             continue;
 449         }
 450
 451       return c;
 452     }
 453 }
 454
 455 /* "destroy" function for procedure casereader. */
 456 static void
 457 proc_casereader_destroy (struct casereader *reader, void *ds_)
 458 {
 459   struct dataset *ds = ds_;
 460   struct ccase *c;
 461
 462   /* We are always the subreader for a casereader_buffer, so if we're being
 463      destroyed then it's because the casereader_buffer has read all the cases
 464      that it ever will. */
 465   ds->shim = NULL;
 466
 467   /* Make sure transformations happen for every input case, in
 468      case they have side effects, and ensure that the replacement
 469      active file gets all the cases it should. */
 470   while ((c = casereader_read (reader)) != NULL)
 471     case_unref (c);
 472
 473   ds->proc_state = PROC_CLOSED;
 474   ds->ok = casereader_destroy (ds->source) && ds->ok;
 475   ds->source = NULL;
 476   dataset_set_source (ds, NULL);
 477 }
 478
 479 /* Must return false if the source casereader, a transformation,
 480    or the sink casewriter signaled an error.  (If a temporary
 481    transformation signals an error, then the return value is
 482    false, but the replacement active file may still be
 483    untainted.) */
 484 bool
 485 proc_commit (struct dataset *ds)
 486 {
 487   if (ds->shim != NULL)
 488     casereader_shim_slurp (ds->shim);
 489
 490   assert (ds->proc_state == PROC_CLOSED);
 491   ds->proc_state = PROC_COMMITTED;
 492
 493   dataset_set_unsaved (ds);
 494
 495   /* Free memory for lagged cases. */
 496   while (!deque_is_empty (&ds->lag))
 497     case_unref (ds->lag_cases[deque_pop_back (&ds->lag)]);
 498   free (ds->lag_cases);
 499
 500   /* Dictionary from before TEMPORARY becomes permanent. */
 501   proc_cancel_temporary_transformations (ds);
 502
 503   if (!ds->discard_output)
 504     {
 505       /* Finish compacting. */
 506       if (ds->compactor != NULL)
 507         {
 508           case_map_destroy (ds->compactor);
 509           ds->compactor = NULL;
 510
 511           dict_delete_scratch_vars (ds->dict);
 512           dict_compact_values (ds->dict);
 513         }
 514
 515       /* Old data sink becomes new data source. */
 516       if (ds->sink != NULL)
 517         ds->source = casewriter_make_reader (ds->sink);
 518     }
 519   else
 520     {
 521       ds->source = NULL;
 522       ds->discard_output = false;
 523     }
 524   ds->sink = NULL;
 525
 526   caseinit_clear (ds->caseinit);
 527   caseinit_mark_as_preinited (ds->caseinit, ds->dict);
 528
 529   dict_clear_vectors (ds->dict);
 530   ds->permanent_dict = NULL;
 531   return proc_cancel_all_transformations (ds) && ds->ok;
 532 }
 533
 534 /* Casereader class for procedure execution. */
 535 static const struct casereader_class proc_casereader_class =
 536   {
 537     proc_casereader_read,
 538     proc_casereader_destroy,
 539     NULL,
 540     NULL,
 541   };
 542
 543 /* Updates last_proc_invocation. */
 544 static void
 545 update_last_proc_invocation (struct dataset *ds)
 546 {
 547   ds->last_proc_invocation = time (NULL);
 548 }
 549 \f
 550 /* Returns a pointer to the lagged case from N_BEFORE cases before the
 551    current one, or NULL if there haven't been that many cases yet. */
 552 const struct ccase *
 553 lagged_case (const struct dataset *ds, int n_before)
 554 {
 555   assert (n_before >= 1);
 556   assert (n_before <= ds->n_lag);
 557
 558   if (n_before <= deque_count (&ds->lag))
 559     return ds->lag_cases[deque_front (&ds->lag, n_before - 1)];
 560   else
 561     return NULL;
 562 }
 563 \f
 564 /* Returns the current set of permanent transformations,
 565    and clears the permanent transformations.
 566    For use by INPUT PROGRAM. */
 567 struct trns_chain *
 568 proc_capture_transformations (struct dataset *ds)
 569 {
 570   struct trns_chain *chain;
 571
 572   assert (ds->temporary_trns_chain == NULL);
 573   chain = ds->permanent_trns_chain;
 574   ds->cur_trns_chain = ds->permanent_trns_chain = trns_chain_create ();
 575
 576   if ( ds->xform_callback)
 577     ds->xform_callback (false, ds->xform_callback_aux);
 578
 579   return chain;
 580 }
 581
 582 /* Adds a transformation that processes a case with PROC and
 583    frees itself with FREE to the current set of transformations.
 584    The functions are passed AUX as auxiliary data. */
 585 void
 586 add_transformation (struct dataset *ds, trns_proc_func *proc, trns_free_func *free, void *aux)
 587 {
 588   trns_chain_append (ds->cur_trns_chain, NULL, proc, free, aux);
 589   if ( ds->xform_callback)
 590     ds->xform_callback (true, ds->xform_callback_aux);
 591 }
 592
 593 /* Adds a transformation that processes a case with PROC and
 594    frees itself with FREE to the current set of transformations.
 595    When parsing of the block of transformations is complete,
 596    FINALIZE will be called.
 597    The functions are passed AUX as auxiliary data. */
 598 void
 599 add_transformation_with_finalizer (struct dataset *ds,
 600                                    trns_finalize_func *finalize,
 601                                    trns_proc_func *proc,
 602                                    trns_free_func *free, void *aux)
 603 {
 604   trns_chain_append (ds->cur_trns_chain, finalize, proc, free, aux);
 605
 606   if ( ds->xform_callback)
 607     ds->xform_callback (true, ds->xform_callback_aux);
 608 }
 609
 610 /* Returns the index of the next transformation.
 611    This value can be returned by a transformation procedure
 612    function to indicate a "jump" to that transformation. */
 613 size_t
 614 next_transformation (const struct dataset *ds)
 615 {
 616   return trns_chain_next (ds->cur_trns_chain);
 617 }
 618
 619 /* Returns true if the next call to add_transformation() will add
 620    a temporary transformation, false if it will add a permanent
 621    transformation. */
 622 bool
 623 proc_in_temporary_transformations (const struct dataset *ds)
 624 {
 625   return ds->temporary_trns_chain != NULL;
 626 }
 627
 628 /* Marks the start of temporary transformations.
 629    Further calls to add_transformation() will add temporary
 630    transformations. */
 631 void
 632 proc_start_temporary_transformations (struct dataset *ds)
 633 {
 634   if (!proc_in_temporary_transformations (ds))
 635     {
 636       add_case_limit_trns (ds);
 637
 638       ds->permanent_dict = dict_clone (ds->dict);
 639
 640       trns_chain_finalize (ds->permanent_trns_chain);
 641       ds->temporary_trns_chain = ds->cur_trns_chain = trns_chain_create ();
 642
 643       if ( ds->xform_callback)
 644         ds->xform_callback (true, ds->xform_callback_aux);
 645     }
 646 }
 647
 648 /* Converts all the temporary transformations, if any, to
 649    permanent transformations.  Further transformations will be
 650    permanent.
 651    Returns true if anything changed, false otherwise. */
 652 bool
 653 proc_make_temporary_transformations_permanent (struct dataset *ds)
 654 {
 655   if (proc_in_temporary_transformations (ds))
 656     {
 657       trns_chain_finalize (ds->temporary_trns_chain);
 658       trns_chain_splice (ds->permanent_trns_chain, ds->temporary_trns_chain);
 659       ds->temporary_trns_chain = NULL;
 660
 661       dict_destroy (ds->permanent_dict);
 662       ds->permanent_dict = NULL;
 663
 664       return true;
 665     }
 666   else
 667     return false;
 668 }
 669
 670 /* Cancels all temporary transformations, if any.  Further
 671    transformations will be permanent.
 672    Returns true if anything changed, false otherwise. */
 673 bool
 674 proc_cancel_temporary_transformations (struct dataset *ds)
 675 {
 676   if (proc_in_temporary_transformations (ds))
 677     {
 678       dict_destroy (ds->dict);
 679       ds->dict = ds->permanent_dict;
 680       ds->permanent_dict = NULL;
 681
 682       trns_chain_destroy (ds->temporary_trns_chain);
 683       ds->temporary_trns_chain = NULL;
 684
 685       if ( ds->xform_callback)
 686         ds->xform_callback (!trns_chain_is_empty (ds->permanent_trns_chain),
 687                             ds->xform_callback_aux);
 688
 689       return true;
 690     }
 691   else
 692     return false;
 693 }
 694
 695 /* Cancels all transformations, if any.
 696    Returns true if successful, false on I/O error. */
 697 bool
 698 proc_cancel_all_transformations (struct dataset *ds)
 699 {
 700   bool ok;
 701   assert (ds->proc_state == PROC_COMMITTED);
 702   ok = trns_chain_destroy (ds->permanent_trns_chain);
 703   ok = trns_chain_destroy (ds->temporary_trns_chain) && ok;
 704   ds->permanent_trns_chain = ds->cur_trns_chain = trns_chain_create ();
 705   ds->temporary_trns_chain = NULL;
 706   if ( ds->xform_callback)
 707     ds->xform_callback (false, ds->xform_callback_aux);
 708
 709   return ok;
 710 }
 711 \f
 712
 713 void
 714 dataset_add_transform_change_callback (struct dataset *ds,
 715                                        transformation_change_callback_func *cb,
 716                                        void *aux)
 717 {
 718   ds->xform_callback = cb;
 719   ds->xform_callback_aux = aux;
 720 }
 721
 722 /* Causes output from the next procedure to be discarded, instead
 723    of being preserved for use as input for the next procedure. */
 724 void
 725 proc_discard_output (struct dataset *ds)
 726 {
 727   ds->discard_output = true;
 728 }
 729
 730
 731 /* Checks whether DS has a corrupted active file.  If so,
 732    discards it and returns false.  If not, returns true without
 733    doing anything. */
 734 bool
 735 dataset_end_of_command (struct dataset *ds)
 736 {
 737   if (ds->source != NULL)
 738     {
 739       if (casereader_error (ds->source))
 740         {
 741           dataset_clear (ds);
 742           return false;
 743         }
 744       else
 745         {
 746           const struct taint *taint = casereader_get_taint (ds->source);
 747           taint_reset_successor_taint (CONST_CAST (struct taint *, taint));
 748           assert (!taint_has_tainted_successor (taint));
 749         }
 750     }
 751   return true;
 752 }
 753 \f
 754 static trns_proc_func case_limit_trns_proc;
 755 static trns_free_func case_limit_trns_free;
 756
 757 /* Adds a transformation that limits the number of cases that may
 758    pass through, if DS->DICT has a case limit. */
 759 static void
 760 add_case_limit_trns (struct dataset *ds)
 761 {
 762   casenumber case_limit = dict_get_case_limit (ds->dict);
 763   if (case_limit != 0)
 764     {
 765       casenumber *cases_remaining = xmalloc (sizeof *cases_remaining);
 766       *cases_remaining = case_limit;
 767       add_transformation (ds, case_limit_trns_proc, case_limit_trns_free,
 768                           cases_remaining);
 769       dict_set_case_limit (ds->dict, 0);
 770     }
 771 }
 772
 773 /* Limits the maximum number of cases processed to
 774    *CASES_REMAINING. */
 775 static int
 776 case_limit_trns_proc (void *cases_remaining_,
 777                       struct ccase **c UNUSED, casenumber case_nr UNUSED)
 778 {
 779   size_t *cases_remaining = cases_remaining_;
 780   if (*cases_remaining > 0)
 781     {
 782       (*cases_remaining)--;
 783       return TRNS_CONTINUE;
 784     }
 785   else
 786     return TRNS_DROP_CASE;
 787 }
 788
 789 /* Frees the data associated with a case limit transformation. */
 790 static bool
 791 case_limit_trns_free (void *cases_remaining_)
 792 {
 793   size_t *cases_remaining = cases_remaining_;
 794   free (cases_remaining);
 795   return true;
 796 }
 797 \f
 798 static trns_proc_func filter_trns_proc;
 799
 800 /* Adds a temporary transformation to filter data according to
 801    the variable specified on FILTER, if any. */
 802 static void
 803 add_filter_trns (struct dataset *ds)
 804 {
 805   struct variable *filter_var = dict_get_filter (ds->dict);
 806   if (filter_var != NULL)
 807     {
 808       proc_start_temporary_transformations (ds);
 809       add_transformation (ds, filter_trns_proc, NULL, filter_var);
 810     }
 811 }
 812
 813 /* FILTER transformation. */
 814 static int
 815 filter_trns_proc (void *filter_var_,
 816                   struct ccase **c UNUSED, casenumber case_nr UNUSED)
 817
 818 {
 819   struct variable *filter_var = filter_var_;
 820   double f = case_num (*c, filter_var);
 821   return (f != 0.0 && !var_is_num_missing (filter_var, f, MV_ANY)
 822           ? TRNS_CONTINUE : TRNS_DROP_CASE);
 823 }
 824
 825
 826 void
 827 dataset_need_lag (struct dataset *ds, int n_before)
 828 {
 829   ds->n_lag = MAX (ds->n_lag, n_before);
 830 }