-Time-stamp: <2004-03-14 21:37:40 blp>
+Time-stamp: <2004-03-14 22:12:02 blp>
TODO
----
Ugly cases:
-LAG records cases in step 4.
+LAG records cases in step 3.
AGGREGATE: When output goes to an external file, this is just an ordinary
-procedure. When output goes to the active file, step 4 should be skipped,
+procedure. When output goes to the active file, step 3 should be skipped,
because AGGREGATE creates its own case sink and writes to it in step 7. Also,
TEMPORARY has no effect and we just cancel it. Regardless of direction of
output, we should not implement AGGREGATE through a transformation because that
MATCH FILES: Similar to AGGREGATE. This is a procedure. When the active file
is used for input, it reads the active file; otherwise, it just cancels all the
-transformations and deletes the original active file. Step 4 should be
+transformations and deletes the original active file. Step 3 should be
skipped, because MATCH FILES creates its own case sink and writes to it in step
7. TEMPORARY is not allowed.
+Sun Mar 14 22:48:12 2004 Ben Pfaff <blp@gnu.org>
+
+ * command.def: Add CASESTOVARS, VARSTOCASES unimplemented commands.
+
+ * dictionary.c: (dict_rename_var) Add assertion.
+ (dict_contains_var) Check by index instead of name.
+
+Sun Mar 14 22:01:02 2004 Ben Pfaff <blp@gnu.org>
+
+ Get rid of compaction_necessary, compaction_nval, compaction_case.
+ Redo VFM interface. Replace disk_sink and memory_sink by
+ storage_sink, disk_source and memory_source by storage_source.
+
+ * vfm.h: (struct case_sink) Add `dict', `idx_to_fv', `value_cnt'
+ members.
+
+ * vfm.c:
+ (struct write_case_data) Remove `begin_func', `end_func',
+ `func_aux' members. Add `aux', `trns_case', `sink_case',
+ `cases_written', `cases_analyzed' members.
+ (global var compaction_necessary) Make static.
+ (global var compaction_nval) Removed.
+ (global var compaction_case) Removed.
+ (static var case_count) Removed.
+ (struct procedure_aux_data) Removed.
+ (struct split_aux_data) Removed.
+ (procedure) Remove begin_func, end_func parameters. Rewrite.
+ (static var not_canceled) Removed.
+ (process_active_file) Removed.
+ (process_active_file_write_case) Removed.
+ (process_active_file_output_case) Removed.
+ (prepare_for_writing) Moved into open_active_file().
+ (arrange_compaction) Ditto.
+ (setup_lag) Ditto.
+ (open_active_file) Rewrote.
+ (write_case) New function.
+ [DEBUGGING] (index_to_varname) Removed.
+ (execute_transformations) New function.
+ (exclude_this_case) Renamed filter_case(), changed interface.
+ (clear_case) Added struct ccase * parameter to interface.
+ (close_active_file) Added struct write_case_data * parameter,
+ rewrote.
+ (disk_sink_create) Removed.
+ (disk_sink_destroy) Removed.
+ (disk_sink_make_source) Removed.
+ (disk_sink_write) Removed.
+ (disk_source_count) Removed.
+ (disk_source_destroy) Removed.
+ (disk_source_read) Removed.
+ (global var disk_sink_class) Removed.
+ (global var disk_source_class) Removed.
+ (global var memory_sink_class) Removed.
+ (global var memory_source_class) Removed.
+ (memory_sink_create) Removed.
+ (memory_sink_destroy) Removed.
+ (memory_sink_make_source) Removed.
+ (memory_sink_write) Removed.
+ (memory_source_count) Removed.
+ (memory_source_destroy) Removed.
+ (memory_source_get_cases) Removed.
+ (memory_source_read) Removed.
+ (memory_source_set_cases) Removed.
+ (struct disk_stream_info) Removed.
+ (struct memory_sink_info) Removed.
+ (struct memory_source_info) Removed.
+ (write_active_file_to_disk) Removed.
+ (destroy_storage_stream_info) New function.
+ (global var null_sink_class) New var.
+ (global var storage_sink_class) New var.
+ (global var storage_source_class) New var.
+ (open_storage_file) New function.
+ (storage_sink_destroy) New function.
+ (storage_sink_make_source) New function.
+ (storage_sink_open) New function.
+ (storage_sink_write) New function.
+ (storage_source_count) New function.
+ (storage_source_destroy) New function.
+ (storage_source_get_cases) New function.
+ (storage_source_on_disk) New function.
+ (storage_source_read) New function.
+ (storage_source_set_cases) New function.
+ (storage_source_to_disk) New function.
+ (storage_to_disk) New function.
+ (struct storage_stream_info) New structure.
+ (write_storage_file) New function.
+ (procedure_write_case) Removed.
+ (create_case_source) Add `struct dictionary *' parameter, all
+ references updated.
+ (create_case_sink) Ditto.
+ (free_case_sink) New function.
+ (struct split_aux_data) New structure.
+ (procedure_with_splits) New function implementing what procedure()
+ used to.
+ (SPLIT_FILE_proc_func) Removed.
+ (procedure_with_splits_callback) New function.
+ (equal_splits) New function.
+
+ * aggregate.c: Pass around a struct instead of using statics.
+ (static var outfile) Remove.
+ (enum type) Give it tag `missing_treatment'.
+ (static var missing) Remove.
+ (static var sort) Remove.
+ (static var agr_first) Remove.
+ (static var agr_next) Remove.
+ (static var case_count) Remove.
+ (static var prev_case) Remove.
+ (static var buf64_1xx) Remove.
+ (static var buf_1xx) Remove.
+ (struct agr_proc) New structure incorporating the above.
+ (cmd_aggregate) Use new struct. Clean up error handling using
+ agr_destroy(). Completely rewrite actual implementation of
+ aggregation.
+ (create_sysfile) Add struct agr_proc * parameter, modify
+ accordingly.
+ (parse_aggregate_functions) Ditto.
+ (free_aggregate_functions) Ditto. Rename agr_destroy().
+ (aggregate_single_case) Add struct agr_proc * parameter, modify
+ accordingly.
+ (accumulate_aggregate_info) Ditto.
+ (dump_aggregate_info) Ditto.
+ (initialize_aggregate_info) Ditto.
+ (agr_00x_trns_proc) Removed.
+ (agr_00x_end_func) Removed.
+ (agr_10x_trns_proc) Removed.
+ (agr_10x_trns_free) Removed.
+ (agr_10x_end_func) Removed.
+ (agr_11x_read) Removed.
+ (agr_11x_finish) Removed.
+ [DEBUGGING] (debug_print) Removed.
+ (write_case_to_sfm) Add struct agr_proc * parameter, modify
+ accordingly.
+ (agr_to_active_file) New function.
+ (presorted_agr_to_sysfile) New function.
+ (sort_agr_to_sysfile) New function.
+
+ * autorecode.c: (cmd_autorecode) Use procedure_with_splits().
+
+ * crosstabs.q: (internal_cmd_crosstabs) Ditto.
+
+ * descript.q: (cmd_descriptives) Ditto.
+
+ * dfm.c: (cmd_begin_data) Check for storage_source_class. Adapt
+ to new procedure() interface.
+
+ * command.c: (cmd_execute) Adapt to new procedure() interface.
+
+ * dictionary.c: (dict_compact_values) Also delete scratch
+ variables.
+ (dict_get_compacted_value_cnt) New function.
+ (dict_get_compacted_idx_to_fv) New function.
+
+ * flip.c: (cmd_flip) Warn about and cancel TEMPORARY.
+ (cmd_flip) Adapt to new procedure() interface.
+ (flip_sink_write) Use sink->idx_to_fv.
+
+ * frequencies.q: (internal_cmd_frequencies) Use
+ procedure_with_splits().
+
+ * get.c: (cmd_save_internal) Adapt to new procedure() interface.
+ (static var mtf_sink) New static var.
+ (static var mtf_case) New static var.
+ (cmd_match_files) Warn about and cancel TEMPORARY. Redo the way
+ we actually implement the matching.
+ (mtf_delete_file_in_place) Use mtf_case.
+ (mtf_processing) Use mtf_case and mtf_sink.
+ (cmd_export) Adapt to new procedure() interface.
+
+ * levene.c: (levene) Use procedure_with_splits().
+
+ * list.q: (cmd_list) Use procedure_with_splits().
+
+ * matrix-data.c: (read_matrices_without_rowtype) Adapt to new
+ procedure() interface.
+ (read_matrices_with_rowtype) Ditto.
+
+ * modify-vars.c; (cmd_modify_vars) Warn about and cancel
+ TEMPORARY. Adapt to new procedure() interface.
+
+ * rename-vars.c: Warn about and cancel TEMPORARY.
+
+ * sort.c: (cmd_sort_cases) Warn about TEMPORARY.
+ (sort_cases) Use dict_get_compacted_value_cnt() instead of
+ compaction_nval. Adapt to new procedure() interface. Use
+ storage_source_to_disk().
+ (do_internal_sort) Don't try to dump the cases to memory.
+ (compare_case_lists) Pass null idx_to_fv.
+ (struct initial_run_state) Add `idx_to_fv' member. Remove
+ `case_size' member.
+ (write_initial_runs) Don't initialize irs->case_size. Adapt to
+ new procedure() interface. Reset irs->idx_to_fv after calling
+ procedure().
+ (sort_sink_write) Set irs->idx_to_fv. Use case_size from struct
+ sort_cases_pgm. Pass irs, not struct sort_cases_pgm to
+ push_heap().
+ (destroy_initial_run_state) Don't dereference irs after freeing
+ it.
+ (allocate_cases) Don't calculate case_size locally.
+ (compare_record) Add idx_to_fv parameter.
+ (compare_record_run) Change parameter from struct sort_cases_pgm *
+ to struct initial_run_state *. Pass irs->idx_to_fv to
+ compare_record().
+ (compare_record_run) Third parameter now a struct
+ initial_run_state *.
+ (output_record) No need for out_case anymore. Pass irs, not
+ struct sort_cases_pgm to pop_heap(). Use case_size from struct
+ sort_cases_pgm.
+ (merge) Use case_size from struct sort_cases_pgm.
+ (merge_once) Use case_size from struct sort_cases_pgm.
+ Pass null pointer to compare_record() as idx_to_fv.
+ (global var sort_sink_class) Make static.
+
+ * t-test.q: (cmd_t_test) Use procedure_with_splits().
+
+ * temporary.c: Remove debugging crap.
+
Sat Mar 13 14:19:52 WST 2004 John Darrington <john@darrington.wattle.id.au>
* t-test.q, levene.c: Fixed up the handling of MISSING values
#include "vfm.h"
#include "vfmP.h"
-#include "debug-print.h"
-
/* Specifies how to make an aggregate variable. */
struct agr_var
{
{"NU", 0, NUMERIC, {FMT_F, 7, 0}},
};
-/* Output file, or NULL for the active file. */
-static struct file_handle *outfile;
-
/* Missing value types. */
-enum
+enum missing_treatment
{
ITEMWISE, /* Missing values item by item. */
COLUMNWISE /* Missing values column by column. */
};
-/* ITEMWISE or COLUMNWISE. */
-static int missing;
-
-/* Sort program. */
-static struct sort_cases_pgm *sort;
-
-/* Aggregate variables. */
-static struct agr_var *agr_first, *agr_next;
-
-/* Aggregate dictionary. */
-static struct dictionary *agr_dict;
-
-/* Number of cases passed through aggregation. */
-static int case_count;
-
-/* Last values of the break variables. */
-static union value *prev_case;
-
-/* Buffers for use by the 10x transformation. */
-static flt64 *buf64_1xx;
-static struct ccase *buf_1xx;
+/* An entire AGGREGATE procedure. */
+struct agr_proc
+ {
+ /* We have either an output file or a sink. */
+ struct file_handle *out_file; /* Output file, or null if none. */
+ struct case_sink *sink; /* Sink, or null if none. */
+
+ enum missing_treatment missing; /* How to treat missing values. */
+ struct sort_cases_pgm *sort; /* Sort program. */
+ struct agr_var *vars; /* First aggregate variable. */
+ struct dictionary *dict; /* Aggregate dictionary. */
+ int case_cnt; /* Counts aggregated cases. */
+ union value *prev_break; /* Last values of break variables. */
+ struct ccase *agr_case; /* Aggregate case for output. */
+ flt64 *sfm_agr_case; /* Aggregate case in SFM format. */
+ };
-static void initialize_aggregate_info (void);
+static void initialize_aggregate_info (struct agr_proc *);
/* Prototypes. */
-static int parse_aggregate_functions (void);
-static void free_aggregate_functions (void);
-static int aggregate_single_case (const struct ccase *input,
+static int parse_aggregate_functions (struct agr_proc *);
+static void agr_destroy (struct agr_proc *);
+static int aggregate_single_case (struct agr_proc *agr,
+ const struct ccase *input,
struct ccase *output);
-static int create_sysfile (void);
-
-static trns_proc_func agr_00x_trns_proc, agr_10x_trns_proc;
-static trns_free_func agr_10x_trns_free;
-static void agr_00x_end_func (void *aux);
-static void agr_10x_end_func (void *);
-static read_sort_output_func agr_11x_read;
-static void agr_11x_finish (void);
-
-#if DEBUGGING
-static void debug_print (int flags);
-#endif
+static void dump_aggregate_info (struct agr_proc *agr, struct ccase *output);
+static int create_sysfile (struct agr_proc *);
+
+/* Aggregating to the active file. */
+static int agr_to_active_file (struct ccase *, void *aux);
+
+/* Aggregating to a system file. */
+static void write_case_to_sfm (struct agr_proc *agr);
+static int presorted_agr_to_sysfile (struct ccase *, void *aux);
+static int sort_agr_to_sysfile (const struct ccase *, void *aux);
\f
/* Parsing. */
int
cmd_aggregate (void)
{
+ struct agr_proc agr;
+
/* Have we seen these subcommands? */
unsigned seen = 0;
- outfile = NULL;
- missing = ITEMWISE;
- sort = NULL;
- prev_case = NULL;
+ agr.out_file = NULL;
+ agr.sink = NULL;
+ agr.missing = ITEMWISE;
+ agr.sort = NULL;
+ agr.vars = NULL;
+ agr.dict = NULL;
+ agr.case_cnt = 0;
+ agr.prev_break = NULL;
- agr_dict = dict_create ();
- dict_set_label (agr_dict, dict_get_label (default_dict));
- dict_set_documents (agr_dict, dict_get_documents (default_dict));
+ agr.dict = dict_create ();
+ dict_set_label (agr.dict, dict_get_label (default_dict));
+ dict_set_documents (agr.dict, dict_get_documents (default_dict));
lex_match_id ("AGGREGATE");
/* Read most of the subcommands. */
for (;;)
{
- lex_match('/');
+ lex_match ('/');
if (lex_match_id ("OUTFILE"))
{
if (seen & 1)
{
- destroy_sort_cases_pgm (sort);
- dict_destroy (agr_dict);
msg (SE, _("%s subcommand given multiple times."),"OUTFILE");
- return CMD_FAILURE;
+ goto lossage;
}
seen |= 1;
lex_match ('=');
if (lex_match ('*'))
- outfile = NULL;
+ agr.out_file = NULL;
else
{
- outfile = fh_parse_file_handle ();
- if (outfile == NULL)
- {
- destroy_sort_cases_pgm (sort);
- dict_destroy (agr_dict);
- return CMD_FAILURE;
- }
+ agr.out_file = fh_parse_file_handle ();
+ if (agr.out_file == NULL)
+ goto lossage;
}
}
else if (lex_match_id ("MISSING"))
lex_match ('=');
if (!lex_match_id ("COLUMNWISE"))
{
- destroy_sort_cases_pgm (sort);
- dict_destroy (agr_dict);
lex_error (_("while expecting COLUMNWISE"));
- return CMD_FAILURE;
+ goto lossage;
}
- missing = COLUMNWISE;
+ agr.missing = COLUMNWISE;
}
else if (lex_match_id ("DOCUMENT"))
seen |= 2;
{
if (seen & 8)
{
- destroy_sort_cases_pgm (sort);
- dict_destroy (agr_dict);
msg (SE, _("%s subcommand given multiple times."),"BREAK");
- return CMD_FAILURE;
+ goto lossage;
}
seen |= 8;
lex_match ('=');
- sort = parse_sort ();
- if (sort == NULL)
- {
- dict_destroy (agr_dict);
- return CMD_FAILURE;
- }
+ agr.sort = parse_sort ();
+ if (agr.sort == NULL)
+ goto lossage;
{
int i;
- for (i = 0; i < sort->var_cnt; i++)
+ for (i = 0; i < agr.sort->var_cnt; i++)
{
struct variable *v;
- v = dict_clone_var (agr_dict, sort->vars[i], sort->vars[i]->name);
+ v = dict_clone_var (agr.dict, agr.sort->vars[i],
+ agr.sort->vars[i]->name);
assert (v != NULL);
}
}
msg (SW, _("BREAK subcommand not specified."));
/* Read in the aggregate functions. */
- if (!parse_aggregate_functions ())
- {
- free_aggregate_functions ();
- destroy_sort_cases_pgm (sort);
- return CMD_FAILURE;
- }
+ if (!parse_aggregate_functions (&agr))
+ goto lossage;
/* Delete documents. */
if (!(seen & 2))
- dict_set_documents (agr_dict, NULL);
+ dict_set_documents (agr.dict, NULL);
/* Cancel SPLIT FILE. */
- dict_set_split_vars (agr_dict, NULL, 0);
+ dict_set_split_vars (agr.dict, NULL, 0);
-#if DEBUGGING
- debug_print (seen);
-#endif
-
/* Initialize. */
- case_count = 0;
- initialize_aggregate_info ();
-
- /* How to implement all this... There are three important variables:
- whether output is going to the active file (0) or a separate file
- (1); whether the input data is presorted (0) or needs sorting
- (1); whether there is a temporary transformation (1) or not (0).
- The eight cases are as follows:
-
- 000 (0): Pass it through an aggregate transformation that
- modifies the data.
-
- 001 (1): Cancel the temporary transformation and handle as 000.
-
- 010 (2): Set up a SORT CASES and aggregate the output, writing
- the results to the active file.
-
- 011 (3): Cancel the temporary transformation and handle as 010.
-
- 100 (4): Pass it through an aggregate transformation that doesn't
- modify the data but merely writes it to the output file.
-
- 101 (5): Handled as 100.
-
- 110 (6): Set up a SORT CASES and capture the output, aggregate
- it, write it to the output file without modifying the active
- file.
-
- 111 (7): Handled as 110. */
-
- {
- unsigned type = 0;
-
- if (outfile != NULL)
- type |= 4;
- if (sort != NULL && (seen & 4) == 0)
- type |= 2;
- if (temporary)
- type |= 1;
-
- switch (type)
- {
- case 3:
- cancel_temporary ();
- /* fall through */
- case 2:
- sort_cases (sort, 0);
- goto case0;
-
- case 1:
- cancel_temporary ();
- /* fall through */
- case 0:
- case0:
- {
- struct trns_header *t = xmalloc (sizeof *t);
- t->proc = agr_00x_trns_proc;
- t->free = NULL;
- add_transformation (t);
-
- temporary = 2;
- temp_dict = agr_dict;
- temp_trns = n_trns;
-
- agr_dict = NULL;
-
- procedure (NULL, NULL, agr_00x_end_func, NULL);
- break;
- }
+ agr.case_cnt = 0;
+ agr.agr_case = xmalloc (dict_get_case_size (agr.dict));
+ initialize_aggregate_info (&agr);
- case 4:
- case 5:
- {
- if (!create_sysfile ())
- goto lossage;
-
- {
- struct trns_header *t = xmalloc (sizeof *t);
- t->proc = agr_10x_trns_proc;
- t->free = agr_10x_trns_free;
- add_transformation (t);
-
- procedure (NULL, NULL, agr_10x_end_func, NULL);
- }
-
- break;
- }
-
- case 6:
- case 7:
- sort_cases (sort, 1);
-
- if (!create_sysfile ())
- goto lossage;
- read_sort_output (sort, agr_11x_read, NULL);
- agr_11x_finish ();
-
- break;
-
- default:
- assert (0);
- }
- }
-
- free (buf64_1xx);
- free (buf_1xx);
-
- /* Clean up. */
- destroy_sort_cases_pgm (sort);
- free_aggregate_functions ();
- free (prev_case);
+ /* Output to active file or external file? */
+ if (agr.out_file == NULL)
+ {
+ /* The active file will be replaced by the aggregated data,
+ so TEMPORARY is moot. */
+ cancel_temporary ();
+
+ if (agr.sort != NULL && (seen & 4) == 0)
+ sort_cases (agr.sort, 0);
+
+ agr.sink = create_case_sink (&storage_sink_class, agr.dict, NULL);
+ if (agr.sink->class->open != NULL)
+ agr.sink->class->open (agr.sink);
+ vfm_sink = create_case_sink (&null_sink_class, default_dict, NULL);
+ procedure (agr_to_active_file, &agr);
+ if (agr.case_cnt > 0)
+ {
+ dump_aggregate_info (&agr, agr.agr_case);
+ agr.sink->class->write (agr.sink, agr.agr_case);
+ }
+ dict_destroy (default_dict);
+ default_dict = agr.dict;
+ agr.dict = NULL;
+ vfm_source = agr.sink->class->make_source (agr.sink);
+ free_case_sink (agr.sink);
+ }
+ else
+ {
+ if (!create_sysfile (&agr))
+ goto lossage;
+
+ if (agr.sort != NULL && (seen & 4) == 0)
+ {
+ /* Sorting is needed. */
+ sort_cases (agr.sort, 1);
+ read_sort_output (agr.sort, sort_agr_to_sysfile, NULL);
+ }
+ else
+ {
+ /* Active file is already sorted. */
+ procedure (presorted_agr_to_sysfile, &agr);
+ }
+
+ if (agr.case_cnt > 0)
+ {
+ dump_aggregate_info (&agr, agr.agr_case);
+ write_case_to_sfm (&agr);
+ }
+ fh_close_handle (agr.out_file);
+ }
+ agr_destroy (&agr);
return CMD_SUCCESS;
lossage:
- /* Clean up. */
- destroy_sort_cases_pgm (sort);
- free_aggregate_functions ();
- free (prev_case);
-
+ agr_destroy (&agr);
return CMD_FAILURE;
}
-/* Create a system file for use in aggregation to an external file,
- and allocate temporary buffers for writing out cases. */
+/* Create a system file for use in aggregation to an external
+ file. */
static int
-create_sysfile (void)
+create_sysfile (struct agr_proc *agr)
{
struct sfm_write_info w;
- w.h = outfile;
- w.dict = agr_dict;
+ w.h = agr->out_file;
+ w.dict = agr->dict;
w.compress = set_scompression;
if (!sfm_write_dictionary (&w))
- {
- free_aggregate_functions ();
- destroy_sort_cases_pgm (sort);
- dict_destroy (agr_dict);
- return 0;
- }
-
- buf64_1xx = xmalloc (sizeof *buf64_1xx * w.case_size);
- buf_1xx = xmalloc (dict_get_case_size (agr_dict));
+ return 0;
+ agr->sfm_agr_case = xmalloc (sizeof *agr->sfm_agr_case * w.case_size);
+
return 1;
}
/* Parse all the aggregate functions. */
static int
-parse_aggregate_functions (void)
+parse_aggregate_functions (struct agr_proc *agr)
{
- agr_first = agr_next = NULL;
+ struct agr_var *tail; /* Tail of linked list starting at agr->vars. */
/* Parse everything. */
+ tail = NULL;
for (;;)
{
char **dest;
struct agr_var *v = xmalloc (sizeof *v);
/* Add variable to chain. */
- if (agr_first)
- agr_next = agr_next->next = v;
+ if (agr->vars != NULL)
+ tail->next = v;
else
- agr_first = agr_next = v;
- agr_next->next = NULL;
+ agr->vars = v;
+ tail = v;
+ tail->next = NULL;
/* Create the target variable in the aggregate
dictionary. */
{
struct variable *destvar;
- agr_next->function = func_index;
+ v->function = func_index;
if (src)
{
int output_width;
- agr_next->src = src[i];
+ v->src = src[i];
if (src[i]->type == ALPHA)
{
- agr_next->function |= FSTRING;
- agr_next->string = xmalloc (src[i]->width);
+ v->function |= FSTRING;
+ v->string = xmalloc (src[i]->width);
}
- if (agr_next->src->type == NUMERIC || function->alpha_type == NUMERIC)
+ if (v->src->type == NUMERIC || function->alpha_type == NUMERIC)
output_width = 0;
else
- output_width = agr_next->src->width;
+ output_width = v->src->width;
if (function->alpha_type == ALPHA)
- destvar = dict_clone_var (agr_dict, agr_next->src, dest[i]);
+ destvar = dict_clone_var (agr->dict, v->src, dest[i]);
else
{
- destvar = dict_create_var (agr_dict, dest[i], output_width);
+ destvar = dict_create_var (agr->dict, dest[i], output_width);
if (output_width == 0)
destvar->print = destvar->write = function->format;
if (output_width == 0 && dict_get_weight (default_dict) != NULL
}
}
} else {
- agr_next->src = NULL;
- destvar = dict_create_var (agr_dict, dest[i], 0);
+ v->src = NULL;
+ destvar = dict_create_var (agr->dict, dest[i], 0);
}
if (!destvar)
else if (function->alpha_type == ALPHA)
destvar->print = destvar->write = function->format;
- agr_next->dest = destvar;
+ v->dest = destvar;
}
- agr_next->include_missing = include_missing;
+ v->include_missing = include_missing;
- if (agr_next->src != NULL)
+ if (v->src != NULL)
{
int j;
- if (agr_next->src->type == NUMERIC)
+ if (v->src->type == NUMERIC)
for (j = 0; j < function->n_args; j++)
- agr_next->arg[j].f = arg[j].f;
+ v->arg[j].f = arg[j].f;
else
for (j = 0; j < function->n_args; j++)
- agr_next->arg[j].c = xstrdup (arg[j].c);
+ v->arg[j].c = xstrdup (arg[j].c);
}
}
if (src && n_src && src[0]->type == ALPHA)
for (i = 0; i < function->n_args; i++)
{
- free(arg[i].c);
+ free (arg[i].c);
arg[i].c = NULL;
}
free (src);
}
}
-/* Frees all the state for the AGGREGATE procedure. */
+/* Destroys AGR. */
static void
-free_aggregate_functions (void)
+agr_destroy (struct agr_proc *agr)
{
struct agr_var *iter, *next;
- if (agr_dict)
- dict_destroy (agr_dict);
- for (iter = agr_first; iter; iter = next)
+ if (agr->dict != NULL)
+ dict_destroy (agr->dict);
+ if (agr->sort != NULL)
+ destroy_sort_cases_pgm (agr->sort);
+ for (iter = agr->vars; iter; iter = next)
{
next = iter->next;
}
free (iter);
}
+ free (agr->prev_break);
+ free (agr->agr_case);
}
\f
/* Execution. */
-static void accumulate_aggregate_info (const struct ccase *input);
-static void dump_aggregate_info (struct ccase *output);
+static void accumulate_aggregate_info (struct agr_proc *,
+ const struct ccase *);
+static void dump_aggregate_info (struct agr_proc *, struct ccase *);
/* Processes a single case INPUT for aggregation. If output is
- warranted, it is written to case OUTPUT, which may be (but need not
- be) an alias to INPUT. Returns -1 when output is performed, -2
- otherwise. */
-/* The code in this function has an eerie similarity to
- vfm.c:SPLIT_FILE_procfunc()... */
+ warranted, writes it to OUTPUT and returns nonzero.
+ Otherwise, returns zero and OUTPUT is unmodified. */
static int
-aggregate_single_case (const struct ccase *input, struct ccase *output)
+aggregate_single_case (struct agr_proc *agr,
+ const struct ccase *input, struct ccase *output)
{
/* The first case always begins a new break group. We also need to
preserve the values of the case for later comparison. */
- if (case_count++ == 0)
+ if (agr->case_cnt++ == 0)
{
int n_elem = 0;
{
int i;
- for (i = 0; i < sort->var_cnt; i++)
- n_elem += sort->vars[i]->nv;
+ for (i = 0; i < agr->sort->var_cnt; i++)
+ n_elem += agr->sort->vars[i]->nv;
}
- prev_case = xmalloc (sizeof *prev_case * n_elem);
+ agr->prev_break = xmalloc (sizeof *agr->prev_break * n_elem);
- /* Copy INPUT into prev_case. */
+ /* Copy INPUT into prev_break. */
{
- union value *iter = prev_case;
+ union value *iter = agr->prev_break;
int i;
- for (i = 0; i < sort->var_cnt; i++)
+ for (i = 0; i < agr->sort->var_cnt; i++)
{
- struct variable *v = sort->vars[i];
+ struct variable *v = agr->sort->vars[i];
if (v->type == NUMERIC)
(iter++)->f = input->data[v->fv].f;
}
}
- accumulate_aggregate_info (input);
+ accumulate_aggregate_info (agr, input);
- return -2;
+ return 0;
}
/* Compare the value of each break variable to the values on the
previous case. */
{
- union value *iter = prev_case;
+ union value *iter = agr->prev_break;
int i;
- for (i = 0; i < sort->var_cnt; i++)
+ for (i = 0; i < agr->sort->var_cnt; i++)
{
- struct variable *v = sort->vars[i];
+ struct variable *v = agr->sort->vars[i];
switch (v->type)
{
}
}
- accumulate_aggregate_info (input);
+ accumulate_aggregate_info (agr, input);
- return -2;
+ return 0;
not_equal:
/* The values of the break variable are different from the values on
the previous case. That means that it's time to dump aggregate
info. */
- dump_aggregate_info (output);
- initialize_aggregate_info ();
- accumulate_aggregate_info (input);
+ dump_aggregate_info (agr, output);
+ initialize_aggregate_info (agr);
+ accumulate_aggregate_info (agr, input);
- /* Copy INPUT into prev_case. */
+ /* Copy INPUT into prev_break. */
{
- union value *iter = prev_case;
+ union value *iter = agr->prev_break;
int i;
- for (i = 0; i < sort->var_cnt; i++)
+ for (i = 0; i < agr->sort->var_cnt; i++)
{
- struct variable *v = sort->vars[i];
+ struct variable *v = agr->sort->vars[i];
if (v->type == NUMERIC)
(iter++)->f = input->data[v->fv].f;
}
}
- return -1;
+ return 1;
}
/* Accumulates aggregation data from the case INPUT. */
static void
-accumulate_aggregate_info (const struct ccase *input)
+accumulate_aggregate_info (struct agr_proc *agr,
+ const struct ccase *input)
{
struct agr_var *iter;
double weight;
weight = dict_get_case_weight (default_dict, input);
- for (iter = agr_first; iter; iter = iter->next)
+ for (iter = agr->vars; iter; iter = iter->next)
if (iter->src)
{
const union value *v = &input->data[iter->src->fv];
more of the break variables. Make an output record from the
accumulated statistics in the OUTPUT case. */
static void
-dump_aggregate_info (struct ccase *output)
+dump_aggregate_info (struct agr_proc *agr, struct ccase *output)
{
- debug_printf (("(dumping "));
-
{
int n_elem = 0;
{
int i;
- for (i = 0; i < sort->var_cnt; i++)
- n_elem += sort->vars[i]->nv;
+ for (i = 0; i < agr->sort->var_cnt; i++)
+ n_elem += agr->sort->vars[i]->nv;
}
- debug_printf (("n_elem=%d:", n_elem));
- memcpy (output->data, prev_case, sizeof (union value) * n_elem);
+ memcpy (output->data, agr->prev_break, sizeof (union value) * n_elem);
}
{
struct agr_var *i;
- for (i = agr_first; i; i = i->next)
+ for (i = agr->vars; i; i = i->next)
{
union value *v = &output->data[i->dest->fv];
- debug_printf ((" %d,%d", i->dest->fv, i->dest->nv));
-
- if (missing == COLUMNWISE && i->missing != 0
+ if (agr->missing == COLUMNWISE && i->missing != 0
&& (i->function & FUNC) != N && (i->function & FUNC) != NU
&& (i->function & FUNC) != NMISS && (i->function & FUNC) != NUMISS)
{
}
}
}
- debug_printf ((") "));
}
/* Resets the state for all the aggregate functions. */
static void
-initialize_aggregate_info (void)
+initialize_aggregate_info (struct agr_proc *agr)
{
struct agr_var *iter;
- for (iter = agr_first; iter; iter = iter->next)
+ for (iter = agr->vars; iter; iter = iter->next)
{
iter->missing = 0;
switch (iter->function)
/* Aggregate each case as it comes through. Cases which aren't needed
are dropped. */
static int
-agr_00x_trns_proc (struct trns_header *h UNUSED, struct ccase *c,
- int case_num UNUSED)
+agr_to_active_file (struct ccase *c, void *agr_)
{
- int code = aggregate_single_case (c, compaction_case);
- debug_printf (("%d ", code));
- return code;
-}
+ struct agr_proc *agr = agr_;
-/* Output the last aggregate case. It's okay to call the vfm_sink's
- write() method here because end_func is called so soon after all
- the cases have been output; very little has been cleaned up at this
- point. */
-static void
-agr_00x_end_func (void *aux UNUSED)
-{
- /* Ensure that info for the last break group gets written to the
- active file. */
- dump_aggregate_info (compaction_case);
- vfm_sink->class->write (vfm_sink, compaction_case);
+ if (aggregate_single_case (agr, c, agr->agr_case))
+ agr->sink->class->write (agr->sink, agr->agr_case);
+
+ return 1;
}
-/* Transform the aggregate case buf_1xx, in internal format, to system
- file format, in buf64_1xx, and write the resultant case to the
- system file. */
+/* Writes AGR->agr_case to AGR->out_file. */
static void
-write_case_to_sfm (void)
+write_case_to_sfm (struct agr_proc *agr)
{
- flt64 *p = buf64_1xx;
+ flt64 *p;
int i;
- for (i = 0; i < dict_get_var_cnt (agr_dict); i++)
+ p = agr->sfm_agr_case;
+ for (i = 0; i < dict_get_var_cnt (agr->dict); i++)
{
- struct variable *v = dict_get_var (agr_dict, i);
+ struct variable *v = dict_get_var (agr->dict, i);
if (v->type == NUMERIC)
{
- double src = buf_1xx->data[v->fv].f;
+ double src = agr->agr_case->data[v->fv].f;
if (src == SYSMIS)
*p++ = -FLT64_MAX;
else
}
else
{
- memcpy (p, buf_1xx->data[v->fv].s, v->width);
+ memcpy (p, agr->agr_case->data[v->fv].s, v->width);
memset (&((char *) p)[v->width], ' ',
REM_RND_UP (v->width, sizeof (flt64)));
p += DIV_RND_UP (v->width, sizeof (flt64));
}
}
- sfm_write_case (outfile, buf64_1xx, p - buf64_1xx);
+ sfm_write_case (agr->out_file, agr->sfm_agr_case, p - agr->sfm_agr_case);
}
/* Aggregate the current case and output it if we passed a
breakpoint. */
static int
-agr_10x_trns_proc (struct trns_header *h UNUSED, struct ccase *c,
- int case_num UNUSED)
+presorted_agr_to_sysfile (struct ccase *c, void *agr_)
{
- int code = aggregate_single_case (c, buf_1xx);
-
- assert (code == -2 || code == -1);
- if (code == -1)
- write_case_to_sfm ();
- return -1;
-}
-
-/* Close the system file now that we're done with it. */
-static void
-agr_10x_trns_free (struct trns_header *h UNUSED)
-{
- fh_close_handle (outfile);
-}
-
-/* Ensure that info for the last break group gets written to the
- system file. */
-static void
-agr_10x_end_func (void *aux UNUSED)
-{
- dump_aggregate_info (buf_1xx);
- write_case_to_sfm ();
-}
-
-/* Runs case C through the aggregater and outputs it to the
- system file if appropriate. */
-static int
-agr_11x_read (const struct ccase *c, void *aux UNUSED)
-{
- int code = aggregate_single_case (c, buf_1xx);
-
- assert (code == -2 || code == -1);
- if (code == -1)
- write_case_to_sfm ();
-
+ sort_agr_to_sysfile (c, agr_);
return 1;
}
-/* Finishes up writing the last case if necessary. */
-static void
-agr_11x_finish (void)
-{
- if (case_count)
- {
- dump_aggregate_info (buf_1xx);
- write_case_to_sfm ();
- }
- fh_close_handle (outfile);
-}
-\f
-/* Debugging. */
-#if DEBUGGING
-/* Print out useful debugging information. */
-static void
-debug_print (int flags)
+/* Aggregate the current case and output it if we passed a
+ breakpoint. */
+static int
+sort_agr_to_sysfile (const struct ccase *c, void *agr_)
{
- printf ("AGGREGATE\n /OUTFILE=%s\n",
- outfile ? fh_handle_filename (outfile) : "*");
+ struct agr_proc *agr = agr_;
- if (missing == COLUMNWISE)
- puts (" /MISSING=COLUMNWISE");
-
- if (flags & 2)
- puts (" /DOCUMENT");
- if (flags & 4)
- puts (" /PRESORTED");
-
- {
- int i;
+ if (aggregate_single_case (agr, c, agr->agr_case))
+ write_case_to_sfm (agr);
- printf (" /BREAK=");
- for (i = 0; i < sort->var_cnt; i++)
- printf ("%s(%c) ", sort->vars[i]->name,
- sort->vars[i]->p.srt.order == SRT_ASCEND ? 'A' : 'D');
- putc ('\n', stdout);
- }
-
- {
- struct agr_var *iter;
-
- for (iter = agr_first; iter; iter = iter->next)
- {
- struct agr_func *f = &agr_func_tab[iter->function & FUNC];
-
- printf (" /%s", iter->dest->name);
- if (iter->dest->label)
- printf ("'%s'", iter->dest->label);
- printf ("=%s(%s", f->name, iter->src->name);
- if (f->n_args)
- {
- int i;
-
- for (i = 0; i < f->n_args; i++)
- {
- putc (',', stdout);
- if (iter->src->type == NUMERIC)
- printf ("%g", iter->arg[i].f);
- else
- printf ("%.*s", iter->src->width, iter->arg[i].c);
- }
- }
- printf (")\n");
- }
- }
+ return 1;
}
-
-#endif /* DEBUGGING */
h_trans[i] = hsh_create (10, compare_numeric_value,
hash_numeric_value, NULL, NULL);
- procedure (NULL, autorecode_proc_func, NULL, NULL);
+ procedure_with_splits (NULL, autorecode_proc_func, NULL, NULL);
for (i = 0; i < nv_dest; i++)
{
cmd_execute (void)
{
lex_match_id ("EXECUTE");
- procedure (NULL, NULL, NULL, NULL);
+ procedure (NULL, NULL);
return lex_end_of_command ();
}
-
#define assert_not_safer() \
do { \
if (set_safer) \
} \
} while(0)
-
-
/* Parses, performs the ERASE command. */
int
cmd_erase (void)
DEFCMD ("AUTORECODE", ERRO, ERRO, PROC, PROC, cmd_autorecode)
DEFCMD ("BEGIN DATA", ERRO, ERRO, PROC, PROC, cmd_begin_data)
DEFCMD ("BREAK", ERRO, INPU, TRAN, TRAN, cmd_break)
+UNIMPL ("CASESTOVARS", ERRO, ERRO, PROC, PROC)
DEFCMD ("CLEAR TRANSFORMATIONS", ERRO, INPU, TRAN, TRAN, cmd_clear_transformations)
DEFCMD ("COMPUTE", ERRO, INPU, TRAN, TRAN, cmd_compute)
DEFCMD ("CORRELATIONS", ERRO, ERRO, PROC, PROC, cmd_correlations)
UNIMPL ("UPDATE", TRAN, ERRO, TRAN, TRAN)
DEFCMD ("VALUE LABELS", ERRO, INPU, TRAN, TRAN, cmd_value_labels)
DEFCMD ("VARIABLE LABELS", ERRO, INPU, TRAN, TRAN, cmd_variable_labels)
+UNIMPL ("VARSTOCASES", ERRO, ERRO, PROC, PROC)
DEFCMD ("VECTOR", ERRO, INPU, TRAN, TRAN, cmd_vector)
DEFCMD ("WEIGHT", ERRO, INPU, TRAN, TRAN, cmd_weight)
DEFCMD ("WRITE", ERRO, INPU, TRAN, TRAN, cmd_write)
else
write = CRS_WR_NONE;
- procedure (precalc, mode == GENERAL ? calc_general : calc_integer, postcalc,
- NULL);
+ procedure_with_splits (precalc,
+ mode == GENERAL ? calc_general : calc_integer,
+ postcalc, NULL);
return CMD_SUCCESS;
}
/* Data pass! */
bad_weight = 0;
- procedure (precalc, calc, postcalc, NULL);
+ procedure_with_splits (precalc, calc, postcalc, NULL);
if (bad_weight)
msg (SW, _("At least one case in the data file had a weight value "
/* FIXME: figure out the *exact* conditions, not these really
lenient conditions. */
if (vfm_source == NULL
- || case_source_is_class (vfm_source, &memory_source_class)
- || case_source_is_class (vfm_source, &disk_source_class)
+ || case_source_is_class (vfm_source, &storage_source_class)
|| case_source_is_class (vfm_source, &sort_source_class))
{
msg (SE, _("This command is not valid here since the current "
/* We don't actually read from the inline file. The input procedure
is what reads from it. */
getl_prompt = GETL_PRPT_DATA;
- procedure (NULL, NULL, NULL, NULL);
+ procedure (NULL, NULL);
ext = inline_file->ext;
assert (v != NULL);
assert (new_name != NULL);
assert (strlen (new_name) >= 1 && strlen (new_name) <= 8);
+ assert (dict_contains_var (d, v));
if (!strcmp (v->name, new_name))
return;
assert (d != NULL);
assert (v != NULL);
- return dict_lookup_var (d, v->name) == v;
+ return v->index >= 0 && v->index < d->var_cnt && d->var[v->index] == v;
}
/* Compares two double pointers to variables, which should point
return sizeof (union value) * dict_get_next_value_idx (d);
}
-/* Reassigns values in dictionary D so that fragmentation is
- eliminated. */
+/* Deletes scratch variables in dictionary D and reassigns values
+ so that fragmentation is eliminated. */
void
dict_compact_values (struct dictionary *d)
{
size_t i;
- d->next_value_idx = 0;
+ for (i = 0; i < d->var_cnt; )
+ {
+ struct variable *v = d->var[i];
+
+ if (dict_class_from_id (v->name) != DC_SCRATCH)
+ {
+ v->fv = d->next_value_idx;
+ d->next_value_idx += v->nv;
+ i++;
+ }
+ else
+ dict_delete_var (default_dict, v);
+ }
+}
+
+/* Returns the number of values that would be used by a case if
+ dict_compact_values() were called. */
+size_t
+dict_get_compacted_value_cnt (const struct dictionary *d)
+{
+ size_t i;
+ size_t cnt;
+
+ cnt = 0;
+ for (i = 0; i < d->var_cnt; i++)
+ if (dict_class_from_id (d->var[i]->name) != DC_SCRATCH)
+ cnt += d->var[i]->nv;
+ return cnt;
+}
+
+/* Creates and returns an array mapping from a dictionary index
+ to the `fv' that the corresponding variable will have after
+ calling dict_compact_values(). Scratch variables receive -1
+ for `fv' because dict_compact_values() will delete them. */
+int *
+dict_get_compacted_idx_to_fv (const struct dictionary *d)
+{
+ size_t i;
+ size_t next_value_idx;
+ int *idx_to_fv;
+
+ idx_to_fv = xmalloc (d->var_cnt * sizeof *idx_to_fv);
+ next_value_idx = 0;
for (i = 0; i < d->var_cnt; i++)
{
struct variable *v = d->var[i];
- v->fv = d->next_value_idx;
- d->next_value_idx += v->nv;
+ if (dict_class_from_id (v->name) != DC_SCRATCH)
+ {
+ idx_to_fv[i] = next_value_idx;
+ next_value_idx += v->nv;
+ }
+ else
+ idx_to_fv[i] = -1;
}
+ return idx_to_fv;
}
/* Returns the SPLIT FILE vars (see cmd_split_file()). Call
{
struct flip_pgm *flip;
+ if (temporary != 0)
+ {
+ msg (SM, _("FLIP ignores TEMPORARY. "
+ "Temporary transformations will be made permanent."));
+ cancel_temporary ();
+ }
+
flip = xmalloc (sizeof *flip);
flip->var = NULL;
flip->var_cnt = 0;
temp_trns = temporary = 0;
vfm_sink = flip_sink_create (flip);
flip->new_names_tail = NULL;
- procedure (NULL, NULL, NULL, NULL);
+ procedure (NULL, NULL);
/* Flip the data we read. */
flip_file (flip);
flip->case_cnt = 1;
- return create_case_sink (&flip_sink_class, info);
+ return create_case_sink (&flip_sink_class, default_dict, info);
}
/* Writes case C to the FLIP sink. */
v->next = NULL;
if (flip->new_names->type == NUMERIC)
{
- double f = c->data[flip->new_names->fv].f;
+ double f = c->data[sink->idx_to_fv[flip->new_names->index]].f;
if (f == SYSMIS)
strcpy (v->name, "VSYSMIS");
else
{
int width = min (flip->new_names->width, 8);
- memcpy (v->name, c->data[flip->new_names->fv].s, width);
+ memcpy (v->name, c->data[sink->idx_to_fv[flip->new_names->index]].s,
+ width);
v->name[width] = 0;
}
/* Write to external file. */
for (i = 0; i < flip->var_cnt; i++)
if (flip->var[i]->type == NUMERIC)
- info->output_buf[i].f = c->data[flip->var[i]->fv].f;
+ info->output_buf[i].f = c->data[sink->idx_to_fv[flip->var[i]->index]].f;
else
info->output_buf[i].f = SYSMIS;
cmd.sort = FRQ_AVALUE;
/* Do it! */
- procedure (precalc, calc, postcalc, NULL);
+ procedure_with_splits (precalc, calc, postcalc, NULL);
return CMD_SUCCESS;
}
if (save_cmd == CMD_SAVE)
{
- procedure (NULL, save_write_case_func, NULL, t);
+ procedure (save_write_case_func, t);
save_trns_free (&t->h);
}
else
/* Sequence numbers for each variable in mtf_master. */
static unsigned *mtf_seq_nums;
+/* Sink for MATCH FILES output. */
+static struct case_sink *mtf_sink;
+
+/* Case used for MATCH FILES output. */
+static struct ccase *mtf_case;
+
static void mtf_free (void);
static void mtf_free_file (struct mtf_file *file);
static int mtf_merge_dictionary (struct mtf_file *f);
"file has been defined."));
goto lossage;
}
+
+ if (temporary != 0)
+ {
+ msg (SE,
+ _("MATCH FILES may not be used after TEMPORARY when "
+ "the active file is an input source. "
+ "Temporary transformations will be made permanent."));
+ cancel_temporary ();
+ }
}
else
{
if (!(seen & 2))
discard_variables ();
- temporary = 2;
- temp_dict = mtf_master;
- temp_trns = n_trns;
+ mtf_sink = create_case_sink (&storage_sink_class, mtf_master, NULL);
mtf_seq_nums = xmalloc (dict_get_var_cnt (mtf_master)
* sizeof *mtf_seq_nums);
memset (mtf_seq_nums, 0,
dict_get_var_cnt (mtf_master) * sizeof *mtf_seq_nums);
+ mtf_case = xmalloc (dict_get_case_size (mtf_master));
+
+ mtf_read_nonactive_records (NULL);
+ if (seen & 2)
+ procedure (mtf_processing, NULL);
+ mtf_processing_finish (NULL);
- process_active_file (mtf_read_nonactive_records, mtf_processing,
- mtf_processing_finish, NULL);
+ dict_destroy (default_dict);
+ default_dict = mtf_master;
mtf_master = NULL;
+ vfm_source = mtf_sink->class->make_source (mtf_sink);
+ free_case_sink (mtf_sink);
mtf_free ();
return CMD_SUCCESS;
struct variable *v = dict_get_var (f->dict, i);
if (v->type == NUMERIC)
- compaction_case->data[v->p.mtf.master->fv].f = SYSMIS;
+ mtf_case->data[v->p.mtf.master->fv].f = SYSMIS;
else
- memset (compaction_case->data[v->p.mtf.master->fv].s, ' ',
- v->width);
+ memset (mtf_case->data[v->p.mtf.master->fv].s, ' ', v->width);
}
}
-
+
mtf_free_file (f);
}
assert (v->type == NUMERIC || v->type == ALPHA);
if (v->type == NUMERIC)
- compaction_case->data[v->p.mtf.master->fv].f = record[v->fv].f;
+ mtf_case->data[v->p.mtf.master->fv].f = record[v->fv].f;
else
- memcpy (compaction_case->data[v->p.mtf.master->fv].s,
+ memcpy (mtf_case->data[v->p.mtf.master->fv].s,
record[v->fv].s, v->width);
}
}
v->p.mtf.master->fv);
#endif
if (v->type == NUMERIC)
- compaction_case->data[v->p.mtf.master->fv].f = SYSMIS;
+ mtf_case->data[v->p.mtf.master->fv].f = SYSMIS;
else
- memset (compaction_case->data[v->p.mtf.master->fv].s, ' ',
+ memset (mtf_case->data[v->p.mtf.master->fv].s, ' ',
v->width);
}
}
/* 6. Write the output record. */
- process_active_file_output_case (compaction_case);
+ mtf_sink->class->write (mtf_sink, mtf_case);
/* 7. Read another record from each input file FILE and TABLE
that we stored values from above. If we come to the end of
t->case_buf = xmalloc (sizeof *t->case_buf * t->nvar);
dict_destroy (dict);
- procedure (NULL, export_write_case_func, NULL, t);
+ procedure (export_write_case_func, t);
save_trns_free (&t->h);
return CMD_SUCCESS;
l.missing = missing;
l.is_missing = value_is_missing;
- procedure(levene_precalc, levene_calc, levene_postcalc, &l);
- procedure(levene2_precalc,levene2_calc,levene2_postcalc,&l);
+ procedure_with_splits (levene_precalc, levene_calc, levene_postcalc, &l);
+ procedure_with_splits (levene2_precalc, levene2_calc, levene2_postcalc, &l);
}
determine_layout ();
case_num = 0;
- procedure (write_all_headers, list_cases, NULL, NULL);
+ procedure_with_splits (write_all_headers, list_cases, NULL, NULL);
free (line_buf);
clean_up ();
vfm_source = create_case_source (&matrix_data_without_rowtype_source_class,
default_dict, mx);
- procedure (NULL, NULL, NULL, NULL);
+ procedure (NULL, NULL);
free (split_values);
free (nr_factor_values);
vfm_source = create_case_source (&matrix_data_with_rowtype_source_class,
default_dict, mx);
- procedure (NULL, NULL, NULL, NULL);
+ procedure (NULL, NULL);
free (split_values);
fh_close_handle (mx->data_file);
size_t i;
+ if (temporary != 0)
+ {
+ msg (SE, _("MODIFY VARS may not be used after TEMPORARY. "
+ "Temporary transformations will be made permanent."));
+ cancel_temporary ();
+ }
+
lex_match_id ("MODIFY");
lex_match_id ("VARS");
if (already_encountered & (1 | 4))
{
/* Read the data. */
- procedure (NULL, NULL, NULL, NULL);
+ procedure (NULL, NULL);
}
if (!rearrange_dict (default_dict, &vm))
int i;
+ if (temporary != 0)
+ {
+ msg (SE, _("RENAME VARS may not be used after TEMPORARY. "
+ "Temporary transformations will be made permanent."));
+ cancel_temporary ();
+ }
+
lex_match_id ("RENAME");
lex_match_id ("VARIABLES");
/* Other prototypes. */
static int compare_record (const union value *, const union value *,
- const struct sort_cases_pgm *);
+ const struct sort_cases_pgm *, int *idx_to_fv);
static int compare_case_lists (const void *, const void *, void *);
static struct internal_sort *do_internal_sort (struct sort_cases_pgm *,
int separate);
scp = parse_sort ();
if (scp == NULL)
return CMD_FAILURE;
-
- cancel_temporary ();
+
+ if (temporary != 0)
+ {
+ msg (SE, _("SORT CASES may not be used after TEMPORARY. "
+ "Temporary transformations will be made permanent."));
+ cancel_temporary ();
+ }
success = sort_cases (scp, 0);
destroy_sort_cases_pgm (scp);
int
sort_cases (struct sort_cases_pgm *scp, int separate)
{
- scp->case_size = sizeof (union value) * compaction_nval;
+ scp->case_size
+ = sizeof (union value) * dict_get_compacted_value_cnt (default_dict);
/* Not sure this is necessary but it's good to be safe. */
if (separate && case_source_is_class (vfm_source, &sort_source_class))
- procedure (NULL, NULL, NULL, NULL);
+ procedure (NULL, NULL);
/* SORT CASES cancels PROCESS IF. */
expr_free (process_if_expr);
return 1;
/* Fall back to an external sort. */
- write_active_file_to_disk ();
+ if (vfm_source != NULL
+ && case_source_is_class (vfm_source, &storage_source_class))
+ storage_source_to_disk (vfm_source);
scp->xsrt = do_external_sort (scp, separate);
if (scp->xsrt != NULL)
return 1;
struct case_list **results;
};
-/* If a reasonable situation is set up, do an internal sort of the
- data. Return success. */
+/* If the data is in memory, do an internal sort. Return
+ success. */
static struct internal_sort *
do_internal_sort (struct sort_cases_pgm *scp, int separate)
{
isrt = xmalloc (sizeof *isrt);
isrt->results = NULL;
- if (!case_source_is_class (vfm_source, &disk_source_class))
+ if (case_source_is_class (vfm_source, &storage_source_class)
+ && !storage_source_on_disk (vfm_source))
{
- if (!case_source_is_class (vfm_source, &memory_source_class))
- procedure (NULL, NULL, NULL, NULL);
-
- if (case_source_is_class (vfm_source, &memory_source_class))
- {
- struct case_list *case_list;
- struct case_list **case_array;
- int case_cnt;
- int i;
+ struct case_list *case_list;
+ struct case_list **case_array;
+ int case_cnt;
+ int i;
- case_cnt = vfm_source->class->count (vfm_source);
- if (case_cnt <= 0)
- return isrt;
+ case_cnt = vfm_source->class->count (vfm_source);
+ if (case_cnt <= 0)
+ return isrt;
- if (case_cnt > set_max_workspace / sizeof *case_array)
- goto error;
+ if (case_cnt > set_max_workspace / sizeof *case_array)
+ goto error;
- case_list = memory_source_get_cases (vfm_source);
- case_array = malloc (sizeof *case_array * (case_cnt + 1));
- if (case_array == NULL)
- goto error;
+ case_list = storage_source_get_cases (vfm_source);
+ case_array = malloc (sizeof *case_array * (case_cnt + 1));
+ if (case_array == NULL)
+ goto error;
- for (i = 0; case_list != NULL; i++)
- {
- case_array[i] = case_list;
- case_list = case_list->next;
- }
- assert (i == case_cnt);
- case_array[case_cnt] = NULL;
+ for (i = 0; case_list != NULL; i++)
+ {
+ case_array[i] = case_list;
+ case_list = case_list->next;
+ }
+ assert (i == case_cnt);
+ case_array[case_cnt] = NULL;
- sort (case_array, case_cnt, sizeof *case_array,
- compare_case_lists, scp);
+ sort (case_array, case_cnt, sizeof *case_array,
+ compare_case_lists, scp);
- if (!separate)
- {
- memory_source_set_cases (vfm_source, case_array[0]);
- for (i = 1; i <= case_cnt; i++)
- case_array[i - 1]->next = case_array[i];
- free (case_array);
- }
- else
- isrt->results = case_array;
+ if (!separate)
+ {
+ storage_source_set_cases (vfm_source, case_array[0]);
+ for (i = 1; i <= case_cnt; i++)
+ case_array[i - 1]->next = case_array[i];
+ free (case_array);
+ }
+ else
+ isrt->results = case_array;
- return isrt;
- }
+ return isrt;
}
error:
struct case_list *a = *pa;
struct case_list *b = *pb;
- return compare_record (a->c.data, b->c.data, scp);
+ return compare_record (a->c.data, b->c.data, scp, NULL);
}
\f
/* External sort. */
{
struct external_sort *xsrt;
+ int *idx_to_fv; /* Translation table copied from sink. */
+
/* Reservoir. */
struct record_run *records; /* Records arranged as a heap. */
size_t record_cnt; /* Current number of records. */
/* Run currently being output. */
int file_idx; /* Temporary file number. */
size_t case_cnt; /* Number of cases so far. */
- size_t case_size; /* Number of bytes in a case. */
FILE *output_file; /* Output file. */
struct case_list *last_output;/* Record last output. */
int okay; /* Zero if an error has been encountered. */
};
+static const struct case_sink_class sort_sink_class;
+
static void destroy_initial_run_state (struct initial_run_state *irs);
static int allocate_cases (struct initial_run_state *);
static struct case_list *grab_case (struct initial_run_state *);
static void end_run (struct initial_run_state *irs);
static int compare_record_run (const struct record_run *,
const struct record_run *,
- struct sort_cases_pgm *);
+ struct initial_run_state *);
static int compare_record_run_minheap (const void *, const void *, void *);
/* Writes initial runs for XSRT, sending them to a separate file
irs->last_output = NULL;
irs->file_idx = 0;
irs->case_cnt = 0;
- irs->case_size = dict_get_case_size (default_dict);
irs->okay = 1;
if (!allocate_cases (irs))
goto done;
{
if (vfm_sink != NULL && vfm_sink->class->destroy != NULL)
vfm_sink->class->destroy (vfm_sink);
- vfm_sink = create_case_sink (&sort_sink_class, irs);
+ vfm_sink = create_case_sink (&sort_sink_class, default_dict, irs);
xsrt->scp->ref_cnt++;
}
/* Create initial runs. */
start_run (irs);
- procedure (NULL, NULL, NULL, NULL);
+ procedure (NULL, NULL);
+ irs->idx_to_fv = NULL;
while (irs->record_cnt > 0 && irs->okay)
output_record (irs);
end_run (irs);
if (!irs->okay)
return;
+ irs->idx_to_fv = sink->idx_to_fv;
+
/* Compose record_run for this run and add to heap. */
assert (irs->record_cnt < irs->record_cap);
new_record_run = irs->records + irs->record_cnt++;
new_record_run->record = grab_case (irs);
- memcpy (new_record_run->record->c.data, c->data, irs->case_size);
+ memcpy (new_record_run->record->c.data, c->data, irs->xsrt->scp->case_size);
new_record_run->run = irs->file_idx;
if (irs->last_output != NULL
&& compare_record (c->data, irs->last_output->c.data,
- irs->xsrt->scp) < 0)
+ irs->xsrt->scp, sink->idx_to_fv) < 0)
new_record_run->run = irs->xsrt->next_file_idx;
push_heap (irs->records, irs->record_cnt, sizeof *irs->records,
- compare_record_run_minheap, irs->xsrt->scp);
+ compare_record_run_minheap, irs);
/* Output a record if the reservoir is full. */
if (irs->record_cnt == irs->record_cap && irs->okay)
}
free (irs->records);
- free (irs);
-
if (irs->output_file != NULL)
close_temp_file (irs->xsrt, irs->file_idx, irs->output_file);
+
+ free (irs);
}
/* Allocates room for lots of cases as a buffer. */
static int
allocate_cases (struct initial_run_state *irs)
{
- size_t case_size; /* Size of one case, in bytes. */
int approx_case_cost; /* Approximate memory cost of one case in bytes. */
int max_cases; /* Maximum number of cases to allocate. */
int i;
/* Allocate as many cases as we can within the workspace
limit. */
- case_size = dict_get_case_size (default_dict);
approx_case_cost = (sizeof *irs->records
+ sizeof *irs->free_list
- + case_size
+ + irs->xsrt->scp->case_size
+ 4 * sizeof (void *));
max_cases = set_max_workspace / approx_case_cost;
irs->records = malloc (sizeof *irs->records * max_cases);
for (i = 0; i < max_cases; i++)
{
struct case_list *c;
- c = malloc (sizeof *c + case_size - sizeof (union value));
+ c = malloc (sizeof *c
+ + irs->xsrt->scp->case_size
+ - sizeof (union value));
if (c == NULL)
{
max_cases = i;
A and B, and returns a strcmp()-type result. */
static int
compare_record (const union value *a, const union value *b,
- const struct sort_cases_pgm *scp)
+ const struct sort_cases_pgm *scp,
+ int *idx_to_fv)
{
int i;
for (i = 0; i < scp->var_cnt; i++)
{
struct variable *v = scp->vars[i];
- int fv = v->fv;
+ int fv;
int result;
+ if (idx_to_fv != NULL)
+ fv = idx_to_fv[v->index];
+ else
+ fv = v->fv;
+
if (v->type == NUMERIC)
{
double af = a[fv].f;
static int
compare_record_run (const struct record_run *a,
const struct record_run *b,
- struct sort_cases_pgm *scp)
+ struct initial_run_state *irs)
{
if (a->run != b->run)
return a->run > b->run ? 1 : -1;
else
- return compare_record (a->record->c.data, b->record->c.data, scp);
+ return compare_record (a->record->c.data, b->record->c.data,
+ irs->xsrt->scp, irs->idx_to_fv);
}
/* Compares record-run tuples A and B on run number first, then
on the current record according to SCP, but in descending
order. */
static int
-compare_record_run_minheap (const void *a, const void *b, void *scp)
+compare_record_run_minheap (const void *a, const void *b, void *irs)
{
- return -compare_record_run (a, b, scp);
+ return -compare_record_run (a, b, irs);
}
/* Begins a new initial run, specifically its output file. */
output_record (struct initial_run_state *irs)
{
struct record_run *record_run;
- struct ccase *out_case;
/* Extract minimum case from heap. */
assert (irs->record_cnt > 0);
pop_heap (irs->records, irs->record_cnt--, sizeof *irs->records,
- compare_record_run_minheap, irs->xsrt->scp);
+ compare_record_run_minheap, irs);
record_run = irs->records + irs->record_cnt;
/* Bail if an error has occurred. */
if (!irs->okay)
return;
- /* Obtain case data to write to disk. */
- out_case = &record_run->record->c;
- if (compaction_necessary)
- {
- compact_case (compaction_case, out_case);
- out_case = compaction_case;
- }
-
/* Start new run if necessary. */
assert (record_run->run == irs->file_idx
|| record_run->run == irs->xsrt->next_file_idx);
/* Write to disk. */
if (irs->output_file != NULL
&& !write_temp_file (irs->xsrt, irs->file_idx, irs->output_file,
- out_case->data,
- sizeof *out_case->data * compaction_nval))
+ &record_run->record->c, irs->xsrt->scp->case_size))
irs->okay = 0;
/* This record becomes last_output. */
merge (struct external_sort *xsrt)
{
struct merge_state mrg; /* State of merge. */
- size_t case_size; /* Size of one case, in bytes. */
size_t approx_case_cost; /* Approximate memory cost of one case. */
int max_order; /* Maximum order of merge. */
size_t dummy_run_cnt; /* Number of dummy runs to insert. */
mrg.xsrt = xsrt;
/* Allocate as many cases as possible into cases. */
- case_size = dict_get_case_size (default_dict);
- approx_case_cost = sizeof *mrg.cases + case_size + 4 * sizeof (void *);
+ approx_case_cost = (sizeof *mrg.cases
+ + xsrt->scp->case_size + 4 * sizeof (void *));
mrg.case_cnt = set_max_workspace / approx_case_cost;
mrg.cases = malloc (sizeof *mrg.cases * mrg.case_cnt);
if (mrg.cases == NULL)
goto done;
for (i = 0; i < mrg.case_cnt; i++)
{
- mrg.cases[i] = malloc (case_size);
+ mrg.cases[i] = malloc (xsrt->scp->case_size);
if (mrg.cases[i] == NULL)
{
mrg.case_cnt = i;
max_order = MAX_MERGE_ORDER;
if (mrg.case_cnt / max_order < MIN_BUFFER_SIZE_RECS)
max_order = mrg.case_cnt / MIN_BUFFER_SIZE_RECS;
- else if (mrg.case_cnt / max_order * case_size < MIN_BUFFER_SIZE_BYTES)
- max_order = mrg.case_cnt / (MIN_BUFFER_SIZE_BYTES / case_size);
+ else if (mrg.case_cnt / max_order * xsrt->scp->case_size
+ < MIN_BUFFER_SIZE_BYTES)
+ max_order = mrg.case_cnt / (MIN_BUFFER_SIZE_BYTES / xsrt->scp->case_size);
if (max_order < 2)
max_order = 2;
if (max_order > xsrt->run_cnt)
{
struct run runs[MAX_MERGE_ORDER];
FILE *output_file = NULL;
- size_t case_size;
int success = 0;
int i;
goto error;
/* Merge. */
- case_size = dict_get_case_size (default_dict);
while (run_cnt > 0)
{
struct run *min_run;
for (i = 1; i < run_cnt; i++)
if (compare_record ((*runs[i].buffer_head)->data,
(*min_run->buffer_head)->data,
- mrg->xsrt->scp) < 0)
+ mrg->xsrt->scp, NULL) < 0)
min_run = runs + i;
/* Write minimum to output file. */
if (!write_temp_file (mrg->xsrt, min_run->file_idx, output_file,
- (*min_run->buffer_head)->data, case_size))
+ (*min_run->buffer_head)->data,
+ mrg->xsrt->scp->case_size))
goto error;
/* Remove case from buffer. */
{
if (!read_temp_file (mrg->xsrt, run->file_idx, run->file,
(*run->buffer_tail)->data,
- dict_get_case_size (default_dict)))
+ mrg->xsrt->scp->case_size))
return 0;
run->unread_case_cnt--;
irs->xsrt->scp);
}
-const struct case_sink_class sort_sink_class =
+static const struct case_sink_class sort_sink_class =
{
"SORT CASES",
NULL,
else
value_is_missing = is_missing;
- procedure(common_precalc,common_calc,common_postcalc, NULL);
+ procedure_with_splits (common_precalc, common_calc, common_postcalc, NULL);
switch(mode)
{
case T_1_SAMPLE:
- procedure(one_sample_precalc,one_sample_calc,one_sample_postcalc, NULL);
+ procedure_with_splits (one_sample_precalc, one_sample_calc,
+ one_sample_postcalc, NULL);
break;
case T_PAIRED:
- procedure(paired_precalc,paired_calc,paired_postcalc, NULL);
+ procedure_with_splits (paired_precalc, paired_calc, paired_postcalc,
+ NULL);
break;
case T_IND_SAMPLES:
- procedure(group_precalc,group_calc,group_postcalc, NULL);
+ procedure_with_splits(group_precalc,group_calc,group_postcalc, NULL);
levene(indep_var, cmd.n_variables, cmd.v_variables,
(cmd.miss == TTS_LISTWISE)?LEV_LISTWISE:LEV_ANALYSIS ,
value_is_missing);
#include "value-labels.h"
#include "var.h"
-#include "debug-print.h"
-
int temporary;
struct dictionary *temp_dict;
int temp_trns;
temporary = 1;
temp_dict = dict_clone (default_dict);
temp_trns = n_trns;
- debug_printf (("TEMPORARY: temp_trns=%d\n", temp_trns));
return lex_end_of_command ();
}
int index; /* Index into its dictionary's var[]. */
int type; /* NUMERIC or ALPHA. */
- /* Also important but parse_variables() doesn't need it. Still,
- check before reordering. */
int width; /* Size of string variables in chars. */
int fv, nv; /* Index into `value's, number of values. */
unsigned init : 1; /* 1=VFM must init and possibly reinit. */
int dict_get_next_value_idx (const struct dictionary *);
size_t dict_get_case_size (const struct dictionary *);
+
void dict_compact_values (struct dictionary *);
+size_t dict_get_compacted_value_cnt (const struct dictionary *);
+int *dict_get_compacted_idx_to_fv (const struct dictionary *);
struct variable *const *dict_get_split_vars (const struct dictionary *);
size_t dict_get_split_cnt (const struct dictionary *);
struct variable *var_set_get_var (struct var_set *vs, size_t idx);
struct variable *var_set_lookup_var (struct var_set *vs, const char *name);
void var_set_destroy (struct var_set *vs);
-
\f
/* Variable parsers. */
/* Procedure execution data. */
struct write_case_data
{
- /* 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. */
+ /* Function to call for each case. */
+ int (*proc_func) (struct ccase *, void *); /* Function. */
+ void *aux; /* Auxiliary data. */
- /* Extra auxiliary data. */
- void *aux;
+ 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. */
/* Nonzero if the case needs to have values deleted before being
stored, zero otherwise. */
-int compaction_necessary;
-
-/* Number of values after compaction. */
-int compaction_nval;
-
-/* Temporary case buffer with enough room for `compaction_nval'
- `value's. */
-struct ccase *compaction_case;
+static int compaction_necessary;
/* Nonzero means that we've overflowed our allotted workspace.
After that happens once during a session, we always store the
/* Time at which vfm was last invoked. */
time_t last_vfm_invocation;
-/* Number of cases passed to proc_func(). */
-static int case_count;
-
/* 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 struct ccase *create_trns_case (struct dictionary *dict);
+static struct ccase *create_trns_case (struct dictionary *);
static void open_active_file (void);
-static void close_active_file (struct write_case_data *);
-static int SPLIT_FILE_proc_func (struct ccase *, void *);
-static void finish_compaction (void);
-static void lag_case (const struct ccase *);
-static write_case_func procedure_write_case;
-static void clear_case (struct ccase *);
-static int exclude_this_case (const struct ccase *, int case_num);
+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 compact_case (struct ccase *dest, const struct ccase *src);
+static void clear_case (struct ccase *c);
+static void close_active_file (void);
\f
/* Public functions. */
-/* Auxiliary data for executing a procedure. */
-struct procedure_aux_data
- {
- struct ccase *trns_case; /* Case used for transformations. */
- size_t cases_written; /* Number of cases written so far. */
- };
-
-/* Auxiliary data for SPLIT FILE. */
-struct split_aux_data
- {
- struct ccase *prev_case; /* Data in previous case. */
- };
-
-/* Reads all the cases from the active file, transforms them by
- the active set of transformations, passes each of them to
- PROC_FUNC, and writes them to a new active file.
+/* 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
- Divides the active file into zero or more series of one or more
- cases each. BEGIN_FUNC is called before each series. END_FUNC is
- called after each series.
+ 1. Execute permanent transformations. If these drop the case,
+ start the next case from step 1.
- Arbitrary user-specified data AUX is passed to BEGIN_FUNC,
- PROC_FUNC, and END_FUNC as auxiliary data. */
+ 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 (void (*begin_func) (void *),
- int (*proc_func) (struct ccase *, void *),
- void (*end_func) (void *),
- void *func_aux)
+procedure (int (*proc_func) (struct ccase *, void *), void *aux)
{
static int recursive_call;
- struct write_case_data procedure_write_data;
- struct procedure_aux_data proc_aux;
-
- struct write_case_data split_file_data;
- struct split_aux_data split_aux;
- int split;
+ struct write_case_data wc_data;
assert (++recursive_call == 1);
- proc_aux.cases_written = 0;
- proc_aux.trns_case = create_trns_case (default_dict);
-
- /* Normally we just use the data passed by the user. */
- procedure_write_data.begin_func = begin_func;
- procedure_write_data.proc_func = proc_func;
- procedure_write_data.end_func = end_func;
- procedure_write_data.func_aux = func_aux;
- procedure_write_data.aux = &proc_aux;
-
- /* Under SPLIT FILE, we add a layer of indirection. */
- split = dict_get_split_cnt (default_dict) > 0;
- if (split)
- {
- split_file_data = procedure_write_data;
- split_file_data.aux = &split_aux;
-
- split_aux.prev_case = xmalloc (dict_get_case_size (default_dict));
-
- procedure_write_data.begin_func = NULL;
- procedure_write_data.proc_func = SPLIT_FILE_proc_func;
- procedure_write_data.end_func = end_func;
- procedure_write_data.func_aux = &split_file_data;
- }
+ wc_data.proc_func = proc_func;
+ wc_data.aux = aux;
+ wc_data.trns_case = create_trns_case (default_dict);
+ wc_data.sink_case = xmalloc (dict_get_case_size (default_dict));
+ wc_data.cases_written = 0;
last_vfm_invocation = time (NULL);
open_active_file ();
if (vfm_source != NULL)
vfm_source->class->read (vfm_source,
- proc_aux.trns_case,
- procedure_write_case, &procedure_write_data);
- close_active_file (&procedure_write_data);
-
- if (split)
- free (split_aux.prev_case);
+ wc_data.trns_case,
+ write_case, &wc_data);
+ close_active_file ();
- free (proc_aux.trns_case);
+ free (wc_data.sink_case);
+ free (wc_data.trns_case);
assert (--recursive_call == 0);
}
-\f
-/* Active file processing support. Subtly different semantics from
- procedure(). */
-
-static write_case_func process_active_file_write_case;
-
-/* The case_func might want us to stop calling it. */
-static int not_canceled;
-
-/* Reads all the cases from the active file and passes them
- one-by-one to CASE_FUNC. Before any cases are passed, calls
- BEGIN_FUNC. After all the cases have been passed, calls
- END_FUNC. BEGIN_FUNC, CASE_FUNC, and END_FUNC can write to
- the output file by calling process_active_file_output_case().
-
- process_active_file() ignores TEMPORARY, SPLIT FILE, and N. */
-void
-process_active_file (void (*begin_func) (void *),
- int (*case_func) (struct ccase *, void *),
- void (*end_func) (void *),
- void *func_aux)
-{
- struct procedure_aux_data proc_aux;
- struct write_case_data process_active_write_data;
-
- proc_aux.cases_written = 0;
- proc_aux.trns_case = create_trns_case (default_dict);
-
- process_active_write_data.begin_func = begin_func;
- process_active_write_data.proc_func = case_func;
- process_active_write_data.end_func = end_func;
- process_active_write_data.func_aux = func_aux;
- process_active_write_data.aux = &proc_aux;
-
- not_canceled = 1;
-
- open_active_file ();
- begin_func (func_aux);
- if (vfm_source != NULL)
- vfm_source->class->read (vfm_source, proc_aux.trns_case,
- process_active_file_write_case,
- &process_active_write_data);
- end_func (func_aux);
- close_active_file (&process_active_write_data);
-}
-
-/* Pass the current case to case_func. */
-static int
-process_active_file_write_case (struct write_case_data *wc_data)
-{
- struct procedure_aux_data *proc_aux = wc_data->aux;
- int cur_trns; /* Index of current transformation. */
-
- for (cur_trns = f_trns; cur_trns != temp_trns; )
- {
- int code;
-
- code = t_trns[cur_trns]->proc (t_trns[cur_trns], proc_aux->trns_case,
- case_count + 1);
- switch (code)
- {
- case -1:
- /* Next transformation. */
- cur_trns++;
- break;
- case -2:
- /* Delete this case. */
- goto done;
- default:
- /* Go to that transformation. */
- cur_trns = code;
- break;
- }
- }
-
- if (n_lag)
- lag_case (proc_aux->trns_case);
-
- /* Call the procedure if FILTER and PROCESS IF don't prohibit it. */
- if (not_canceled && !exclude_this_case (proc_aux->trns_case, case_count + 1))
- not_canceled = wc_data->proc_func (proc_aux->trns_case, wc_data->func_aux);
-
- case_count++;
-
- done:
- clear_case (proc_aux->trns_case);
-
- return 1;
-}
-/* Write the given case to the active file. */
-void
-process_active_file_output_case (const struct ccase *c)
-{
- vfm_sink->class->write (vfm_sink, c);
-}
-\f
/* 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. */
}
return c;
}
-\f
-/* Opening the active file. */
-/* It might be usefully noted that the following several functions are
- given in the order that they are called by open_active_file(). */
-
-/* Prepare to write to the replacement active file. */
+/* Makes all preparations for reading from the data source and writing
+ to the data sink. */
static void
-prepare_for_writing (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)
{
- if (workspace_overflow)
- vfm_sink = create_case_sink (&disk_sink_class, NULL);
- else
- vfm_sink = create_case_sink (&memory_sink_class, NULL);
+ int i;
+
+ lag_count = 0;
+ lag_head = 0;
+ lag_queue = xmalloc (n_lag * sizeof *lag_queue);
+ for (i = 0; i < n_lag; i++)
+ lag_queue[i] = xmalloc (dict_get_case_size (temp_dict));
}
+
+ /* Close any unclosed DO IF or LOOP constructs. */
+ discard_ctl_stack ();
}
-/* Arrange for compacting the output cases for storage. */
-static void
-arrange_compaction (void)
+/* 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)
{
- int count_values = 0;
+ /* 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);
- {
- int i;
-
- /* Count up the number of `value's that will be output. */
- for (i = 0; i < dict_get_var_cnt (temp_dict); i++)
- {
- struct variable *v = dict_get_var (temp_dict, i);
-
- if (dict_class_from_id (v->name) != DC_SCRATCH)
- {
- assert (v->nv > 0);
- count_values += v->nv;
- }
- }
- assert (temporary == 2
- || count_values <= dict_get_next_value_idx (temp_dict));
- }
+ /* Write case to replacement active file. */
+ if (vfm_sink->class->write != NULL)
+ {
+ if (compaction_necessary)
+ {
+ compact_case (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);
+ }
- /* Compaction is only necessary if the number of `value's to output
- differs from the number already present. */
- compaction_nval = count_values;
- if (temporary == 2 || count_values != dict_get_next_value_idx (temp_dict))
- compaction_necessary = 1;
- else
- compaction_necessary = 0;
+ /* Execute temporary transformations. */
+ if (!execute_transformations (wc_data->trns_case, t_trns, temp_trns, n_trns,
+ wc_data->cases_written))
+ goto done;
- if (vfm_sink->class->open != NULL)
- vfm_sink->class->open (vfm_sink);
+ /* 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++;
- if (compaction_necessary)
- compaction_case = xmalloc (sizeof (struct ccase)
- + sizeof (union value) * (compaction_nval - 1));
+ /* 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;
}
-#if DEBUGGING
-/* Returns the name of the variable that owns the index CCASE_INDEX
- into ccase. */
-static const char *
-index_to_varname (int ccase_index)
+/* 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 i;
+ int idx;
- for (i = 0; i < default_dict.nvar; i++)
+ for (idx = first_idx; idx != last_idx; )
{
- struct variable *v = default_dict.var[i];
-
- if (ccase_index >= v->fv && ccase_index < v->fv + v->nv)
- return default_dict.var[i]->name;
+ 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 _("<NOVAR>");
+
+ 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_num)
+{
+ /* FILTER. */
+ struct variable *filter_var = dict_get_filter (default_dict);
+ if (filter_var != NULL)
+ {
+ double f = c->data[filter_var->fv].f;
+ if (f == 0.0 || f == SYSMIS || is_num_user_missing (f, filter_var))
+ return 1;
+ }
+
+ /* PROCESS IF. */
+ if (process_if_expr != NULL
+ && expr_evaluate (process_if_expr, c, case_num, NULL) != 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++;
+ memcpy (lag_queue[lag_head], c, dict_get_case_size (temp_dict));
+ if (++lag_head >= n_lag)
+ lag_head = 0;
}
-#endif
-/* Sets all the lag-related variables based on value of n_lag. */
+/* Copies case SRC to case DEST, compacting it in the process. */
static void
-setup_lag (void)
+compact_case (struct ccase *dest, const struct ccase *src)
{
int i;
+ int nval = 0;
+ size_t var_cnt;
- if (n_lag == 0)
- return;
+ assert (compaction_necessary);
+
+ /* Copy all the variables except scratch variables from SRC to
+ DEST. */
+ var_cnt = dict_get_var_cnt (default_dict);
+ for (i = 0; i < var_cnt; i++)
+ {
+ struct variable *v = dict_get_var (default_dict, i);
+
+ if (dict_class_from_id (v->name) == DC_SCRATCH)
+ continue;
- lag_count = 0;
- lag_head = 0;
- lag_queue = xmalloc (n_lag * sizeof *lag_queue);
- for (i = 0; i < n_lag; i++)
- lag_queue[i] = xmalloc (dict_get_case_size (temp_dict));
+ if (v->type == NUMERIC)
+ dest->data[nval++] = src->data[v->fv];
+ else
+ {
+ int w = DIV_RND_UP (v->width, sizeof (union value));
+
+ memcpy (&dest->data[nval], &src->data[v->fv], w * sizeof (union value));
+ nval += w;
+ }
+ }
}
-/* There is a lot of potential confusion in the vfm and related
- routines over the number of `value's at each stage of the process.
- Here is each nval count, with explanation, as set up by
- open_active_file():
-
- temp_dict->nval: Number of `value's in the cases after the
- transformations leading up to TEMPORARY have been performed.
-
- compaction_nval: Number of `value's in the cases after the
- transformations leading up to TEMPORARY have been performed
- and the case has been compacted by compact_case(), if
- compaction is necessary. This the number of `value's in the
- cases saved by the sink stream. (However, note that the cases
- passed to the sink stream have not yet been compacted. It is
- the responsibility of the data sink to call compact_case().)
- `compaction' becomes the new value of default_dict.nval after
- the procedure is completed.
-
- default_dict.nval: This is often an alias for temp_dict->nval.
- As such it can really have no separate existence until the
- procedure is complete. For this reason it should *not* be
- referenced inside the execution of a procedure. */
-/* Makes all preparations for reading from the data source and writing
- to the data sink. */
+/* Clears the variables in C that need to be cleared between
+ processing cases. */
static void
-open_active_file (void)
+clear_case (struct ccase *c)
{
- /* Sometimes we want to refer to the dictionary that applies to the
- data actually written to the sink. This is either temp_dict or
- default_dict. However, if TEMPORARY is not on, then temp_dict
- does not apply. So, we can set temp_dict to default_dict in this
- case. */
- if (!temporary)
+ size_t var_cnt = dict_get_var_cnt (default_dict);
+ size_t i;
+
+ for (i = 0; i < var_cnt; i++)
{
- temp_trns = n_trns;
- temp_dict = default_dict;
+ struct variable *v = dict_get_var (default_dict, i);
+ if (v->init && v->reinit)
+ {
+ if (v->type == NUMERIC)
+ c->data[v->fv].f = SYSMIS;
+ else
+ memset (c->data[v->fv].s, ' ', v->width);
+ }
}
-
- /* No cases passed to the procedure yet. */
- case_count = 0;
-
- /* The rest. */
- prepare_for_writing ();
- arrange_compaction ();
- discard_ctl_stack ();
- setup_lag ();
}
-\f
+
/* Closes the active file. */
static void
-close_active_file (struct write_case_data *data)
+close_active_file (void)
{
- /* Close the current case group. */
- if (case_count && data->end_func != NULL)
- data->end_func (data->func_aux);
-
- /* Stop lagging (catch up?). */
- if (n_lag)
+ /* Free memory for lag queue, and turn off lagging. */
+ if (n_lag > 0)
{
int i;
n_lag = 0;
}
- /* Assume the dictionary from right before TEMPORARY, if any. Turn
- off TEMPORARY. */
+ /* Dictionary from before TEMPORARY becomes permanent.. */
if (temporary)
{
dict_destroy (default_dict);
/* Finish compaction. */
if (compaction_necessary)
- finish_compaction ();
+ dict_compact_values (default_dict);
- /* Old data sink --> New data source. */
+ /* Free data source. */
if (vfm_source != NULL)
{
if (vfm_source->class->destroy != NULL)
free (vfm_source);
}
+ /* Old data sink becomes new data source. */
if (vfm_sink->class->make_source != NULL)
vfm_source = vfm_sink->class->make_source (vfm_sink);
else
vfm_source = NULL;
-
- /* Old data sink is gone now. */
- free (vfm_sink);
+ free_case_sink (vfm_sink);
vfm_sink = NULL;
- /* Cancel TEMPORARY. */
+ /* Cancel TEMPORARY, PROCESS IF, FILTER, N OF CASES, vectors,
+ and get rid of all the transformations. */
cancel_temporary ();
-
- /* Free temporary cases. */
- free (compaction_case);
- compaction_case = NULL;
-
- /* Cancel PROCESS IF. */
expr_free (process_if_expr);
process_if_expr = NULL;
-
- /* Cancel FILTER if temporary. */
if (dict_get_filter (default_dict) != NULL && !FILTER_before_TEMPORARY)
dict_set_filter (default_dict, NULL);
-
- /* Cancel transformations. */
- cancel_transformations ();
-
- /* Turn off case limiter. */
dict_set_case_limit (default_dict, 0);
-
- /* Clear VECTOR vectors. */
dict_clear_vectors (default_dict);
+ cancel_transformations ();
}
\f
-/* Disk case stream. */
+/* Storage case stream. */
-/* Information about disk sink or source. */
-struct disk_stream_info
+/* Information about storage sink or source. */
+struct storage_stream_info
{
- FILE *file; /* Output file. */
- size_t case_cnt; /* Number of cases written so far. */
+ size_t case_cnt; /* Number of cases. */
size_t case_size; /* Number of bytes in case. */
+ enum { DISK, MEMORY } mode; /* Where is data stored? */
+
+ /* Disk storage. */
+ FILE *file; /* Data file. */
+
+ /* Memory storage. */
+ int max_cases; /* Maximum cases before switching to disk. */
+ struct case_list *head; /* First case in list. */
+ struct case_list *tail; /* Last case in list. */
};
-/* Initializes the disk sink. */
+static void open_storage_file (struct storage_stream_info *info);
+
+/* Initializes a storage sink. */
static void
-disk_sink_create (struct case_sink *sink)
+storage_sink_open (struct case_sink *sink)
{
- struct disk_stream_info *info = xmalloc (sizeof *info);
- info->file = tmpfile ();
+ struct storage_stream_info *info;
+
+ sink->aux = info = xmalloc (sizeof *info);
info->case_cnt = 0;
- info->case_size = compaction_nval;
- sink->aux = info;
- if (info->file == NULL)
+ info->case_size = sink->value_cnt * sizeof (union value);
+ info->file = NULL;
+ info->max_cases = 0;
+ info->head = info->tail = NULL;
+ if (workspace_overflow)
{
- msg (ME, _("An error occurred attempting to create a temporary "
- "file for use as the active file: %s."),
- strerror (errno));
- err_failure ();
+ info->mode = DISK;
+ open_storage_file (info);
+ }
+ else
+ {
+ info->mode = MEMORY;
+ info->max_cases = (set_max_workspace
+ / (sizeof (struct case_list) + info->case_size));
}
}
-/* Writes case C to the disk sink. */
+/* Creates a new temporary file and puts it into INFO. */
static void
-disk_sink_write (struct case_sink *sink, const struct ccase *c)
+open_storage_file (struct storage_stream_info *info)
{
- struct disk_stream_info *info = sink->aux;
- const union value *src_case;
-
- if (compaction_necessary)
+ info->file = tmpfile ();
+ if (info->file == NULL)
{
- compact_case (compaction_case, c);
- src_case = compaction_case->data;
+ msg (ME, _("An error occurred creating a temporary "
+ "file for use as the active file: %s."),
+ strerror (errno));
+ err_failure ();
}
- else src_case = c->data;
+}
- info->case_cnt++;
- if (fwrite (src_case, sizeof *src_case * compaction_nval, 1,
- info->file) != 1)
+/* Writes the VALUE_CNT values in VALUES to FILE. */
+static void
+write_storage_file (FILE *file, const union value *values, size_t value_cnt)
+{
+ if (fwrite (values, sizeof *values * value_cnt, 1, file) != 1)
{
- msg (ME, _("An error occurred while attempting to write to a "
+ msg (ME, _("An error occurred writing to a "
"temporary file used as the active file: %s."),
strerror (errno));
err_failure ();
}
}
-/* Destroys the sink's internal data. */
+/* If INFO represents records in memory, moves them to disk.
+ Each comprises VALUE_CNT `union value's. */
static void
-disk_sink_destroy (struct case_sink *sink)
+storage_to_disk (struct storage_stream_info *info, size_t value_cnt)
{
- struct disk_stream_info *info = sink->aux;
- if (info->file != NULL)
- fclose (info->file);
+ struct case_list *cur, *next;
+
+ if (info->mode == MEMORY)
+ {
+ info->mode = DISK;
+ open_storage_file (info);
+ for (cur = info->head; cur; cur = next)
+ {
+ next = cur->next;
+ write_storage_file (info->file, cur->c.data, value_cnt);
+ free (cur);
+ }
+ info->head = info->tail = NULL;
+ }
}
-/* Closes and destroys the sink and returns a disk source to read
- back the written data. */
-static struct case_source *
-disk_sink_make_source (struct case_sink *sink)
+/* Destroys storage stream represented by INFO. */
+static void
+destroy_storage_stream_info (struct storage_stream_info *info)
{
- struct disk_stream_info *info = sink->aux;
-
- /* Rewind the file. */
- assert (info->file != NULL);
- if (fseek (info->file, 0, SEEK_SET) != 0)
+ if (info->mode == DISK)
{
- msg (ME, _("An error occurred while attempting to rewind a "
- "temporary file used as the active file: %s."),
- strerror (errno));
- err_failure ();
+ if (info->file != NULL)
+ fclose (info->file);
}
+ else
+ {
+ struct case_list *cur, *next;
- return create_case_source (&disk_source_class, default_dict, info);
-}
-
-/* Disk sink. */
-const struct case_sink_class disk_sink_class =
- {
- "disk",
- disk_sink_create,
- disk_sink_write,
- disk_sink_destroy,
- disk_sink_make_source,
- };
-\f
-/* Disk source. */
-
-/* Returns the number of cases that will be read by
- disk_source_read(). */
-static int
-disk_source_count (const struct case_source *source)
-{
- struct disk_stream_info *info = source->aux;
-
- return info->case_cnt;
-}
-
-/* Reads all cases from the disk source and passes them one by one to
- write_case(). */
-static void
-disk_source_read (struct case_source *source,
- struct ccase *c,
- write_case_func *write_case, write_case_data wc_data)
-{
- struct disk_stream_info *info = source->aux;
- int i;
-
- for (i = 0; i < info->case_cnt; i++)
- {
- if (!fread (c, info->case_size, 1, info->file))
- {
- msg (ME, _("An error occurred while attempting to read from "
- "a temporary file created for the active file: %s."),
- strerror (errno));
- err_failure ();
- break;
- }
-
- if (!write_case (wc_data))
- break;
+ for (cur = info->head; cur; cur = next)
+ {
+ next = cur->next;
+ free (cur);
+ }
}
+ free (info);
}
-/* Destroys the source's internal data. */
-static void
-disk_source_destroy (struct case_source *source)
-{
- struct disk_stream_info *info = source->aux;
- if (info->file != NULL)
- fclose (info->file);
- free (info);
-}
-
-/* Disk source. */
-const struct case_source_class disk_source_class =
- {
- "disk",
- disk_source_count,
- disk_source_read,
- disk_source_destroy,
- };
-\f
-/* Memory case stream. */
-
-/* Memory sink data. */
-struct memory_sink_info
- {
- size_t case_cnt; /* Number of cases. */
- size_t case_size; /* Case size in bytes. */
- int max_cases; /* Maximum cases before switching to disk. */
- struct case_list *head; /* First case in list. */
- struct case_list *tail; /* Last case in list. */
- };
-
-/* Memory source data. */
-struct memory_source_info
- {
- size_t case_cnt; /* Number of cases. */
- size_t case_size; /* Case size in bytes. */
- struct case_list *cases; /* List of cases. */
- };
-
-/* Creates the SINK memory sink. */
-static void
-memory_sink_create (struct case_sink *sink)
-{
- struct memory_sink_info *info;
-
- sink->aux = info = xmalloc (sizeof *info);
-
- assert (compaction_nval > 0);
- info->case_cnt = 0;
- info->case_size = compaction_nval * sizeof (union value);
- info->max_cases = set_max_workspace / info->case_size;
- info->head = info->tail = NULL;
-}
-
-/* Writes case C to memory sink SINK. */
+/* Writes case C to the storage sink SINK. */
static void
-memory_sink_write (struct case_sink *sink, const struct ccase *c)
+storage_sink_write (struct case_sink *sink, const struct ccase *c)
{
- struct memory_sink_info *info = sink->aux;
- size_t case_size;
- struct case_list *new_case;
+ struct storage_stream_info *info = sink->aux;
- case_size = sizeof (struct case_list)
- + ((compaction_nval - 1) * sizeof (union value));
- new_case = malloc (case_size);
-
- /* If we've got memory to spare then add it to the linked list. */
- if (info->case_cnt <= info->max_cases && new_case != NULL)
+ info->case_cnt++;
+ if (info->mode == MEMORY)
{
- info->case_cnt++;
+ struct case_list *new_case;
+
+ /* Copy case. */
+ new_case = xmalloc (sizeof (struct case_list)
+ + ((sink->value_cnt - 1) * sizeof (union value)));
+ memcpy (&new_case->c, c, sizeof (union value) * sink->value_cnt);
/* Append case to linked list. */
new_case->next = NULL;
info->head = new_case;
info->tail = new_case;
- /* Copy data into case. */
- if (compaction_necessary)
- compact_case (&new_case->c, c);
- else
- memcpy (&new_case->c, c, sizeof (union value) * compaction_nval);
- }
- else
- {
- /* Out of memory. Write the active file to disk. */
- struct case_list *cur, *next;
-
- /* Notify the user. */
- if (!new_case)
- msg (MW, _("Virtual memory exhausted. Writing active file "
- "to disk."));
- else
- msg (MW, _("Workspace limit of %d KB (%d cases at %d bytes each) "
- "overflowed. Writing active file to disk."),
- set_max_workspace / 1024, info->max_cases,
- compaction_nval * sizeof (union value));
-
- free (new_case);
-
- /* Switch to a disk sink. */
- vfm_sink = create_case_sink (&disk_sink_class, NULL);
- vfm_sink->class->open (vfm_sink);
- workspace_overflow = 1;
-
- /* Write the cases to disk and destroy them. We can't call
- vfm->sink->write() because of compaction. */
- for (cur = info->head; cur; cur = next)
- {
- next = cur->next;
- if (fwrite (cur->c.data, sizeof (union value) * compaction_nval, 1,
- vfm_sink->aux) != 1)
- {
- msg (ME, _("An error occurred while attempting to "
- "write to a temporary file created as the "
- "active file: %s."),
- strerror (errno));
- err_failure ();
- }
- free (cur);
- }
-
- /* Write the current case to disk. */
- vfm_sink->class->write (vfm_sink, c);
- }
-}
-
-/* If the data is stored in memory, causes it to be written to disk.
- To be called only *between* procedure()s, not within them. */
-void
-write_active_file_to_disk (void)
-{
- if (case_source_is_class (vfm_source, &memory_source_class))
- {
- struct memory_source_info *info = vfm_source->aux;
+ /* Dump all the cases to disk if we've run out of
+ workspace. */
+ if (info->case_cnt > info->max_cases)
+ {
+ workspace_overflow = 1;
+ msg (MW, _("Workspace limit of %d KB (%d cases at %d bytes each) "
+ "overflowed. Writing active file to disk."),
+ set_max_workspace / 1024, info->max_cases,
+ sizeof (struct case_list) + info->case_size);
- /* Switch to a disk sink. */
- vfm_sink = create_case_sink (&disk_sink_class, NULL);
- vfm_sink->class->open (vfm_sink);
- workspace_overflow = 1;
-
- /* Write the cases to disk and destroy them. We can't call
- vfm->sink->write() because of compaction. */
- {
- struct case_list *cur, *next;
-
- for (cur = info->cases; cur; cur = next)
- {
- next = cur->next;
- if (fwrite (cur->c.data, sizeof *cur->c.data * compaction_nval, 1,
- vfm_sink->aux) != 1)
- {
- msg (ME, _("An error occurred while attempting to "
- "write to a temporary file created as the "
- "active file: %s."),
- strerror (errno));
- err_failure ();
- }
- free (cur);
- }
- }
-
- vfm_source = vfm_sink->class->make_source (vfm_sink);
- vfm_sink = NULL;
+ storage_to_disk (info, sink->value_cnt);
+ }
}
+ else
+ write_storage_file (info->file, c->data, sink->value_cnt);
}
-/* Destroy all memory sink data. */
+/* Destroys internal data in SINK. */
static void
-memory_sink_destroy (struct case_sink *sink)
+storage_sink_destroy (struct case_sink *sink)
{
- struct memory_sink_info *info = sink->aux;
- struct case_list *cur, *next;
-
- for (cur = info->head; cur; cur = next)
- {
- next = cur->next;
- free (cur);
- }
- free (info);
+ destroy_storage_stream_info (sink->aux);
}
-/* Switch the memory stream from sink to source mode. */
+/* Closes and destroys the sink and returns a storage source to
+ read back the written data. */
static struct case_source *
-memory_sink_make_source (struct case_sink *sink)
+storage_sink_make_source (struct case_sink *sink)
{
- struct memory_sink_info *sink_info = sink->aux;
- struct memory_source_info *source_info;
-
- source_info = xmalloc (sizeof *source_info);
- source_info->case_cnt = sink_info->case_cnt;
- source_info->case_size = sink_info->case_size;
- source_info->cases = sink_info->head;
+ struct storage_stream_info *info = sink->aux;
- free (sink_info);
+ if (info->mode == DISK)
+ {
+ /* Rewind the file. */
+ assert (info->file != NULL);
+ if (fseek (info->file, 0, SEEK_SET) != 0)
+ {
+ msg (ME, _("An error occurred while attempting to rewind a "
+ "temporary file used as the active file: %s."),
+ strerror (errno));
+ err_failure ();
+ }
+ }
- return create_case_source (&memory_source_class,
- default_dict, source_info);
+ return create_case_source (&storage_source_class, sink->dict, info);
}
-const struct case_sink_class memory_sink_class =
+/* Storage sink. */
+const struct case_sink_class storage_sink_class =
{
- "memory",
- memory_sink_create,
- memory_sink_write,
- memory_sink_destroy,
- memory_sink_make_source,
+ "storage",
+ storage_sink_open,
+ storage_sink_write,
+ storage_sink_destroy,
+ storage_sink_make_source,
};
+\f
+/* Storage source. */
-/* Returns the number of cases in the source. */
+/* Returns the number of cases that will be read by
+ storage_source_read(). */
static int
-memory_source_count (const struct case_source *source)
+storage_source_count (const struct case_source *source)
{
- struct memory_source_info *info = source->aux;
+ struct storage_stream_info *info = source->aux;
return info->case_cnt;
}
-/* Reads the case stream from memory and passes it to write_case(). */
+/* Reads all cases from the storage source and passes them one by one to
+ write_case(). */
static void
-memory_source_read (struct case_source *source,
- struct ccase *c,
- write_case_func *write_case, write_case_data wc_data)
+storage_source_read (struct case_source *source,
+ struct ccase *c,
+ write_case_func *write_case, write_case_data wc_data)
{
- struct memory_source_info *info = source->aux;
+ struct storage_stream_info *info = source->aux;
- while (info->cases != NULL)
+ if (info->mode == DISK)
{
- struct case_list *iter = info->cases;
- memcpy (c, &iter->c, info->case_size);
- if (!write_case (wc_data))
- break;
+ int i;
+
+ for (i = 0; i < info->case_cnt; i++)
+ {
+ if (!fread (c, info->case_size, 1, info->file))
+ {
+ msg (ME, _("An error occurred while attempting to read from "
+ "a temporary file created for the active file: %s."),
+ strerror (errno));
+ err_failure ();
+ break;
+ }
+
+ if (!write_case (wc_data))
+ break;
+ }
+ }
+ else
+ {
+ while (info->head != NULL)
+ {
+ struct case_list *iter = info->head;
+ memcpy (c, &iter->c, info->case_size);
+ if (!write_case (wc_data))
+ break;
- info->cases = iter->next;
- free (iter);
+ info->head = iter->next;
+ free (iter);
+ }
+ info->tail = NULL;
}
}
-/* Destroy all memory source data. */
+/* Destroys the source's internal data. */
static void
-memory_source_destroy (struct case_source *source)
+storage_source_destroy (struct case_source *source)
{
- struct memory_source_info *info = source->aux;
- struct case_list *cur, *next;
-
- for (cur = info->cases; cur; cur = next)
- {
- next = cur->next;
- free (cur);
- }
- free (info);
+ destroy_storage_stream_info (source->aux);
}
-/* Returns the list of cases in memory source SOURCE. */
+/* Storage source. */
+const struct case_source_class storage_source_class =
+ {
+ "storage",
+ storage_source_count,
+ storage_source_read,
+ storage_source_destroy,
+ };
+
+/* Returns nonzero only if SOURCE is stored on disk (instead of
+ in memory). */
+int
+storage_source_on_disk (const struct case_source *source)
+{
+ struct storage_stream_info *info = source->aux;
+
+ return info->mode == DISK;
+}
+
+/* Returns the list of cases in storage source SOURCE. */
struct case_list *
-memory_source_get_cases (const struct case_source *source)
+storage_source_get_cases (const struct case_source *source)
{
- struct memory_source_info *info = source->aux;
+ struct storage_stream_info *info = source->aux;
- return info->cases;
+ assert (info->mode == MEMORY);
+ return info->head;
}
/* Sets the list of cases in memory source SOURCE to CASES. */
void
-memory_source_set_cases (const struct case_source *source,
- struct case_list *cases)
+storage_source_set_cases (const struct case_source *source,
+ struct case_list *cases)
{
- struct memory_source_info *info = source->aux;
+ struct storage_stream_info *info = source->aux;
- info->cases = cases;
+ assert (info->mode == MEMORY);
+ info->head = cases;
}
-/* Memory stream. */
-const struct case_source_class memory_source_class =
- {
- "memory",
- memory_source_count,
- memory_source_read,
- memory_source_destroy,
- };
-\f
-/* Add C to the lag queue. */
-static void
-lag_case (const struct ccase *c)
+/* If SOURCE has its cases in memory, writes them to disk. */
+void
+storage_source_to_disk (struct case_source *source)
{
- if (lag_count < n_lag)
- lag_count++;
- memcpy (lag_queue[lag_head], c, dict_get_case_size (temp_dict));
- if (++lag_head >= n_lag)
- lag_head = 0;
+ struct storage_stream_info *info = source->aux;
+
+ storage_to_disk (info, source->value_cnt);
}
+\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 *
}
}
-/* 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. */
-int
-procedure_write_case (write_case_data wc_data)
-{
- struct procedure_aux_data *proc_aux = wc_data->aux;
-
- /* Index of current transformation. */
- int cur_trns;
-
- /* Return value: whether it's reasonable to write any more cases. */
- int more_cases = 1;
-
- cur_trns = f_trns;
- for (;;)
- {
- /* Output the case if this is temp_trns. */
- if (cur_trns == temp_trns)
- {
- int case_limit;
-
- if (n_lag)
- lag_case (proc_aux->trns_case);
-
- vfm_sink->class->write (vfm_sink, proc_aux->trns_case);
-
- proc_aux->cases_written++;
- case_limit = dict_get_case_limit (default_dict);
- if (case_limit != 0 && proc_aux->cases_written >= case_limit)
- more_cases = 0;
- }
-
- /* Are we done? */
- if (cur_trns >= n_trns)
- break;
-
- /* Decide which transformation should come next. */
- {
- int code;
-
- code = t_trns[cur_trns]->proc (t_trns[cur_trns], proc_aux->trns_case,
- proc_aux->cases_written + 1);
- switch (code)
- {
- case -1:
- /* Next transformation. */
- cur_trns++;
- break;
- case -2:
- /* Delete this case. */
- goto done;
- default:
- /* Go to that transformation. */
- cur_trns = code;
- break;
- }
- }
- }
-
- /* Call the beginning of group function. */
- if (!case_count && wc_data->begin_func != NULL)
- wc_data->begin_func (wc_data->func_aux);
-
- /* Call the procedure if there is one and FILTER and PROCESS IF
- don't prohibit it. */
- if (wc_data->proc_func != NULL
- && !exclude_this_case (proc_aux->trns_case, proc_aux->cases_written + 1))
- wc_data->proc_func (proc_aux->trns_case, wc_data->func_aux);
-
- case_count++;
-
-done:
- clear_case (proc_aux->trns_case);
-
- /* Return previously determined value. */
- return more_cases;
-}
-
-/* Clears the variables in C that need to be cleared between
- processing cases. */
-static void
-clear_case (struct ccase *c)
-{
- /* FIXME? This is linear in the number of variables, but
- doesn't need to be, so it's an easy optimization target. */
- 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)
- c->data[v->fv].f = SYSMIS;
- else
- memset (c->data[v->fv].s, ' ', v->width);
- }
- }
-}
-
-/* Returns nonzero if case C with case number CASE_NUM should be
- exclude as specified on FILTER or PROCESS IF, otherwise
- zero. */
-static int
-exclude_this_case (const struct ccase *c, int case_num)
-{
- /* FILTER. */
- struct variable *filter_var = dict_get_filter (default_dict);
- if (filter_var != NULL)
- {
- double f = c->data[filter_var->fv].f;
- if (f == 0.0 || f == SYSMIS || is_num_user_missing (f, filter_var))
- return 1;
- }
-
- /* PROCESS IF. */
- if (process_if_expr != NULL
- && expr_evaluate (process_if_expr, c, case_num, NULL) != 1.0)
- return 1;
-
- return 0;
-}
-
/* Appends TRNS to t_trns[], the list of all transformations to be
performed on data as it is read from the active file. */
void
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,
+ const struct dictionary *dict,
+ void *aux)
+{
+ struct case_source *source = xmalloc (sizeof *source);
+ source->class = class;
+ source->value_cnt = dict_get_next_value_idx (dict);
+ source->aux = aux;
+ return source;
+}
-/* Dumps out the values of all the split variables for the case C. */
-static void
-dump_splits (struct ccase *c)
+/* Returns nonzero if a case source is "complex". */
+int
+case_source_is_complex (const struct case_source *source)
{
- struct variable *const *split;
- struct tab_table *t;
- size_t split_cnt;
- int i;
+ return source != NULL && (source->class == &input_program_source_class
+ || source->class == &file_type_source_class);
+}
- split_cnt = dict_get_split_cnt (default_dict);
- 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;
+/* 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;
+}
- 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, &c->data[v->fv]);
-
- temp_buf[v->print.w] = 0;
- tab_text (t, 1, i + 1, TAT_PRINTF, "%.*s", v->print.w, temp_buf);
+/* Creates a case sink 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->dict = dict;
+ sink->idx_to_fv = dict_get_compacted_idx_to_fv (dict);
+ sink->value_cnt = dict_get_compacted_value_cnt (dict);
+ sink->aux = aux;
+ return sink;
+}
- val_lab = val_labs_find (v->val_labs, c->data[v->fv]);
- if (val_lab)
- tab_text (t, 2, i + 1, TAB_LEFT, val_lab);
- }
- tab_flags (t, SOMF_NO_TITLE);
- tab_submit (t);
+/* Destroys case sink SINK. It is the caller's responsible to
+ call the sink's destroy function, if any. */
+void
+free_case_sink (struct case_sink *sink)
+{
+ free (sink->idx_to_fv);
+ 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;
+ split_aux.prev_case = xmalloc (dict_get_case_size (default_dict));
+ split_aux.begin_func = begin_func;
+ split_aux.proc_func = proc_func;
+ split_aux.end_func = end_func;
+ split_aux.func_aux = func_aux;
+
+ procedure (procedure_with_splits_callback, &split_aux);
+
+ if (split_aux.case_count > 0 && end_func != NULL)
+ end_func (func_aux);
+ free (split_aux.prev_case);
}
-/* This proc_func is substituted for the user-supplied proc_func when
- SPLIT FILE is active. This function forms a wrapper around that
- proc_func by dividing the input into series. */
+/* procedure() callback used by procedure_with_splits(). */
static int
-SPLIT_FILE_proc_func (struct ccase *c, void *data_)
+procedure_with_splits_callback (struct ccase *c, void *split_aux_)
{
- struct write_case_data *data = data_;
- struct split_aux_data *split_aux = data->aux;
- struct variable *const *split;
- size_t split_cnt;
- size_t i;
+ struct split_aux_data *split_aux = split_aux_;
- /* The first case always begins a new series. We also need to
- preserve the values of the case for later comparison. */
- if (case_count == 0)
+ /* Start a new series if needed. */
+ if (split_aux->case_count == 0
+ || !equal_splits (c, split_aux->prev_case))
{
- memcpy (split_aux->prev_case, c, dict_get_case_size (default_dict));
+ if (split_aux->case_count > 0 && split_aux->end_func != NULL)
+ split_aux->end_func (split_aux->func_aux);
dump_splits (c);
- if (data->begin_func != NULL)
- data->begin_func (data->func_aux);
-
- return data->proc_func (c, data->func_aux);
+ memcpy (split_aux->prev_case, c, dict_get_case_size (default_dict));
+
+ if (split_aux->begin_func != NULL)
+ split_aux->begin_func (split_aux->func_aux);
}
- /* Compare the value of each SPLIT FILE variable to the values on
- the previous case. */
+ 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)
+{
+ struct variable *const *split;
+ size_t split_cnt;
+ size_t i;
+
split = dict_get_split_vars (default_dict);
split_cnt = dict_get_split_cnt (default_dict);
for (i = 0; i < split_cnt; i++)
switch (v->type)
{
case NUMERIC:
- if (c->data[v->fv].f != split_aux->prev_case->data[v->fv].f)
- goto not_equal;
+ if (a->data[v->fv].f != b->data[v->fv].f)
+ return 0;
break;
case ALPHA:
- if (memcmp (c->data[v->fv].s,
- split_aux->prev_case->data[v->fv].s, v->width))
- goto not_equal;
+ if (memcmp (a->data[v->fv].s, b->data[v->fv].s, v->width))
+ return 0;
break;
default:
assert (0);
}
}
- return data->proc_func (c, data->func_aux);
-
-not_equal:
- /* The values of the SPLIT FILE variable are different from the
- values on the previous case. That means that it's time to begin
- a new series. */
- if (data->end_func != NULL)
- data->end_func (data->func_aux);
- dump_splits (c);
- if (data->begin_func != NULL)
- data->begin_func (data->func_aux);
- memcpy (split_aux->prev_case, c, dict_get_case_size (default_dict));
- return data->proc_func (c, data->func_aux);
-}
-\f
-/* Case compaction. */
-
-/* Copies case SRC to case DEST, compacting it in the process. */
-void
-compact_case (struct ccase *dest, const struct ccase *src)
-{
- int i;
- int nval = 0;
- size_t var_cnt;
-
- assert (compaction_necessary);
-
- if (temporary == 2)
- {
- if (dest != compaction_case)
- memcpy (dest, compaction_case, sizeof (union value) * compaction_nval);
- return;
- }
- /* Copy all the variables except the scratch variables from SRC to
- DEST. */
- var_cnt = dict_get_var_cnt (default_dict);
- for (i = 0; i < var_cnt; i++)
- {
- struct variable *v = dict_get_var (default_dict, i);
-
- if (dict_class_from_id (v->name) == DC_SCRATCH)
- continue;
-
- if (v->type == NUMERIC)
- dest->data[nval++] = src->data[v->fv];
- else
- {
- int w = DIV_RND_UP (v->width, sizeof (union value));
-
- memcpy (&dest->data[nval], &src->data[v->fv], w * sizeof (union value));
- nval += w;
- }
- }
+ return 1;
}
-/* Reassigns `fv' for each variable. Deletes scratch variables. */
+/* Dumps out the values of all the split variables for the case C. */
static void
-finish_compaction (void)
+dump_splits (struct ccase *c)
{
+ struct variable *const *split;
+ struct tab_table *t;
+ size_t split_cnt;
int i;
- for (i = 0; i < dict_get_var_cnt (default_dict); )
- {
- struct variable *v = dict_get_var (default_dict, i);
-
- if (dict_class_from_id (v->name) == DC_SCRATCH)
- dict_delete_var (default_dict, v);
- else
- i++;
- }
- dict_compact_values (default_dict);
-}
-
-/* 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,
- const struct dictionary *dict,
- void *aux)
-{
- struct case_source *source = xmalloc (sizeof *source);
- source->class = class;
- source->value_cnt = dict_get_next_value_idx (dict);
- source->aux = aux;
- return source;
-}
+ split_cnt = dict_get_split_cnt (default_dict);
+ if (split_cnt == 0)
+ return;
-/* 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);
-}
+ 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;
-/* 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;
-}
+ 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, &c->data[v->fv]);
+
+ temp_buf[v->print.w] = 0;
+ tab_text (t, 1, i + 1, TAT_PRINTF, "%.*s", v->print.w, temp_buf);
-/* Creates a case sink with class CLASS and auxiliary data
- AUX. */
-struct case_sink *
-create_case_sink (const struct case_sink_class *class, void *aux)
-{
- struct case_sink *sink = xmalloc (sizeof *sink);
- sink->class = class;
- sink->aux = aux;
- return sink;
+ val_lab = val_labs_find (v->val_labs, c->data[v->fv]);
+ if (val_lab)
+ tab_text (t, 2, i + 1, TAB_LEFT, val_lab);
+ }
+ tab_flags (t, SOMF_NO_TITLE);
+ tab_submit (t);
}
void (*destroy) (struct case_source *);
};
-extern const struct case_source_class memory_source_class;
-extern const struct case_source_class disk_source_class;
+extern const struct case_source_class storage_source_class;
extern const struct case_source_class data_list_source_class;
extern const struct case_source_class file_type_source_class;
extern const struct case_source_class input_program_source_class;
int case_source_is_complex (const struct case_source *);
int case_source_is_class (const struct case_source *,
const struct case_source_class *);
-struct case_list *memory_source_get_cases (const struct case_source *);
-void memory_source_set_cases (const struct case_source *,
- struct case_list *);
+
+int storage_source_on_disk (const struct case_source *);
+struct case_list *storage_source_get_cases (const struct case_source *);
+void storage_source_set_cases (const struct case_source *,
+ struct case_list *);
+void storage_source_to_disk (struct case_source *source);
\f
/* The replacement active file, to which cases are written. */
extern struct case_sink *vfm_sink;
struct case_sink
{
const struct case_sink_class *class; /* Class. */
- void *aux; /* Auxiliary data. */
+ void *aux; /* Auxiliary data. */
+
+ /* Cases written to a case sink belong to a dictionary, but
+ their data is compacted to drop scratch variables. */
+ const struct dictionary *dict; /* Dictionary for cases. */
+ int *idx_to_fv; /* `dict' index -> case `data' index. */
+ size_t value_cnt; /* Number of `union value's in case. */
};
/* A case sink class. */
{
const char *name; /* Identifying name. */
- /* Creates the sink and opens it for writing. */
+ /* Opens the sink for writing. */
void (*open) (struct case_sink *);
/* Writes a case to the sink. */
struct case_source *(*make_source) (struct case_sink *);
};
-extern const struct case_sink_class memory_sink_class;
-extern const struct case_sink_class disk_sink_class;
-extern const struct case_sink_class sort_sink_class;
+extern const struct case_sink_class storage_sink_class;
+extern const struct case_sink_class null_sink_class;
-struct case_sink *create_case_sink (const struct case_sink_class *, void *);
+struct case_sink *create_case_sink (const struct case_sink_class *,
+ const struct dictionary *,
+ void *);
+void free_case_sink (struct case_sink *);
\f
/* Number of cases to lag. */
extern int n_lag;
-void procedure (void (*begin_func) (void *aux),
- int (*proc_func) (struct ccase *, void *aux),
- void (*end_func) (void *aux),
- void *aux);
+void procedure (int (*proc_func) (struct ccase *, void *aux), void *aux);
+void procedure_with_splits (void (*begin_func) (void *aux),
+ int (*proc_func) (struct ccase *, void *aux),
+ void (*end_func) (void *aux),
+ void *aux);
struct ccase *lagged_case (int n_before);
-void compact_case (struct ccase *dest, const struct ccase *src);
-void write_active_file_to_disk (void);
-
-void process_active_file (void (*begin_func) (void *),
- int (*casefunc) (struct ccase *, void *),
- void (*end_func) (void *),
- void *aux);
-void process_active_file_output_case (const struct ccase *);
#endif /* !vfm_h */
#include "var.h"
-/* Nonzero if the case needs to have values deleted before being
- stored, zero otherwise. */
-extern int compaction_necessary;
-
-/* Number of values after compaction. */
-extern int compaction_nval;
-
-/* Temporary case buffer with enough room for `compaction_nval'
- `value's. */
-extern struct ccase *compaction_case;
-
-void compact_case (struct ccase *dest, const struct ccase *src);
-
#endif /* !vfmP_h */
+Sun Mar 14 23:04:14 2004 Ben Pfaff <blp@gnu.org>
+
+ * command/sort.sh: Use numeric, not string, data to avoid spurious
+ valgrind complaints.
+
Wed Mar 10 21:22:03 2004 Ben Pfaff <blp@gnu.org>
* bugs/temporary.sh: Test that basic use of TEMPORARY works.
cat > $TEMPDIR/sort.stat <<EOF
title 'Test SORT procedure'.
-data list file='$here/sort.data' notable /X000 to X126 1-127(a).
+data list file='$here/sort.data' notable /X000 to X126 1-127.
sort by X000 to x005.
print /X000 to X005.
execute.
projects / pspp-builds.git / commitdiff