/* PSPP - computes sample statistics.
- Copyright (C) 1997-9, 2000, 2006 Free Software Foundation, Inc.
+ Copyright (C) 1997-9, 2000, 2006, 2007 Free Software Foundation, Inc.
This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License as
#include <data/any-reader.h>
#include <data/any-writer.h>
-#include <data/case-sink.h>
-#include <data/case-source.h>
#include <data/case.h>
-#include <data/casefile.h>
-#include <data/fastfile.h>
+#include <data/casereader.h>
+#include <data/casewriter.h>
#include <data/format.h>
#include <data/dictionary.h>
#include <data/por-file-writer.h>
#include <data/procedure.h>
#include <data/settings.h>
-#include <data/storage-stream.h>
#include <data/sys-file-writer.h>
#include <data/transformations.h>
#include <data/value-labels.h>
#include <libpspp/compiler.h>
#include <libpspp/hash.h>
#include <libpspp/message.h>
-#include <libpspp/message.h>
#include <libpspp/misc.h>
#include <libpspp/str.h>
+#include <libpspp/taint.h>
#include "gettext.h"
#define _(msgid) gettext (msgid)
IMPORT_CMD
};
-/* Case reader input program. */
-struct case_reader_pgm
- {
- struct any_reader *reader; /* File reader. */
- struct case_map *map; /* Map from file dict to active file dict. */
- struct ccase bounce; /* Bounce buffer. */
- };
-
-static const struct case_source_class case_reader_source_class;
-
-static void case_reader_pgm_free (struct case_reader_pgm *);
+static void get_translate_case (const struct ccase *, struct ccase *,
+ void *map_);
+static bool get_destroy_case_map (void *map_);
/* Parses a GET or IMPORT command. */
static int
parse_read_command (struct lexer *lexer, struct dataset *ds, enum reader_command type)
{
- struct case_reader_pgm *pgm = NULL;
+ struct casereader *reader = NULL;
struct file_handle *fh = NULL;
struct dictionary *dict = NULL;
+ struct case_map *map = NULL;
for (;;)
{
goto error;
}
- discard_variables (ds);
-
- pgm = xmalloc (sizeof *pgm);
- pgm->reader = any_reader_open (fh, &dict);
- pgm->map = NULL;
- case_nullify (&pgm->bounce);
- if (pgm->reader == NULL)
+ reader = any_reader_open (fh, &dict);
+ if (reader == NULL)
goto error;
- case_create (&pgm->bounce, dict_get_next_value_idx (dict));
-
start_case_map (dict);
while (lex_token (lexer) != '.')
goto error;
}
- pgm->map = finish_case_map (dict);
+ map = finish_case_map (dict);
+ if (map != NULL)
+ reader = casereader_create_translator (reader,
+ dict_get_next_value_idx (dict),
+ get_translate_case,
+ get_destroy_case_map,
+ map);
- dict_destroy (dataset_dict (ds));
- dataset_set_dict (ds, dict);
-
- proc_set_source (ds,
- create_case_source (&case_reader_source_class, pgm));
+ proc_set_active_file (ds, reader, dict);
return CMD_SUCCESS;
error:
- case_reader_pgm_free (pgm);
+ casereader_destroy (reader);
if (dict != NULL)
dict_destroy (dict);
return CMD_CASCADING_FAILURE;
}
-/* Frees a struct case_reader_pgm. */
static void
-case_reader_pgm_free (struct case_reader_pgm *pgm)
+get_translate_case (const struct ccase *input, struct ccase *output,
+ void *map_)
{
- if (pgm != NULL)
- {
- any_reader_close (pgm->reader);
- destroy_case_map (pgm->map);
- case_destroy (&pgm->bounce);
- free (pgm);
- }
-}
-
-/* Reads one case into C.
- Returns true if successful, false at end of file or if an
- I/O error occurred. */
-static bool
-case_reader_source_read (struct case_source *source, struct ccase *c)
-{
- struct case_reader_pgm *pgm = source->aux;
- if (any_reader_read (pgm->reader, pgm->map == NULL ? c : &pgm->bounce))
- {
- if (pgm->map != NULL)
- map_case (pgm->map, &pgm->bounce, c);
- return true;
- }
- else
- return false;
+ struct case_map *map = map_;
+ map_case (map, input, output);
}
-/* Destroys the source.
- Returns true if successful read, false if an I/O occurred
- during destruction or previously. */
static bool
-case_reader_source_destroy (struct case_source *source)
+get_destroy_case_map (void *map_)
{
- struct case_reader_pgm *pgm = source->aux;
- bool ok = !any_reader_error (pgm->reader);
- case_reader_pgm_free (pgm);
- return ok;
+ struct case_map *map = map_;
+ destroy_case_map (map);
+ return true;
}
-
-static const struct case_source_class case_reader_source_class =
- {
- "case reader",
- NULL,
- case_reader_source_read,
- case_reader_source_destroy,
- };
\f
/* GET. */
int
PROC_CMD /* Procedure. */
};
-/* File writer plus a case map. */
-struct case_writer
- {
- struct any_writer *writer; /* File writer. */
- struct case_map *map; /* Map to output file dictionary
- (null pointer for identity mapping). */
- struct ccase bounce; /* Bounce buffer for mapping (if needed). */
- };
-
-/* Destroys AW. */
-static bool
-case_writer_destroy (struct case_writer *aw)
-{
- bool ok = true;
- if (aw != NULL)
- {
- ok = any_writer_close (aw->writer);
- destroy_case_map (aw->map);
- case_destroy (&aw->bounce);
- free (aw);
- }
- return ok;
-}
-
/* Parses SAVE or XSAVE or EXPORT or XEXPORT command.
WRITER_TYPE identifies the type of file to write,
and COMMAND_TYPE identifies the type of command.
included.
On failure, returns a null pointer. */
-static struct case_writer *
+static struct casewriter *
parse_write_command (struct lexer *lexer, struct dataset *ds,
enum writer_type writer_type,
enum command_type command_type,
/* Common data. */
struct file_handle *handle; /* Output file. */
struct dictionary *dict; /* Dictionary for output file. */
- struct case_writer *aw; /* Writer. */
+ struct casewriter *writer; /* Writer. */
+ struct case_map *map; /* Map from input data to data for writer. */
/* Common options. */
bool print_map; /* Print map? TODO. */
handle = NULL;
dict = dict_clone (dataset_dict (ds));
- aw = xmalloc (sizeof *aw);
- aw->writer = NULL;
- aw->map = NULL;
- case_nullify (&aw->bounce);
+ writer = NULL;
+ map = NULL;
print_map = false;
print_short_names = false;
sysfile_opts = sfm_writer_default_options ();
}
dict_compact_values (dict);
- aw->map = finish_case_map (dict);
- if (aw->map != NULL)
- case_create (&aw->bounce, dict_get_next_value_idx (dict));
if (fh_get_referent (handle) == FH_REF_FILE)
{
switch (writer_type)
{
case SYSFILE_WRITER:
- aw->writer = any_writer_from_sfm_writer (
- sfm_open_writer (handle, dict, sysfile_opts));
+ writer = sfm_open_writer (handle, dict, sysfile_opts);
break;
case PORFILE_WRITER:
- aw->writer = any_writer_from_pfm_writer (
- pfm_open_writer (handle, dict, porfile_opts));
+ writer = pfm_open_writer (handle, dict, porfile_opts);
break;
}
}
else
- aw->writer = any_writer_open (handle, dict);
- if (aw->writer == NULL)
+ writer = any_writer_open (handle, dict);
+ if (writer == NULL)
goto error;
+
+ map = finish_case_map (dict);
+ if (map != NULL)
+ writer = casewriter_create_translator (writer,
+ get_translate_case,
+ get_destroy_case_map,
+ map);
dict_destroy (dict);
- return aw;
+ return writer;
error:
- case_writer_destroy (aw);
+ casewriter_destroy (writer);
dict_destroy (dict);
+ destroy_case_map (map);
return NULL;
}
-
-/* Writes case C to writer AW. */
-static bool
-case_writer_write_case (struct case_writer *aw, const struct ccase *c)
-{
- if (aw->map != NULL)
- {
- map_case (aw->map, c, &aw->bounce);
- c = &aw->bounce;
- }
- return any_writer_write (aw->writer, c);
-}
\f
/* SAVE and EXPORT. */
{
bool retain_unselected;
struct variable *saved_filter_variable;
- struct case_writer *aw;
- struct ccase *c;
- bool ok = true;
+ struct casewriter *output;
+ bool ok;
- aw = parse_write_command (lexer, ds, writer_type, PROC_CMD, &retain_unselected);
- if (aw == NULL)
+ output = parse_write_command (lexer, ds, writer_type, PROC_CMD,
+ &retain_unselected);
+ if (output == NULL)
return CMD_CASCADING_FAILURE;
saved_filter_variable = dict_get_filter (dataset_dict (ds));
if (retain_unselected)
dict_set_filter (dataset_dict (ds), NULL);
- proc_open (ds);
- while (ok && proc_read (ds, &c))
- ok = case_writer_write_case (aw, c);
- ok = proc_close (ds) && ok;
+ casereader_transfer (proc_open (ds), output);
+ ok = casewriter_destroy (output);
+ ok = proc_commit (ds) && ok;
dict_set_filter (dataset_dict (ds), saved_filter_variable);
- case_writer_destroy (aw);
return ok ? CMD_SUCCESS : CMD_CASCADING_FAILURE;
}
/* Transformation. */
struct output_trns
{
- struct case_writer *aw; /* Writer. */
+ struct casewriter *writer; /* Writer. */
};
static trns_proc_func output_trns_proc;
parse_output_trns (struct lexer *lexer, struct dataset *ds, enum writer_type writer_type)
{
struct output_trns *t = xmalloc (sizeof *t);
- t->aw = parse_write_command (lexer, ds, writer_type, XFORM_CMD, NULL);
- if (t->aw == NULL)
+ t->writer = parse_write_command (lexer, ds, writer_type, XFORM_CMD, NULL);
+ if (t->writer == NULL)
{
free (t);
return CMD_CASCADING_FAILURE;
output_trns_proc (void *trns_, struct ccase *c, casenumber case_num UNUSED)
{
struct output_trns *t = trns_;
- case_writer_write_case (t->aw, c);
+ struct ccase tmp;
+ case_clone (&tmp, c);
+ casewriter_write (t->writer, &tmp);
return TRNS_CONTINUE;
}
output_trns_free (void *trns_)
{
struct output_trns *t = trns_;
- bool ok = true;
-
- if (t != NULL)
- {
- ok = case_writer_destroy (t->aw);
- free (t);
- }
+ bool ok = casewriter_destroy (t->writer);
+ free (t);
return ok;
}
/* MATCH FILES. */
/* File types. */
-enum
+enum mtf_type
{
MTF_FILE, /* Specified on FILE= subcommand. */
MTF_TABLE /* Specified on TABLE= subcommand. */
};
-/* One of the files on MATCH FILES. */
+/* One of the FILEs or TABLEs on MATCH FILES. */
struct mtf_file
{
- struct mtf_file *next, *prev; /* Next, previous in the list of files. */
- struct mtf_file *next_min; /* Next in the chain of minimums. */
-
- int type; /* One of MTF_*. */
- struct variable **by; /* List of BY variables for this file. */
- struct file_handle *handle; /* File handle. */
- struct any_reader *reader; /* File reader. */
- struct dictionary *dict; /* Dictionary from system file. */
+ struct ll ll; /* In list of all files and tables. */
+
+ enum mtf_type type;
+ int sequence;
+
+ struct variable **by; /* List of BY variables for this file. */
+ struct mtf_variable *vars; /* Variables to copy to output. */
+ size_t var_cnt; /* Number of other variables. */
+
+ struct file_handle *handle; /* Input file handle. */
+ struct dictionary *dict; /* Input file dictionary. */
+ struct casereader *reader; /* Input reader. */
+ struct ccase input; /* Input record (null at end of file). */
/* IN subcommand. */
char *in_name; /* Variable name. */
struct variable *in_var; /* Variable (in master dictionary). */
+ };
- struct ccase input_storage; /* Input record storage. */
- struct ccase *input; /* Input record. */
+struct mtf_variable
+ {
+ struct variable *in_var;
+ struct variable *out_var;
};
/* MATCH FILES procedure. */
struct mtf_proc
{
- struct mtf_file *head; /* First file mentioned on FILE or TABLE. */
- struct mtf_file *tail; /* Last file mentioned on FILE or TABLE. */
+ struct ll_list files; /* List of "struct mtf_file"s. */
+ int nonempty_files; /* FILEs that are not at end-of-file. */
bool ok; /* False if I/O error occurs. */
- size_t by_cnt; /* Number of variables on BY subcommand. */
-
- /* Names of FIRST, LAST variables. */
- char first[LONG_NAME_LEN + 1], last[LONG_NAME_LEN + 1];
-
struct dictionary *dict; /* Dictionary of output file. */
- struct casefile *output; /* MATCH FILES output. */
- struct ccase mtf_case; /* Case used for output. */
+ struct casewriter *output; /* MATCH FILES output. */
+
+ size_t by_cnt; /* Number of variables on BY subcommand. */
- unsigned seq_num; /* Have we initialized this variable? */
- unsigned *seq_nums; /* Sequence numbers for each var in dict. */
+ /* FIRST, LAST.
+ Only if "first" or "last" is nonnull are the remaining
+ members used. */
+ struct variable *first; /* Variable specified on FIRST (if any). */
+ struct variable *last; /* Variable specified on LAST (if any). */
+ struct ccase buffered_case; /* Case ready for output except that we don't
+ know the value for the LAST variable yet. */
+ struct ccase prev_BY_case; /* Case with values of last set of BY vars. */
+ struct variable **prev_BY; /* Last set of BY variables. */
};
-static bool mtf_free (struct mtf_proc *);
-static bool mtf_close_file (struct mtf_file *);
-static int mtf_merge_dictionary (struct dictionary *const, struct mtf_file *);
-static bool mtf_read_records (struct mtf_proc *, struct dataset *);
-static bool mtf_delete_file_in_place (struct mtf_proc *, struct mtf_file **);
+static void mtf_free (struct mtf_proc *);
-static bool mtf_processing (struct mtf_proc *, struct dataset *);
+static bool mtf_close_all_files (struct mtf_proc *);
+static bool mtf_merge_dictionary (struct dictionary *const, struct mtf_file *);
+static bool mtf_read_record (struct mtf_proc *mtf, struct mtf_file *);
-static char *var_type_description (struct variable *);
+static void mtf_process_case (struct mtf_proc *);
-static void set_master (struct variable *, struct variable *master);
-static struct variable *get_master (struct variable *);
+static bool create_flag_var (const char *subcommand_name, const char *var_name,
+ struct dictionary *, struct variable **);
+static char *var_type_description (struct variable *);
/* Parse and execute the MATCH FILES command. */
int
cmd_match_files (struct lexer *lexer, struct dataset *ds)
{
struct mtf_proc mtf;
- struct mtf_file *first_table = NULL;
- struct mtf_file *iter;
+ struct ll *first_table;
+ struct mtf_file *file, *next;
- bool used_active_file = false;
- bool saw_table = false;
bool saw_in = false;
+ struct casereader *active_file = NULL;
- mtf.head = mtf.tail = NULL;
- mtf.by_cnt = 0;
- mtf.first[0] = '\0';
- mtf.last[0] = '\0';
+ char first_name[LONG_NAME_LEN + 1] = "";
+ char last_name[LONG_NAME_LEN + 1] = "";
+
+ struct taint *taint = NULL;
+
+ size_t i;
+
+ ll_init (&mtf.files);
+ mtf.nonempty_files = 0;
+ first_table = ll_null (&mtf.files);
mtf.dict = dict_create ();
mtf.output = NULL;
- case_nullify (&mtf.mtf_case);
- mtf.seq_num = 0;
- mtf.seq_nums = NULL;
+ mtf.by_cnt = 0;
+ mtf.first = mtf.last = NULL;
+ case_nullify (&mtf.buffered_case);
+ case_nullify (&mtf.prev_BY_case);
+ mtf.prev_BY = NULL;
+
dict_set_case_limit (mtf.dict, dict_get_case_limit (dataset_dict (ds)));
lex_match (lexer, '/');
|| lex_id_match (ss_cstr ("TABLE"), ss_cstr (lex_tokid (lexer)))))
{
struct mtf_file *file = xmalloc (sizeof *file);
-
- if (lex_match_id (lexer, "FILE"))
- file->type = MTF_FILE;
- else if (lex_match_id (lexer, "TABLE"))
- {
- file->type = MTF_TABLE;
- saw_table = true;
- }
- else
- NOT_REACHED ();
- lex_match (lexer, '=');
-
file->by = NULL;
file->handle = NULL;
file->reader = NULL;
file->dict = NULL;
file->in_name = NULL;
file->in_var = NULL;
- case_nullify (&file->input_storage);
- file->input = &file->input_storage;
+ file->var_cnt = 0;
+ file->vars = NULL;
+ case_nullify (&file->input);
- /* FILEs go first, then TABLEs. */
- if (file->type == MTF_TABLE || first_table == NULL)
+ if (lex_match_id (lexer, "FILE"))
{
- file->next = NULL;
- file->prev = mtf.tail;
- if (mtf.tail)
- mtf.tail->next = file;
- mtf.tail = file;
- if (mtf.head == NULL)
- mtf.head = file;
- if (file->type == MTF_TABLE && first_table == NULL)
- first_table = file;
+ file->type = MTF_FILE;
+ ll_insert (first_table, &file->ll);
+ mtf.nonempty_files++;
}
- else
+ else if (lex_match_id (lexer, "TABLE"))
{
- assert (file->type == MTF_FILE);
- file->next = first_table;
- file->prev = first_table->prev;
- if (first_table->prev)
- first_table->prev->next = file;
- else
- mtf.head = file;
- first_table->prev = file;
+ file->type = MTF_TABLE;
+ ll_push_tail (&mtf.files, &file->ll);
+ if (first_table == ll_null (&mtf.files))
+ first_table = &file->ll;
}
+ else
+ NOT_REACHED ();
+ lex_match (lexer, '=');
if (lex_match (lexer, '*'))
{
- file->handle = NULL;
- file->reader = NULL;
-
- if (used_active_file)
- {
- msg (SE, _("The active file may not be specified more "
- "than once."));
- goto error;
- }
- used_active_file = true;
-
- if (!proc_has_source (ds))
+ if (!proc_has_active_file (ds))
{
msg (SE, _("Cannot specify the active file since no active "
"file has been defined."));
"the active file is an input source. "
"Temporary transformations will be made permanent."));
- file->dict = dataset_dict (ds);
+ file->dict = dict_clone (dataset_dict (ds));
}
else
{
file->reader = any_reader_open (file->handle, &file->dict);
if (file->reader == NULL)
goto error;
-
- case_create (&file->input_storage,
- dict_get_next_value_idx (file->dict));
}
while (lex_match (lexer, '/'))
{
if (lex_match (lexer, T_BY))
{
+ struct mtf_file *file;
struct variable **by;
+ bool ok;
if (mtf.by_cnt)
{
- msg (SE, _("BY may appear at most once."));
+ lex_sbc_only_once ("BY");
goto error;
}
PV_NO_DUPLICATE | PV_NO_SCRATCH))
goto error;
- for (iter = mtf.head; iter != NULL; iter = iter->next)
+ ok = true;
+ ll_for_each (file, struct mtf_file, ll, &mtf.files)
{
size_t i;
- iter->by = xnmalloc (mtf.by_cnt, sizeof *iter->by);
-
+ file->by = xnmalloc (mtf.by_cnt, sizeof *file->by);
for (i = 0; i < mtf.by_cnt; i++)
{
- iter->by[i] = dict_lookup_var (iter->dict,
- var_get_name (by[i]));
- if (iter->by[i] == NULL)
+ const char *var_name = var_get_name (by[i]);
+ file->by[i] = dict_lookup_var (file->dict, var_name);
+ if (file->by[i] == NULL)
{
- msg (SE, _("File %s lacks BY variable %s."),
- iter->handle ? fh_get_name (iter->handle) : "*",
- var_get_name (by[i]));
- free (by);
- goto error;
+ if (file->handle != NULL)
+ msg (SE, _("File %s lacks BY variable %s."),
+ fh_get_name (file->handle), var_name);
+ else
+ msg (SE, _("Active file lacks BY variable %s."),
+ var_name);
+ ok = false;
}
}
}
free (by);
+
+ if (!ok)
+ goto error;
}
else if (lex_match_id (lexer, "FIRST"))
{
- if (mtf.first[0] != '\0')
+ if (first_name[0] != '\0')
{
- msg (SE, _("FIRST may appear at most once."));
+ lex_sbc_only_once ("FIRST");
goto error;
}
lex_match (lexer, '=');
if (!lex_force_id (lexer))
goto error;
- strcpy (mtf.first, lex_tokid (lexer));
+ strcpy (first_name, lex_tokid (lexer));
lex_get (lexer);
}
else if (lex_match_id (lexer, "LAST"))
{
- if (mtf.last[0] != '\0')
+ if (last_name[0] != '\0')
{
- msg (SE, _("LAST may appear at most once."));
+ lex_sbc_only_once ("LAST");
goto error;
}
lex_match (lexer, '=');
if (!lex_force_id (lexer))
goto error;
- strcpy (mtf.last, lex_tokid (lexer));
+ strcpy (last_name, lex_tokid (lexer));
lex_get (lexer);
}
else if (lex_match_id (lexer, "MAP"))
if (mtf.by_cnt == 0)
{
- if (saw_table)
+ if (first_table != ll_null (&mtf.files))
{
msg (SE, _("BY is required when TABLE is specified."));
goto error;
/* Set up mapping from each file's variables to master
variables. */
- for (iter = mtf.head; iter != NULL; iter = iter->next)
+ ll_for_each (file, struct mtf_file, ll, &mtf.files)
{
- struct dictionary *d = iter->dict;
- int i;
+ size_t in_var_cnt = dict_get_var_cnt (file->dict);
- for (i = 0; i < dict_get_var_cnt (d); i++)
+ file->vars = xnmalloc (in_var_cnt, sizeof *file->vars);
+ file->var_cnt = 0;
+ for (i = 0; i < in_var_cnt; i++)
{
- struct variable *v = dict_get_var (d, i);
- struct variable *mv = dict_lookup_var (mtf.dict, var_get_name (v));
- if (mv != NULL)
- set_master (v, mv);
+ struct variable *in_var = dict_get_var (file->dict, i);
+ struct variable *out_var = dict_lookup_var (mtf.dict,
+ var_get_name (in_var));
+
+ if (out_var != NULL)
+ {
+ struct mtf_variable *mv = &file->vars[file->var_cnt++];
+ mv->in_var = in_var;
+ mv->out_var = out_var;
+ }
}
}
- /* Add IN variables to master dictionary. */
- for (iter = mtf.head; iter != NULL; iter = iter->next)
- if (iter->in_name != NULL)
- {
- struct fmt_spec format = fmt_for_output (FMT_F, 1, 0);
- iter->in_var = dict_create_var (mtf.dict, iter->in_name, 0);
- if (iter->in_var == NULL)
- {
- msg (SE, _("IN variable name %s duplicates an "
- "existing variable name."),
- var_get_name (iter->in_var));
- goto error;
- }
- var_set_both_formats (iter->in_var, &format);
- }
-
- /* MATCH FILES performs an n-way merge on all its input files.
- Abstract algorithm:
-
- 1. Read one input record from every input FILE.
-
- 2. If no FILEs are left, stop. Otherwise, proceed to step 3.
-
- 3. Find the FILE input record(s) that have minimum BY
- values. Store all the values from these input records into
- the output record.
-
- 4. For every TABLE, read another record as long as the BY values
- on the TABLE's input record are less than the FILEs' BY values.
- If an exact match is found, store all the values from the TABLE
- input record into the output record.
-
- 5. Write the output record.
-
- 6. Read another record from each input file FILE and TABLE that
- we stored values from above. If we come to the end of one of the
- input files, remove it from the list of input files.
-
- 7. Repeat from step 2.
+ /* Add IN, FIRST, and LAST variables to master dictionary. */
+ ll_for_each (file, struct mtf_file, ll, &mtf.files)
+ if (!create_flag_var ("IN", file->in_name, mtf.dict, &file->in_var))
+ goto error;
+ if (!create_flag_var ("FIRST", first_name, mtf.dict, &mtf.first)
+ || !create_flag_var ("LAST", last_name, mtf.dict, &mtf.last))
+ goto error;
- FIXME: For merging large numbers of files (more than 10?) a
- better algorithm would use a heap for finding minimum
- values. */
+ dict_compact_values (mtf.dict);
+ mtf.output = autopaging_writer_create (dict_get_next_value_idx (mtf.dict));
+ taint = taint_clone (casewriter_get_taint (mtf.output));
- if (used_active_file)
+ ll_for_each (file, struct mtf_file, ll, &mtf.files)
+ {
+ if (file->reader == NULL)
+ {
+ if (active_file == NULL)
+ {
+ proc_discard_output (ds);
+ file->reader = active_file = proc_open (ds);
+ }
+ else
+ file->reader = casereader_clone (active_file);
+ }
+ taint_propagate (casereader_get_taint (file->reader), taint);
+ }
+
+ ll_for_each_safe (file, next, struct mtf_file, ll, &mtf.files)
+ mtf_read_record (&mtf, file);
+ while (mtf.nonempty_files > 0)
+ mtf_process_case (&mtf);
+ if ((mtf.first != NULL || mtf.last != NULL) && mtf.prev_BY != NULL)
{
- proc_set_sink (ds, create_case_sink (&null_sink_class,
- dataset_dict (ds), NULL));
- proc_open (ds);
+ if (mtf.last != NULL)
+ case_data_rw (&mtf.buffered_case, mtf.last)->f = 1.0;
+ casewriter_write (mtf.output, &mtf.buffered_case);
+ case_nullify (&mtf.buffered_case);
}
- else
- discard_variables (ds);
+ mtf_close_all_files (&mtf);
+ if (active_file != NULL)
+ proc_commit (ds);
- dict_compact_values (mtf.dict);
- mtf.output = fastfile_create (dict_get_next_value_idx (mtf.dict));
- mtf.seq_nums = xcalloc (dict_get_var_cnt (mtf.dict), sizeof *mtf.seq_nums);
- case_create (&mtf.mtf_case, dict_get_next_value_idx (mtf.dict));
+ proc_set_active_file (ds, casewriter_make_reader (mtf.output), mtf.dict);
+ mtf.dict = NULL;
+ mtf.output = NULL;
- if (!mtf_read_records (&mtf, ds))
- goto error;
- while (mtf.head && mtf.head->type == MTF_FILE)
- if (!mtf_processing (&mtf, ds))
- goto error;
- if (!proc_close (ds))
- goto error;
+ mtf_free (&mtf);
- discard_variables (ds);
+ return taint_destroy (taint) ? CMD_SUCCESS : CMD_CASCADING_FAILURE;
- dict_destroy (dataset_dict (ds));
- dataset_set_dict (ds, mtf.dict);
- mtf.dict = NULL;
- proc_set_source (ds, storage_source_create (mtf.output));
- mtf.output = NULL;
-
- return mtf_free (&mtf) ? CMD_SUCCESS : CMD_CASCADING_FAILURE;
-
error:
- proc_close (ds);
+ if (active_file != NULL)
+ proc_commit (ds);
mtf_free (&mtf);
+ taint_destroy (taint);
return CMD_CASCADING_FAILURE;
}
-/* Return a string in a static buffer describing V's variable type and
- width. */
+/* If VAR_NAME is a nonnull pointer to a non-empty string,
+ attempts to create a variable named VAR_NAME, with format
+ F1.0, in DICT, and stores a pointer to the variable in *VAR.
+ Returns true if successful, false if the variable name is a
+ duplicate (in which case a message saying that the variable
+ specified on the given SUBCOMMAND is a duplicate is emitted).
+ Also returns true, without doing anything, if VAR_NAME is null
+ or empty. */
+static bool
+create_flag_var (const char *subcommand, const char *var_name,
+ struct dictionary *dict, struct variable **var)
+{
+ if (var_name != NULL && var_name[0] != '\0')
+ {
+ struct fmt_spec format = fmt_for_output (FMT_F, 1, 0);
+ *var = dict_create_var (dict, var_name, 0);
+ if (*var == NULL)
+ {
+ msg (SE, _("Variable name %s specified on %s subcommand "
+ "duplicates an existing variable name."),
+ subcommand, var_name);
+ return false;
+ }
+ var_set_both_formats (*var, &format);
+ }
+ else
+ *var = NULL;
+ return true;
+}
+
+/* Return a string in an allocated buffer describing V's variable
+ type and width. */
static char *
var_type_description (struct variable *v)
{
- static char buf[2][32];
- static int x = 0;
- char *s;
-
- x ^= 1;
- s = buf[x];
-
if (var_is_numeric (v))
- strcpy (s, "numeric");
+ return xstrdup ("numeric");
else
- sprintf (s, "string with width %d", var_get_width (v));
- return s;
+ return xasprintf ("string with width %d", var_get_width (v));
}
-/* Closes FILE and frees its associated data.
- Returns true if successful, false if an I/O error
- occurred on FILE. */
+/* Closes all the files in MTF and frees their associated data.
+ Returns true if successful, false if an I/O error occurred on
+ any of the files. */
static bool
-mtf_close_file (struct mtf_file *file)
+mtf_close_all_files (struct mtf_proc *mtf)
{
- bool ok = file->reader == NULL || !any_reader_error (file->reader);
- free (file->by);
- any_reader_close (file->reader);
- if (file->handle != NULL)
- dict_destroy (file->dict);
- case_destroy (&file->input_storage);
- free (file->in_name);
- free (file);
- return ok;
-}
-
-/* Free all the data for the MATCH FILES procedure.
- Returns true if successful, false if an I/O error
- occurred. */
-static bool
-mtf_free (struct mtf_proc *mtf)
-{
- struct mtf_file *iter, *next;
+ struct mtf_file *file;
bool ok = true;
- for (iter = mtf->head; iter; iter = next)
+ ll_for_each_preremove (file, struct mtf_file, ll, &mtf->files)
{
- next = iter->next;
- assert (iter->dict != mtf->dict);
- if (!mtf_close_file (iter))
- ok = false;
+ casereader_destroy (file->reader);
+ free (file->by);
+ dict_destroy (file->dict);
+ free (file->in_name);
+ case_destroy (&file->input);
+ free (file->vars);
+ free (file);
}
-
- if (mtf->dict)
- dict_destroy (mtf->dict);
- case_destroy (&mtf->mtf_case);
- free (mtf->seq_nums);
return ok;
}
-/* Remove *FILE from the mtf_file chain. Make *FILE point to the next
- file in the chain, or to NULL if was the last in the chain.
- Returns true if successful, false if an I/O error occurred. */
-static bool
-mtf_delete_file_in_place (struct mtf_proc *mtf, struct mtf_file **file)
+/* Frees all the data for the MATCH FILES procedure. */
+static void
+mtf_free (struct mtf_proc *mtf)
{
- struct mtf_file *f = *file;
- int i;
-
- if (f->prev)
- f->prev->next = f->next;
- if (f->next)
- f->next->prev = f->prev;
- if (f == mtf->head)
- mtf->head = f->next;
- if (f == mtf->tail)
- mtf->tail = f->prev;
- *file = f->next;
-
- if (f->in_var != NULL)
- case_data_rw (&mtf->mtf_case, f->in_var)->f = 0.;
- for (i = 0; i < dict_get_var_cnt (f->dict); i++)
- {
- struct variable *v = dict_get_var (f->dict, i);
- struct variable *mv = get_master (v);
- if (mv != NULL)
- {
- union value *out = case_data_rw (&mtf->mtf_case, mv);
-
- if (var_is_numeric (v))
- out->f = SYSMIS;
- else
- memset (out->s, ' ', var_get_width (v));
- }
- }
-
- return mtf_close_file (f);
+ mtf_close_all_files (mtf);
+ dict_destroy (mtf->dict);
+ casewriter_destroy (mtf->output);
+ case_destroy (&mtf->buffered_case);
+ case_destroy (&mtf->prev_BY_case);
}
-/* Read a record from every input file.
- Returns true if successful, false if an I/O error occurred. */
+/* Reads the next record into FILE, if possible, and update MTF's
+ nonempty_files count if not. */
static bool
-mtf_read_records (struct mtf_proc *mtf, struct dataset *ds)
+mtf_read_record (struct mtf_proc *mtf, struct mtf_file *file)
{
- struct mtf_file *iter, *next;
- bool ok = true;
-
- for (iter = mtf->head; ok && iter != NULL; iter = next)
+ case_destroy (&file->input);
+ if (!casereader_read (file->reader, &file->input))
{
- next = iter->next;
- if (iter->handle
- ? !any_reader_read (iter->reader, iter->input)
- : !proc_read (ds, &iter->input))
- {
- if (!mtf_delete_file_in_place (mtf, &iter))
- ok = false;
- }
+ mtf->nonempty_files--;
+ return false;
}
- return ok;
+ else
+ return true;
}
-/* Compare the BY variables for files A and B; return -1 if A < B, 0
- if A == B, 1 if A > B. */
+/* Compare the BY variables for files A and B; return -1 if A <
+ B, 0 if A == B, 1 if A > B. (If there are no BY variables,
+ then all records are equal.) */
static inline int
mtf_compare_BY_values (struct mtf_proc *mtf,
struct mtf_file *a, struct mtf_file *b)
{
- return case_compare_2dict (a->input, b->input, a->by, b->by, mtf->by_cnt);
+ return case_compare_2dict (&a->input, &b->input, a->by, b->by, mtf->by_cnt);
}
-/* Perform one iteration of steps 3...7 above.
- Returns true if successful, false if an I/O error occurred. */
-static bool
-mtf_processing (struct mtf_proc *mtf, struct dataset *ds)
+/* Processes input files and write one case to the output file. */
+static void
+mtf_process_case (struct mtf_proc *mtf)
{
- struct mtf_file *min_head, *min_tail; /* Files with minimum BY values. */
- struct mtf_file *max_head, *max_tail; /* Files with non-minimum BYs. */
- struct mtf_file *iter, *next;
-
- /* 3. Find the FILE input record(s) that have minimum BY
- values. Store all the values from these input records into
- the output record. */
- min_head = min_tail = mtf->head;
- max_head = max_tail = NULL;
- for (iter = mtf->head->next; iter && iter->type == MTF_FILE;
- iter = iter->next)
+ struct ccase c;
+ struct mtf_file *min;
+ struct mtf_file *file;
+ int min_sequence;
+ size_t i;
+
+ /* Find the set of one or more FILEs whose BY values are
+ minimal, as well as the set of zero or more TABLEs whose BY
+ values equal those of the minimum FILEs.
+
+ After each iteration of the loop, this invariant holds: the
+ FILEs with minimum BY values thus far have "sequence"
+ members equal to min_sequence, and "min" points to one of
+ the mtf_files whose case has those minimum BY values, and
+ similarly for TABLEs. */
+ min_sequence = 0;
+ min = NULL;
+ ll_for_each (file, struct mtf_file, ll, &mtf->files)
+ if (case_is_null (&file->input))
+ file->sequence = -1;
+ else if (file->type == MTF_FILE)
+ {
+ int cmp = min != NULL ? mtf_compare_BY_values (mtf, min, file) : 1;
+ if (cmp <= 0)
+ file->sequence = cmp < 0 ? -1 : min_sequence;
+ else
+ {
+ file->sequence = ++min_sequence;
+ min = file;
+ }
+ }
+ else
+ {
+ int cmp;
+ assert (min != NULL);
+ do
+ {
+ cmp = mtf_compare_BY_values (mtf, min, file);
+ }
+ while (cmp > 0 && mtf_read_record (mtf, file));
+ file->sequence = cmp == 0 ? min_sequence : -1;
+ }
+
+ /* Form the output case from the input cases. */
+ case_create (&c, dict_get_next_value_idx (mtf->dict));
+ for (i = 0; i < dict_get_var_cnt (mtf->dict); i++)
{
- int cmp = mtf_compare_BY_values (mtf, min_head, iter);
- if (cmp < 0)
- {
- if (max_head)
- max_tail = max_tail->next_min = iter;
- else
- max_head = max_tail = iter;
- }
- else if (cmp == 0)
- min_tail = min_tail->next_min = iter;
- else /* cmp > 0 */
- {
- if (max_head)
- {
- max_tail->next_min = min_head;
- max_tail = min_tail;
- }
- else
- {
- max_head = min_head;
- max_tail = min_tail;
- }
- min_head = min_tail = iter;
- }
+ struct variable *v = dict_get_var (mtf->dict, i);
+ value_set_missing (case_data_rw (&c, v), var_get_width (v));
}
-
- /* 4. For every TABLE, read another record as long as the BY
- values on the TABLE's input record are less than the FILEs'
- BY values. If an exact match is found, store all the values
- from the TABLE input record into the output record. */
- for (; iter != NULL; iter = next)
+ ll_for_each_reverse (file, struct mtf_file, ll, &mtf->files)
{
- assert (iter->type == MTF_TABLE);
-
- next = iter->next;
- for (;;)
- {
- int cmp = mtf_compare_BY_values (mtf, min_head, iter);
- if (cmp < 0)
- {
- if (max_head)
- max_tail = max_tail->next_min = iter;
- else
- max_head = max_tail = iter;
- }
- else if (cmp == 0)
- min_tail = min_tail->next_min = iter;
- else /* cmp > 0 */
- {
- if (iter->handle
- ? any_reader_read (iter->reader, iter->input)
- : proc_read (ds, &iter->input))
- continue;
- if (!mtf_delete_file_in_place (mtf, &iter))
- return false;
- }
- break;
- }
+ bool include_file = file->sequence == min_sequence;
+ if (include_file)
+ for (i = 0; i < file->var_cnt; i++)
+ {
+ const struct mtf_variable *mv = &file->vars[i];
+ const union value *in = case_data (&file->input, mv->in_var);
+ union value *out = case_data_rw (&c, mv->out_var);
+ value_copy (out, in, var_get_width (mv->in_var));
+ }
+ if (file->in_var != NULL)
+ case_data_rw (&c, file->in_var)->f = include_file;
}
-
- /* Next sequence number. */
- mtf->seq_num++;
-
- /* Store data to all the records we are using. */
- if (min_tail)
- min_tail->next_min = NULL;
- for (iter = min_head; iter; iter = iter->next_min)
+
+ /* Write the output case. */
+ if (mtf->first == NULL && mtf->last == NULL)
{
- int i;
-
- for (i = 0; i < dict_get_var_cnt (iter->dict); i++)
- {
- struct variable *v = dict_get_var (iter->dict, i);
- struct variable *mv = get_master (v);
- size_t mv_index = mv ? var_get_dict_index (mv) : 0;
-
- if (mv != NULL && mtf->seq_nums[mv_index] != mtf->seq_num)
- {
- const struct ccase *record = iter->input;
- union value *out = case_data_rw (&mtf->mtf_case, mv);
-
- mtf->seq_nums[mv_index] = mtf->seq_num;
- if (var_is_numeric (v))
- out->f = case_num (record, v);
- else
- memcpy (out->s, case_str (record, v), var_get_width (v));
- }
- }
- if (iter->in_var != NULL)
- case_data_rw (&mtf->mtf_case, iter->in_var)->f = 1.;
+ /* With no FIRST or LAST variables, it's trivial. */
+ casewriter_write (mtf->output, &c);
}
-
- /* Store missing values to all the records we're not using. */
- if (max_tail)
- max_tail->next_min = NULL;
- for (iter = max_head; iter; iter = iter->next_min)
+ else
{
- int i;
-
- for (i = 0; i < dict_get_var_cnt (iter->dict); i++)
+ /* It's harder with LAST, because we can't know whether
+ this case is the last in a group until we've prepared
+ the *next* case also. Thus, we buffer the previous
+ output case until the next one is ready.
+
+ We also have to save a copy of one of the previous input
+ cases, so that we can compare the BY variables. We
+ can't compare the BY variables between the current
+ output case and the saved one because the BY variables
+ might not be in the output (the user is allowed to drop
+ them). */
+ bool new_BY;
+ if (mtf->prev_BY != NULL)
{
- struct variable *v = dict_get_var (iter->dict, i);
- struct variable *mv = get_master (v);
- size_t mv_index = mv ? var_get_dict_index (mv) : 0;
+ new_BY = case_compare_2dict (&min->input, &mtf->prev_BY_case,
+ min->by, mtf->prev_BY,
+ mtf->by_cnt);
+ if (mtf->last != NULL)
+ case_data_rw (&mtf->buffered_case, mtf->last)->f = new_BY;
+ casewriter_write (mtf->output, &mtf->buffered_case);
+ }
+ else
+ new_BY = true;
- if (mv != NULL && mtf->seq_nums[mv_index] != mtf->seq_num)
- {
- union value *out = case_data_rw (&mtf->mtf_case, mv);
- mtf->seq_nums[mv_index] = mtf->seq_num;
+ case_move (&mtf->buffered_case, &c);
+ if (mtf->first != NULL)
+ case_data_rw (&mtf->buffered_case, mtf->first)->f = new_BY;
- if (var_is_numeric (v))
- out->f = SYSMIS;
- else
- memset (out->s, ' ', var_get_width (v));
- }
+ if (new_BY)
+ {
+ mtf->prev_BY = min->by;
+ case_destroy (&mtf->prev_BY_case);
+ case_clone (&mtf->prev_BY_case, &min->input);
}
- if (iter->in_var != NULL)
- case_data_rw (&mtf->mtf_case, iter->in_var)->f = 0.;
}
- /* 5. Write the output record. */
- casefile_append (mtf->output, &mtf->mtf_case);
-
- /* 6. Read another record from each input file FILE and TABLE
- that we stored values from above. If we come to the end of
- one of the input files, remove it from the list of input
- files. */
- for (iter = min_head; iter && iter->type == MTF_FILE; iter = next)
- {
- next = iter->next_min;
- if (iter->reader != NULL
- ? !any_reader_read (iter->reader, iter->input)
- : !proc_read (ds, &iter->input))
- if (!mtf_delete_file_in_place (mtf, &iter))
- return false;
- }
- return true;
+ /* Read another record from each input file FILE with minimum
+ values. */
+ ll_for_each (file, struct mtf_file, ll, &mtf->files)
+ if (file->type == MTF_FILE)
+ {
+ if (file->sequence == min_sequence)
+ mtf_read_record (mtf, file);
+ }
+ else
+ break;
}
/* Merge the dictionary for file F into master dictionary M. */
-static int
+static bool
mtf_merge_dictionary (struct dictionary *const m, struct mtf_file *f)
{
struct dictionary *d = f->dict;
dict_set_documents (m, d_docs);
else
{
- char *new_docs;
- size_t new_len;
-
- new_len = strlen (m_docs) + strlen (d_docs);
- new_docs = xmalloc (new_len + 1);
- strcpy (new_docs, m_docs);
- strcat (new_docs, d_docs);
+ char *new_docs = xasprintf ("%s%s", m_docs, d_docs);
dict_set_documents (m, new_docs);
free (new_docs);
}
{
if (var_get_width (mv) != var_get_width (dv))
{
+ char *dv_description = var_type_description (dv);
+ char *mv_description = var_type_description (mv);
msg (SE, _("Variable %s in file %s (%s) has different "
"type or width from the same variable in "
"earlier file (%s)."),
var_get_name (dv), fh_get_name (f->handle),
- var_type_description (dv), var_type_description (mv));
- return 0;
+ dv_description, mv_description);
+ free (dv_description);
+ free (mv_description);
+ return false;
}
if (var_get_width (dv) == var_get_width (mv))
if (var_get_label (dv) && !var_get_label (mv))
var_set_label (mv, var_get_label (dv));
}
- else
+ else
mv = dict_clone_var_assert (m, dv, var_get_name (dv));
}
- return 1;
-}
-
-/* Marks V's master variable as MASTER. */
-static void
-set_master (struct variable *v, struct variable *master)
-{
- var_attach_aux (v, master, NULL);
-}
-
-/* Returns the master variable corresponding to V,
- as set with set_master(). */
-static struct variable *
-get_master (struct variable *v)
-{
- return var_get_aux (v);
+ return true;
}
\f
/* Case map.
{
size_t dst_idx;
- assert (map != NULL);
- assert (src != NULL);
- assert (dst != NULL);
- assert (src != dst);
-
for (dst_idx = 0; dst_idx < map->value_cnt; dst_idx++)
{
int src_idx = map->map[dst_idx];