1 /* PSPP - a program for statistical analysis.
2 Copyright (C) 1997-9, 2000, 2006, 2007, 2009, 2010, 2011, 2013 Free Software Foundation, Inc.
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.
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.
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/>. */
19 #include "data/dataset.h"
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/session.h"
36 #include "data/transformations.h"
37 #include "data/variable.h"
38 #include "libpspp/deque.h"
39 #include "libpspp/misc.h"
40 #include "libpspp/str.h"
41 #include "libpspp/taint.h"
42 #include "libpspp/i18n.h"
44 #include "gl/minmax.h"
45 #include "gl/xalloc.h"
48 /* A dataset is usually part of a session. Within a session its name must
49 unique. The name must either be a valid PSPP identifier or the empty
50 string. (It must be unique within the session even if it is the empty
51 string; that is, there may only be a single dataset within a session with
52 the empty string as its name.) */
53 struct session *session;
55 enum dataset_display display;
57 /* Cases are read from source,
58 their transformation variables are initialized,
59 pass through permanent_trns_chain (which transforms them into
60 the format described by permanent_dict),
62 pass through temporary_trns_chain (which transforms them into
63 the format described by dict),
64 and are finally passed to the procedure. */
65 struct casereader *source;
66 struct caseinit *caseinit;
67 struct trns_chain *permanent_trns_chain;
68 struct dictionary *permanent_dict;
69 struct casewriter *sink;
70 struct trns_chain *temporary_trns_chain;
71 struct dictionary *dict;
73 /* If true, cases are discarded instead of being written to
77 /* The transformation chain that the next transformation will be
79 struct trns_chain *cur_trns_chain;
81 /* The case map used to compact a case, if necessary;
82 otherwise a null pointer. */
83 struct case_map *compactor;
85 /* Time at which proc was last invoked. */
86 time_t last_proc_invocation;
88 /* Cases just before ("lagging") the current one. */
89 int n_lag; /* Number of cases to lag. */
90 struct deque lag; /* Deque of lagged cases. */
91 struct ccase **lag_cases; /* Lagged cases managed by deque. */
96 PROC_COMMITTED, /* No procedure in progress. */
97 PROC_OPEN, /* proc_open called, casereader still open. */
98 PROC_CLOSED /* casereader from proc_open destroyed,
99 but proc_commit not yet called. */
102 casenumber cases_written; /* Cases output so far. */
103 bool ok; /* Error status. */
104 struct casereader_shim *shim; /* Shim on proc_open() casereader. */
106 const struct dataset_callbacks *callbacks;
109 /* Uniquely distinguishes datasets. */
113 static void dataset_changed__ (struct dataset *);
114 static void dataset_transformations_changed__ (struct dataset *,
117 static void add_case_limit_trns (struct dataset *ds);
118 static void add_filter_trns (struct dataset *ds);
120 static void update_last_proc_invocation (struct dataset *ds);
123 dict_callback (struct dictionary *d UNUSED, void *ds_)
125 struct dataset *ds = ds_;
126 dataset_changed__ (ds);
130 dataset_create_finish__ (struct dataset *ds, struct session *session)
132 static unsigned int seqno;
134 dict_set_change_callback (ds->dict, dict_callback, ds);
135 proc_cancel_all_transformations (ds);
136 dataset_set_session (ds, session);
140 /* Creates a new dataset named NAME, adds it to SESSION, and returns it. If
141 SESSION already contains a dataset named NAME, it is deleted and replaced.
142 The dataset initially has an empty dictionary and no data source. */
144 dataset_create (struct session *session, const char *name)
146 struct dataset *ds = XZALLOC (struct dataset);
147 ds->name = xstrdup (name);
148 ds->display = DATASET_FRONT;
149 ds->dict = dict_create (get_default_encoding ());
151 ds->caseinit = caseinit_create ();
153 dataset_create_finish__ (ds, session);
158 /* Creates and returns a new dataset that has the same data and dictionary as
159 OLD named NAME, adds it to the same session as OLD, and returns the new
160 dataset. If SESSION already contains a dataset named NAME, it is deleted
163 OLD must not have any active transformations or temporary state and must
164 not be in the middle of a procedure.
166 Callbacks are not cloned. */
168 dataset_clone (struct dataset *old, const char *name)
172 assert (old->proc_state == PROC_COMMITTED);
173 assert (trns_chain_is_empty (old->permanent_trns_chain));
174 assert (old->permanent_dict == NULL);
175 assert (old->sink == NULL);
176 assert (old->temporary_trns_chain == NULL);
178 new = xzalloc (sizeof *new);
179 new->name = xstrdup (name);
180 new->display = DATASET_FRONT;
181 new->source = casereader_clone (old->source);
182 new->dict = dict_clone (old->dict);
183 new->caseinit = caseinit_clone (old->caseinit);
184 new->last_proc_invocation = old->last_proc_invocation;
187 dataset_create_finish__ (new, old->session);
194 dataset_destroy (struct dataset *ds)
198 dataset_set_session (ds, NULL);
200 dict_unref (ds->dict);
201 caseinit_destroy (ds->caseinit);
202 trns_chain_destroy (ds->permanent_trns_chain);
203 dataset_transformations_changed__ (ds, false);
209 /* Discards the active dataset's dictionary, data, and transformations. */
211 dataset_clear (struct dataset *ds)
213 assert (ds->proc_state == PROC_COMMITTED);
215 dict_clear (ds->dict);
216 fh_set_default_handle (NULL);
220 casereader_destroy (ds->source);
223 proc_cancel_all_transformations (ds);
227 dataset_name (const struct dataset *ds)
233 dataset_set_name (struct dataset *ds, const char *name)
235 struct session *session = ds->session;
240 active = session_active_dataset (session) == ds;
242 session_set_active_dataset (session, NULL);
243 dataset_set_session (ds, NULL);
247 ds->name = xstrdup (name);
251 dataset_set_session (ds, session);
253 session_set_active_dataset (session, ds);
258 dataset_session (const struct dataset *ds)
264 dataset_set_session (struct dataset *ds, struct session *session)
266 if (session != ds->session)
268 if (ds->session != NULL)
269 session_remove_dataset (ds->session, ds);
271 session_add_dataset (session, ds);
275 /* Returns the dictionary within DS. This is always nonnull, although it
276 might not contain any variables. */
278 dataset_dict (const struct dataset *ds)
283 /* Replaces DS's dictionary by DICT, discarding any source and
286 dataset_set_dict (struct dataset *ds, struct dictionary *dict)
288 assert (ds->proc_state == PROC_COMMITTED);
289 assert (ds->dict != dict);
293 dict_unref (ds->dict);
295 dict_set_change_callback (ds->dict, dict_callback, ds);
298 /* Returns the casereader that will be read when a procedure is executed on
299 DS. This can be NULL if none has been set up yet. */
300 const struct casereader *
301 dataset_source (const struct dataset *ds)
306 /* Returns true if DS has a data source, false otherwise. */
308 dataset_has_source (const struct dataset *ds)
310 return dataset_source (ds) != NULL;
313 /* Replaces the active dataset's data by READER. READER's cases must have an
314 appropriate format for DS's dictionary. */
316 dataset_set_source (struct dataset *ds, struct casereader *reader)
318 casereader_destroy (ds->source);
321 caseinit_clear (ds->caseinit);
322 caseinit_mark_as_preinited (ds->caseinit, ds->dict);
324 return reader == NULL || !casereader_error (reader);
327 /* Returns the data source from DS and removes it from DS. Returns a null
328 pointer if DS has no data source. */
330 dataset_steal_source (struct dataset *ds)
332 struct casereader *reader = ds->source;
338 /* Returns a number unique to DS. It can be used to distinguish one dataset
339 from any other within a given program run, even datasets that do not exist
342 dataset_seqno (const struct dataset *ds)
348 dataset_set_callbacks (struct dataset *ds,
349 const struct dataset_callbacks *callbacks,
352 ds->callbacks = callbacks;
353 ds->cb_data = cb_data;
357 dataset_get_display (const struct dataset *ds)
363 dataset_set_display (struct dataset *ds, enum dataset_display display)
365 ds->display = display;
368 /* Returns the last time the data was read. */
370 time_of_last_procedure (struct dataset *ds)
372 if (ds->last_proc_invocation == 0)
373 update_last_proc_invocation (ds);
374 return ds->last_proc_invocation;
377 /* Regular procedure. */
379 /* Executes any pending transformations, if necessary.
380 This is not identical to the EXECUTE command in that it won't
381 always read the source data. This can be important when the
382 source data is given inline within BEGIN DATA...END FILE. */
384 proc_execute (struct dataset *ds)
388 if ((ds->temporary_trns_chain == NULL
389 || trns_chain_is_empty (ds->temporary_trns_chain))
390 && trns_chain_is_empty (ds->permanent_trns_chain))
393 ds->discard_output = false;
394 dict_set_case_limit (ds->dict, 0);
395 dict_clear_vectors (ds->dict);
399 ok = casereader_destroy (proc_open (ds));
400 return proc_commit (ds) && ok;
403 static const struct casereader_class proc_casereader_class;
405 /* Opens dataset DS for reading cases with proc_read. If FILTER is true, then
406 cases filtered out with FILTER BY will not be included in the casereader
407 (which is usually desirable). If FILTER is false, all cases will be
408 included regardless of FILTER BY settings.
410 proc_commit must be called when done. */
412 proc_open_filtering (struct dataset *ds, bool filter)
414 struct casereader *reader;
416 assert (ds->source != NULL);
417 assert (ds->proc_state == PROC_COMMITTED);
419 update_last_proc_invocation (ds);
421 caseinit_mark_for_init (ds->caseinit, ds->dict);
423 /* Finish up the collection of transformations. */
424 add_case_limit_trns (ds);
426 add_filter_trns (ds);
427 trns_chain_finalize (ds->cur_trns_chain);
429 /* Make permanent_dict refer to the dictionary right before
430 data reaches the sink. */
431 if (ds->permanent_dict == NULL)
432 ds->permanent_dict = ds->dict;
435 if (!ds->discard_output)
437 struct dictionary *pd = ds->permanent_dict;
438 size_t compacted_value_cnt = dict_count_values (pd, 1u << DC_SCRATCH);
439 if (compacted_value_cnt < dict_get_next_value_idx (pd))
441 struct caseproto *compacted_proto;
442 compacted_proto = dict_get_compacted_proto (pd, 1u << DC_SCRATCH);
443 ds->compactor = case_map_to_compact_dict (pd, 1u << DC_SCRATCH);
444 ds->sink = autopaging_writer_create (compacted_proto);
445 caseproto_unref (compacted_proto);
449 ds->compactor = NULL;
450 ds->sink = autopaging_writer_create (dict_get_proto (pd));
455 ds->compactor = NULL;
459 /* Allocate memory for lagged cases. */
460 ds->lag_cases = deque_init (&ds->lag, ds->n_lag, sizeof *ds->lag_cases);
462 ds->proc_state = PROC_OPEN;
463 ds->cases_written = 0;
466 /* FIXME: use taint in dataset in place of `ok'? */
467 /* FIXME: for trivial cases we can just return a clone of
470 /* Create casereader and insert a shim on top. The shim allows us to
471 arbitrarily extend the casereader's lifetime, by slurping the cases into
472 the shim's buffer in proc_commit(). That is especially useful when output
473 table_items are generated directly from the procedure casereader (e.g. by
474 the LIST procedure) when we are using an output driver that keeps a
475 reference to the output items passed to it (e.g. the GUI output driver in
477 reader = casereader_create_sequential (NULL, dict_get_proto (ds->dict),
479 &proc_casereader_class, ds);
480 ds->shim = casereader_shim_insert (reader);
484 /* Opens dataset DS for reading cases with proc_read.
485 proc_commit must be called when done. */
487 proc_open (struct dataset *ds)
489 return proc_open_filtering (ds, true);
492 /* Returns true if a procedure is in progress, that is, if
493 proc_open has been called but proc_commit has not. */
495 proc_is_open (const struct dataset *ds)
497 return ds->proc_state != PROC_COMMITTED;
500 /* "read" function for procedure casereader. */
501 static struct ccase *
502 proc_casereader_read (struct casereader *reader UNUSED, void *ds_)
504 struct dataset *ds = ds_;
505 enum trns_result retval = TRNS_DROP_CASE;
508 assert (ds->proc_state == PROC_OPEN);
509 for (; ; case_unref (c))
513 assert (retval == TRNS_DROP_CASE || retval == TRNS_ERROR);
514 if (retval == TRNS_ERROR)
519 /* Read a case from source. */
520 c = casereader_read (ds->source);
523 c = case_unshare_and_resize (c, dict_get_proto (ds->dict));
524 caseinit_init_vars (ds->caseinit, c);
526 /* Execute permanent transformations. */
527 case_nr = ds->cases_written + 1;
528 retval = trns_chain_execute (ds->permanent_trns_chain, TRNS_CONTINUE,
530 caseinit_update_left_vars (ds->caseinit, c);
531 if (retval != TRNS_CONTINUE)
534 /* Write case to collection of lagged cases. */
537 while (deque_count (&ds->lag) >= ds->n_lag)
538 case_unref (ds->lag_cases[deque_pop_back (&ds->lag)]);
539 ds->lag_cases[deque_push_front (&ds->lag)] = case_ref (c);
542 /* Write case to replacement dataset. */
544 if (ds->sink != NULL)
545 casewriter_write (ds->sink,
546 case_map_execute (ds->compactor, case_ref (c)));
548 /* Execute temporary transformations. */
549 if (ds->temporary_trns_chain != NULL)
551 retval = trns_chain_execute (ds->temporary_trns_chain, TRNS_CONTINUE,
552 &c, ds->cases_written);
553 if (retval != TRNS_CONTINUE)
561 /* "destroy" function for procedure casereader. */
563 proc_casereader_destroy (struct casereader *reader, void *ds_)
565 struct dataset *ds = ds_;
568 /* We are always the subreader for a casereader_buffer, so if we're being
569 destroyed then it's because the casereader_buffer has read all the cases
570 that it ever will. */
573 /* Make sure transformations happen for every input case, in
574 case they have side effects, and ensure that the replacement
575 active dataset gets all the cases it should. */
576 while ((c = casereader_read (reader)) != NULL)
579 ds->proc_state = PROC_CLOSED;
580 ds->ok = casereader_destroy (ds->source) && ds->ok;
582 dataset_set_source (ds, NULL);
585 /* Must return false if the source casereader, a transformation,
586 or the sink casewriter signaled an error. (If a temporary
587 transformation signals an error, then the return value is
588 false, but the replacement active dataset may still be
591 proc_commit (struct dataset *ds)
593 if (ds->shim != NULL)
594 casereader_shim_slurp (ds->shim);
596 assert (ds->proc_state == PROC_CLOSED);
597 ds->proc_state = PROC_COMMITTED;
599 dataset_changed__ (ds);
601 /* Free memory for lagged cases. */
602 while (!deque_is_empty (&ds->lag))
603 case_unref (ds->lag_cases[deque_pop_back (&ds->lag)]);
604 free (ds->lag_cases);
606 /* Dictionary from before TEMPORARY becomes permanent. */
607 proc_cancel_temporary_transformations (ds);
609 if (!ds->discard_output)
611 /* Finish compacting. */
612 if (ds->compactor != NULL)
614 case_map_destroy (ds->compactor);
615 ds->compactor = NULL;
617 dict_delete_scratch_vars (ds->dict);
618 dict_compact_values (ds->dict);
621 /* Old data sink becomes new data source. */
622 if (ds->sink != NULL)
623 ds->source = casewriter_make_reader (ds->sink);
628 ds->discard_output = false;
632 caseinit_clear (ds->caseinit);
633 caseinit_mark_as_preinited (ds->caseinit, ds->dict);
635 dict_clear_vectors (ds->dict);
636 ds->permanent_dict = NULL;
637 return proc_cancel_all_transformations (ds) && ds->ok;
640 /* Casereader class for procedure execution. */
641 static const struct casereader_class proc_casereader_class =
643 proc_casereader_read,
644 proc_casereader_destroy,
649 /* Updates last_proc_invocation. */
651 update_last_proc_invocation (struct dataset *ds)
653 ds->last_proc_invocation = time (NULL);
656 /* Returns a pointer to the lagged case from N_BEFORE cases before the
657 current one, or NULL if there haven't been that many cases yet. */
659 lagged_case (const struct dataset *ds, int n_before)
661 assert (n_before >= 1);
662 assert (n_before <= ds->n_lag);
664 if (n_before <= deque_count (&ds->lag))
665 return ds->lag_cases[deque_front (&ds->lag, n_before - 1)];
670 /* Returns the current set of permanent transformations,
671 and clears the permanent transformations.
672 For use by INPUT PROGRAM. */
674 proc_capture_transformations (struct dataset *ds)
676 struct trns_chain *chain;
678 assert (ds->temporary_trns_chain == NULL);
679 chain = ds->permanent_trns_chain;
680 ds->cur_trns_chain = ds->permanent_trns_chain = trns_chain_create ();
681 dataset_transformations_changed__ (ds, false);
686 /* Adds a transformation that processes a case with PROC and
687 frees itself with FREE to the current set of transformations.
688 The functions are passed AUX as auxiliary data. */
690 add_transformation (struct dataset *ds, trns_proc_func *proc, trns_free_func *free, void *aux)
692 trns_chain_append (ds->cur_trns_chain, NULL, proc, free, aux);
693 dataset_transformations_changed__ (ds, true);
696 /* Adds a transformation that processes a case with PROC and
697 frees itself with FREE to the current set of transformations.
698 When parsing of the block of transformations is complete,
699 FINALIZE will be called.
700 The functions are passed AUX as auxiliary data. */
702 add_transformation_with_finalizer (struct dataset *ds,
703 trns_finalize_func *finalize,
704 trns_proc_func *proc,
705 trns_free_func *free, void *aux)
707 trns_chain_append (ds->cur_trns_chain, finalize, proc, free, aux);
708 dataset_transformations_changed__ (ds, true);
711 /* Returns the index of the next transformation.
712 This value can be returned by a transformation procedure
713 function to indicate a "jump" to that transformation. */
715 next_transformation (const struct dataset *ds)
717 return trns_chain_next (ds->cur_trns_chain);
720 /* Returns true if the next call to add_transformation() will add
721 a temporary transformation, false if it will add a permanent
724 proc_in_temporary_transformations (const struct dataset *ds)
726 return ds->temporary_trns_chain != NULL;
729 /* Marks the start of temporary transformations.
730 Further calls to add_transformation() will add temporary
733 proc_start_temporary_transformations (struct dataset *ds)
735 if (!proc_in_temporary_transformations (ds))
737 add_case_limit_trns (ds);
739 ds->permanent_dict = dict_clone (ds->dict);
741 trns_chain_finalize (ds->permanent_trns_chain);
742 ds->temporary_trns_chain = ds->cur_trns_chain = trns_chain_create ();
743 dataset_transformations_changed__ (ds, true);
747 /* Converts all the temporary transformations, if any, to permanent
748 transformations. Further transformations will be permanent.
750 The FILTER command is implemented as a temporary transformation, so a
751 procedure that uses this function should usually use proc_open_filtering()
752 with FILTER false, instead of plain proc_open().
754 Returns true if anything changed, false otherwise. */
756 proc_make_temporary_transformations_permanent (struct dataset *ds)
758 if (proc_in_temporary_transformations (ds))
760 trns_chain_finalize (ds->temporary_trns_chain);
761 trns_chain_splice (ds->permanent_trns_chain, ds->temporary_trns_chain);
762 ds->temporary_trns_chain = NULL;
764 ds->cur_trns_chain = ds->permanent_trns_chain;
766 dict_unref (ds->permanent_dict);
767 ds->permanent_dict = NULL;
775 /* Cancels all temporary transformations, if any. Further
776 transformations will be permanent.
777 Returns true if anything changed, false otherwise. */
779 proc_cancel_temporary_transformations (struct dataset *ds)
781 if (proc_in_temporary_transformations (ds))
783 dict_unref (ds->dict);
784 ds->dict = ds->permanent_dict;
785 ds->permanent_dict = NULL;
787 trns_chain_destroy (ds->temporary_trns_chain);
788 ds->temporary_trns_chain = NULL;
789 dataset_transformations_changed__ (
790 ds, !trns_chain_is_empty (ds->permanent_trns_chain));
797 /* Cancels all transformations, if any.
798 Returns true if successful, false on I/O error. */
800 proc_cancel_all_transformations (struct dataset *ds)
803 assert (ds->proc_state == PROC_COMMITTED);
804 ok = trns_chain_destroy (ds->permanent_trns_chain);
805 ok = trns_chain_destroy (ds->temporary_trns_chain) && ok;
806 ds->permanent_trns_chain = ds->cur_trns_chain = trns_chain_create ();
807 ds->temporary_trns_chain = NULL;
808 dataset_transformations_changed__ (ds, false);
814 store_case_num (void *var_, struct ccase **cc, casenumber case_num)
816 struct variable *var = var_;
818 *cc = case_unshare (*cc);
819 *case_num_rw (*cc, var) = case_num;
821 return TRNS_CONTINUE;
824 /* Add a variable which we can sort by to get back the original order. */
826 add_permanent_ordering_transformation (struct dataset *ds)
828 struct variable *temp_var;
830 temp_var = dict_create_var_assert (ds->dict, "$ORDER", 0);
831 if (proc_in_temporary_transformations (ds))
833 struct variable *perm_var;
835 perm_var = dict_clone_var_in_place_assert (ds->permanent_dict, temp_var);
836 trns_chain_append (ds->permanent_trns_chain, NULL, store_case_num,
838 trns_chain_finalize (ds->permanent_trns_chain);
841 add_transformation (ds, store_case_num, NULL, temp_var);
846 /* Causes output from the next procedure to be discarded, instead
847 of being preserved for use as input for the next procedure. */
849 proc_discard_output (struct dataset *ds)
851 ds->discard_output = true;
855 /* Checks whether DS has a corrupted active dataset. If so,
856 discards it and returns false. If not, returns true without
859 dataset_end_of_command (struct dataset *ds)
861 if (ds->source != NULL)
863 if (casereader_error (ds->source))
870 const struct taint *taint = casereader_get_taint (ds->source);
871 taint_reset_successor_taint (CONST_CAST (struct taint *, taint));
872 assert (!taint_has_tainted_successor (taint));
878 static trns_proc_func case_limit_trns_proc;
879 static trns_free_func case_limit_trns_free;
881 /* Adds a transformation that limits the number of cases that may
882 pass through, if DS->DICT has a case limit. */
884 add_case_limit_trns (struct dataset *ds)
886 casenumber case_limit = dict_get_case_limit (ds->dict);
889 casenumber *cases_remaining = xmalloc (sizeof *cases_remaining);
890 *cases_remaining = case_limit;
891 add_transformation (ds, case_limit_trns_proc, case_limit_trns_free,
893 dict_set_case_limit (ds->dict, 0);
897 /* Limits the maximum number of cases processed to
900 case_limit_trns_proc (void *cases_remaining_,
901 struct ccase **c UNUSED, casenumber case_nr UNUSED)
903 size_t *cases_remaining = cases_remaining_;
904 if (*cases_remaining > 0)
906 (*cases_remaining)--;
907 return TRNS_CONTINUE;
910 return TRNS_DROP_CASE;
913 /* Frees the data associated with a case limit transformation. */
915 case_limit_trns_free (void *cases_remaining_)
917 size_t *cases_remaining = cases_remaining_;
918 free (cases_remaining);
922 static trns_proc_func filter_trns_proc;
924 /* Adds a temporary transformation to filter data according to
925 the variable specified on FILTER, if any. */
927 add_filter_trns (struct dataset *ds)
929 struct variable *filter_var = dict_get_filter (ds->dict);
930 if (filter_var != NULL)
932 proc_start_temporary_transformations (ds);
933 add_transformation (ds, filter_trns_proc, NULL, filter_var);
937 /* FILTER transformation. */
939 filter_trns_proc (void *filter_var_,
940 struct ccase **c, casenumber case_nr UNUSED)
943 struct variable *filter_var = filter_var_;
944 double f = case_num (*c, filter_var);
945 return (f != 0.0 && !var_is_num_missing (filter_var, f, MV_ANY)
946 ? TRNS_CONTINUE : TRNS_DROP_CASE);
951 dataset_need_lag (struct dataset *ds, int n_before)
953 ds->n_lag = MAX (ds->n_lag, n_before);
957 dataset_changed__ (struct dataset *ds)
959 if (ds->callbacks != NULL && ds->callbacks->changed != NULL)
960 ds->callbacks->changed (ds->cb_data);
964 dataset_transformations_changed__ (struct dataset *ds, bool non_empty)
966 if (ds->callbacks != NULL && ds->callbacks->transformations_changed != NULL)
967 ds->callbacks->transformations_changed (non_empty, ds->cb_data);
970 /* Private interface for use by session code. */
973 dataset_set_session__ (struct dataset *ds, struct session *session)
975 ds->session = session;