#include <data/case-sink.h>
#include <data/case.h>
#include <data/casefile.h>
+#include <data/fastfile.h>
#include <data/dictionary.h>
#include <data/file-handle-def.h>
#include <data/procedure.h>
#include <libpspp/misc.h>
#include <libpspp/str.h>
-/*
- 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
{
struct ccase trns_case; /* Case used for transformations. */
struct ccase sink_case; /* Case written to sink, if
- compaction is necessary. */
+ compacting is necessary. */
size_t cases_written; /* Cases output so far. */
};
-/* Cases are read from vfm_source,
+/* Cases are read from proc_source,
pass through permanent_trns_chain (which transforms them into
the format described by permanent_dict),
- are written to vfm_sink,
+ are written to proc_sink,
pass through temporary_trns_chain (which transforms them into
the format described by default_dict),
and are finally passed to the procedure. */
-static struct case_source *vfm_source;
+static struct case_source *proc_source;
static struct trns_chain *permanent_trns_chain;
static struct dictionary *permanent_dict;
-static struct case_sink *vfm_sink;
+static struct case_sink *proc_sink;
static struct trns_chain *temporary_trns_chain;
struct dictionary *default_dict;
otherwise a null pointer. */
static struct dict_compactor *compactor;
-/* Time at which vfm was last invoked. */
-static time_t last_vfm_invocation;
+/* Time at which proc was last invoked. */
+static time_t last_proc_invocation;
/* Lag queue. */
int n_lag; /* Number of cases to lag. */
void *),
bool (*end_func) (void *),
void *aux);
-static void update_last_vfm_invocation (void);
+static void update_last_proc_invocation (void);
static void create_trns_case (struct ccase *, struct dictionary *);
static void open_active_file (void);
static bool write_case (struct write_case_data *wc_data);
time_t
time_of_last_procedure (void)
{
- if (last_vfm_invocation == 0)
- update_last_vfm_invocation ();
- return last_vfm_invocation;
+ if (last_proc_invocation == 0)
+ update_last_proc_invocation ();
+ return last_proc_invocation;
}
\f
/* Regular procedure. */
struct multipass_aux_data aux_data;
bool ok;
- aux_data.casefile = casefile_create (dict_get_next_value_idx (default_dict));
+ aux_data.casefile = fastfile_create (dict_get_next_value_idx (default_dict));
aux_data.proc_func = proc_func;
aux_data.aux = aux;
struct write_case_data wc_data;
bool ok = true;
- assert (vfm_source != NULL);
+ assert (proc_source != NULL);
- update_last_vfm_invocation ();
+ update_last_proc_invocation ();
/* Optimize the trivial case where we're not going to do
anything with the data, by not reading the data at all. */
if (case_func == NULL && end_func == NULL
- && case_source_is_class (vfm_source, &storage_source_class)
- && vfm_sink == NULL
+ && case_source_is_class (proc_source, &storage_source_class)
+ && proc_sink == NULL
&& (temporary_trns_chain == NULL
|| trns_chain_is_empty (temporary_trns_chain))
&& trns_chain_is_empty (permanent_trns_chain))
wc_data.case_func = case_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));
+ case_create (&wc_data.sink_case,
+ dict_get_compacted_value_cnt (default_dict));
wc_data.cases_written = 0;
- ok = vfm_source->class->read (vfm_source,
- &wc_data.trns_case,
- write_case, &wc_data) && ok;
+ ok = proc_source->class->read (proc_source,
+ &wc_data.trns_case,
+ write_case, &wc_data) && ok;
if (end_func != NULL)
ok = end_func (aux) && ok;
return ok;
}
-/* Updates last_vfm_invocation. */
+/* Updates last_proc_invocation. */
static void
-update_last_vfm_invocation (void)
+update_last_proc_invocation (void)
{
- last_vfm_invocation = time (NULL);
+ last_proc_invocation = time (NULL);
}
/* Creates and returns a case, initializing it from the vectors
if (permanent_dict == NULL)
permanent_dict = default_dict;
- /* Figure out compaction. */
- compactor = (dict_needs_compaction (permanent_dict)
+ /* Figure out whether to compact. */
+ compactor = (dict_compacting_would_shrink (permanent_dict)
? dict_make_compactor (permanent_dict)
: NULL);
/* Prepare sink. */
- if (vfm_sink == NULL)
- vfm_sink = create_case_sink (&storage_sink_class, permanent_dict, NULL);
- if (vfm_sink->class->open != NULL)
- vfm_sink->class->open (vfm_sink);
+ if (proc_sink == NULL)
+ proc_sink = create_case_sink (&storage_sink_class, permanent_dict, NULL);
+ if (proc_sink->class->open != NULL)
+ proc_sink->class->open (proc_sink);
/* Allocate memory for lag queue. */
if (n_lag > 0)
/* Write case to replacement active file. */
wc_data->cases_written++;
- if (vfm_sink->class->write != NULL)
+ if (proc_sink->class->write != NULL)
{
if (compactor != NULL)
{
dict_compactor_compact (compactor, &wc_data->sink_case,
&wc_data->trns_case);
- vfm_sink->class->write (vfm_sink, &wc_data->sink_case);
+ proc_sink->class->write (proc_sink, &wc_data->sink_case);
}
else
- vfm_sink->class->write (vfm_sink, &wc_data->trns_case);
+ proc_sink->class->write (proc_sink, &wc_data->trns_case);
}
/* Execute temporary transformations. */
/* Dictionary from before TEMPORARY becomes permanent. */
proc_cancel_temporary_transformations ();
- /* Finish compaction. */
+ /* Finish compacting. */
if (compactor != NULL)
{
dict_compactor_destroy (compactor);
}
/* Free data source. */
- free_case_source (vfm_source);
- vfm_source = NULL;
+ free_case_source (proc_source);
+ proc_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;
+ if (proc_sink->class->make_source != NULL)
+ proc_source = proc_sink->class->make_source (proc_sink);
+ free_case_sink (proc_sink);
+ proc_sink = NULL;
dict_clear_vectors (default_dict);
permanent_dict = NULL;
/* 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. */
/* Callback functions. */
struct split_aux_data split_aux;
bool ok;
- split_aux.case_count = 0;
case_nullify (&split_aux.prev_case);
split_aux.begin_func = begin_func;
split_aux.proc_func = proc_func;
struct split_aux_data *split_aux = split_aux_;
/* Start a new series if needed. */
- if (split_aux->case_count == 0
+ if (case_is_null (&split_aux->prev_case)
|| !equal_splits (c, &split_aux->prev_case))
{
- if (split_aux->case_count > 0 && split_aux->end_func != NULL)
+ if (!case_is_null (&split_aux->prev_case) && split_aux->end_func != NULL)
split_aux->end_func (split_aux->func_aux);
case_destroy (&split_aux->prev_case);
split_aux->begin_func (&split_aux->prev_case, split_aux->func_aux);
}
- split_aux->case_count++;
return (split_aux->proc_func == NULL
|| split_aux->proc_func (c, split_aux->func_aux));
}
{
struct split_aux_data *split_aux = split_aux_;
- if (split_aux->case_count > 0 && split_aux->end_func != NULL)
+ if (!case_is_null (&split_aux->prev_case) && split_aux->end_func != NULL)
split_aux->end_func (split_aux->func_aux);
return true;
}
ok = multipass_split_output (aux);
/* Start a new casefile. */
- aux->casefile = casefile_create (dict_get_next_value_idx (default_dict));
+ aux->casefile = fastfile_create (dict_get_next_value_idx (default_dict));
}
return casefile_append (aux->casefile, c) && ok;
n_lag = 0;
- free_case_source (vfm_source);
- vfm_source = NULL;
+ free_case_source (proc_source);
+ proc_source = NULL;
proc_cancel_all_transformations ();
}
proc_done (void)
{
discard_variables ();
+ dict_destroy (default_dict);
}
/* Sets SINK as the destination for procedure output from the
void
proc_set_sink (struct case_sink *sink)
{
- assert (vfm_sink == NULL);
- vfm_sink = sink;
+ assert (proc_sink == NULL);
+ proc_sink = sink;
}
/* Sets SOURCE as the source for procedure input for the next
void
proc_set_source (struct case_source *source)
{
- assert (vfm_source == NULL);
- vfm_source = source;
+ assert (proc_source == NULL);
+ proc_source = source;
}
/* Returns true if a source for the next procedure has been
bool
proc_has_source (void)
{
- return vfm_source != NULL;
+ return proc_source != NULL;
}
/* Returns the output from the previous procedure.
/* Try to make sure that this function is called immediately
after procedure() or a similar function. */
- assert (vfm_source != NULL);
- assert (case_source_is_class (vfm_source, &storage_source_class));
+ assert (proc_source != NULL);
+ assert (case_source_is_class (proc_source, &storage_source_class));
assert (trns_chain_is_empty (permanent_trns_chain));
assert (!proc_in_temporary_transformations ());
- casefile = storage_source_decapsulate (vfm_source);
- vfm_source = NULL;
+ casefile = storage_source_decapsulate (proc_source);
+ proc_source = NULL;
return casefile;
}
*CASES_REMAINING. */
static int
case_limit_trns_proc (void *cases_remaining_,
- struct ccase *c UNUSED, int case_nr UNUSED)
+ struct ccase *c UNUSED, casenum_t case_nr UNUSED)
{
size_t *cases_remaining = cases_remaining_;
if (*cases_remaining > 0)
/* FILTER transformation. */
static int
filter_trns_proc (void *filter_var_,
- struct ccase *c UNUSED, int case_nr UNUSED)
+ struct ccase *c UNUSED, casenum_t case_nr UNUSED)
{
struct variable *filter_var = filter_var_;
projects / pspp-builds.git / blobdiff