src/language/data-io/get.c

   1 /* PSPP - computes sample statistics.
   2    Copyright (C) 1997-9, 2000, 2006 Free Software Foundation, Inc.
   3    Written by Ben Pfaff <blp@gnu.org>.
   4
   5    This program is free software; you can redistribute it and/or
   6    modify it under the terms of the GNU General Public License as
   7    published by the Free Software Foundation; either version 2 of the
   8    License, or (at your option) any later version.
   9
  10    This program is distributed in the hope that it will be useful, but
  11    WITHOUT ANY WARRANTY; without even the implied warranty of
  12    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13    General Public License for more details.
  14
  15    You should have received a copy of the GNU General Public License
  16    along with this program; if not, write to the Free Software
  17    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  18    02110-1301, USA. */
  19
  20 #include <config.h>
  21
  22 #include <stdlib.h>
  23
  24 #include <data/any-reader.h>
  25 #include <data/any-writer.h>
  26 #include <data/case-sink.h>
  27 #include <data/case-source.h>
  28 #include <data/case.h>
  29 #include <data/casefile.h>
  30 #include <data/fastfile.h>
  31 #include <data/dictionary.h>
  32 #include <data/por-file-writer.h>
  33 #include <data/procedure.h>
  34 #include <data/settings.h>
  35 #include <data/storage-stream.h>
  36 #include <data/sys-file-writer.h>
  37 #include <data/transformations.h>
  38 #include <data/value-labels.h>
  39 #include <data/variable.h>
  40 #include <language/command.h>
  41 #include <language/data-io/file-handle.h>
  42 #include <language/lexer/lexer.h>
  43 #include <language/lexer/variable-parser.h>
  44 #include <libpspp/alloc.h>
  45 #include <libpspp/assertion.h>
  46 #include <libpspp/compiler.h>
  47 #include <libpspp/hash.h>
  48 #include <libpspp/message.h>
  49 #include <libpspp/message.h>
  50 #include <libpspp/misc.h>
  51 #include <libpspp/str.h>
  52
  53 #include "gettext.h"
  54 #define _(msgid) gettext (msgid)
  55
  56 /* Rearranging and reducing a dictionary. */
  57 static void start_case_map (struct dictionary *);
  58 static struct case_map *finish_case_map (struct dictionary *);
  59 static void map_case (const struct case_map *,
  60                       const struct ccase *, struct ccase *);
  61 static void destroy_case_map (struct case_map *);
  62
  63 static bool parse_dict_trim (struct lexer *, struct dictionary *);
  64 \f
  65 /* Reading system and portable files. */
  66
  67 /* Type of command. */
  68 enum reader_command
  69   {
  70     GET_CMD,
  71     IMPORT_CMD
  72   };
  73
  74 /* Case reader input program. */
  75 struct case_reader_pgm
  76   {
  77     struct any_reader *reader;  /* File reader. */
  78     struct case_map *map;       /* Map from file dict to active file dict. */
  79     struct ccase bounce;        /* Bounce buffer. */
  80   };
  81
  82 static const struct case_source_class case_reader_source_class;
  83
  84 static void case_reader_pgm_free (struct case_reader_pgm *);
  85
  86 /* Parses a GET or IMPORT command. */
  87 static int
  88 parse_read_command (struct lexer *lexer, struct dataset *ds, enum reader_command type)
  89 {
  90   struct case_reader_pgm *pgm = NULL;
  91   struct file_handle *fh = NULL;
  92   struct dictionary *dict = NULL;
  93
  94   for (;;)
  95     {
  96       lex_match (lexer, '/');
  97
  98       if (lex_match_id (lexer, "FILE") || lex_token (lexer) == T_STRING)
  99         {
 100           lex_match (lexer, '=');
 101
 102           fh = fh_parse (lexer, FH_REF_FILE | FH_REF_SCRATCH);
 103           if (fh == NULL)
 104             goto error;
 105         }
 106       else if (type == IMPORT_CMD && lex_match_id (lexer, "TYPE"))
 107         {
 108           lex_match (lexer, '=');
 109
 110           if (lex_match_id (lexer, "COMM"))
 111             type = PFM_COMM;
 112           else if (lex_match_id (lexer, "TAPE"))
 113             type = PFM_TAPE;
 114           else
 115             {
 116               lex_error (lexer, _("expecting COMM or TAPE"));
 117               goto error;
 118             }
 119         }
 120       else
 121         break;
 122     }
 123
 124   if (fh == NULL)
 125     {
 126       lex_sbc_missing (lexer, "FILE");
 127       goto error;
 128     }
 129
 130   discard_variables (ds);
 131
 132   pgm = xmalloc (sizeof *pgm);
 133   pgm->reader = any_reader_open (fh, &dict);
 134   pgm->map = NULL;
 135   case_nullify (&pgm->bounce);
 136   if (pgm->reader == NULL)
 137     goto error;
 138
 139   case_create (&pgm->bounce, dict_get_next_value_idx (dict));
 140
 141   start_case_map (dict);
 142
 143   while (lex_token (lexer) != '.')
 144     {
 145       lex_match (lexer, '/');
 146       if (!parse_dict_trim (lexer, dict))
 147         goto error;
 148     }
 149
 150   pgm->map = finish_case_map (dict);
 151
 152   dict_destroy (dataset_dict (ds));
 153   dataset_set_dict (ds, dict);
 154
 155   proc_set_source (ds,
 156                    create_case_source (&case_reader_source_class, pgm));
 157
 158   return CMD_SUCCESS;
 159
 160  error:
 161   case_reader_pgm_free (pgm);
 162   if (dict != NULL)
 163     dict_destroy (dict);
 164   return CMD_CASCADING_FAILURE;
 165 }
 166
 167 /* Frees a struct case_reader_pgm. */
 168 static void
 169 case_reader_pgm_free (struct case_reader_pgm *pgm)
 170 {
 171   if (pgm != NULL)
 172     {
 173       any_reader_close (pgm->reader);
 174       destroy_case_map (pgm->map);
 175       case_destroy (&pgm->bounce);
 176       free (pgm);
 177     }
 178 }
 179
 180 /* Clears internal state related to case reader input procedure. */
 181 static void
 182 case_reader_source_destroy (struct case_source *source)
 183 {
 184   struct case_reader_pgm *pgm = source->aux;
 185   case_reader_pgm_free (pgm);
 186 }
 187
 188 /* Reads all the cases from the data file into C and passes them
 189    to WRITE_CASE one by one, passing WC_DATA.
 190    Returns true if successful, false if an I/O error occurred. */
 191 static bool
 192 case_reader_source_read (struct case_source *source,
 193                          struct ccase *c,
 194                          write_case_func *write_case, write_case_data wc_data)
 195 {
 196   struct case_reader_pgm *pgm = source->aux;
 197   bool ok = true;
 198
 199   do
 200     {
 201       bool got_case;
 202       if (pgm->map == NULL)
 203         got_case = any_reader_read (pgm->reader, c);
 204       else
 205         {
 206           got_case = any_reader_read (pgm->reader, &pgm->bounce);
 207           if (got_case)
 208             map_case (pgm->map, &pgm->bounce, c);
 209         }
 210       if (!got_case)
 211         break;
 212
 213       ok = write_case (wc_data);
 214     }
 215   while (ok);
 216
 217   return ok && !any_reader_error (pgm->reader);
 218 }
 219
 220 static const struct case_source_class case_reader_source_class =
 221   {
 222     "case reader",
 223     NULL,
 224     case_reader_source_read,
 225     case_reader_source_destroy,
 226   };
 227 \f
 228 /* GET. */
 229 int
 230 cmd_get (struct lexer *lexer, struct dataset *ds)
 231 {
 232   return parse_read_command (lexer, ds, GET_CMD);
 233 }
 234
 235 /* IMPORT. */
 236 int
 237 cmd_import (struct lexer *lexer, struct dataset *ds)
 238 {
 239   return parse_read_command (lexer, ds, IMPORT_CMD);
 240 }
 241 \f
 242 /* Writing system and portable files. */
 243
 244 /* Type of output file. */
 245 enum writer_type
 246   {
 247     SYSFILE_WRITER,     /* System file. */
 248     PORFILE_WRITER      /* Portable file. */
 249   };
 250
 251 /* Type of a command. */
 252 enum command_type
 253   {
 254     XFORM_CMD,          /* Transformation. */
 255     PROC_CMD            /* Procedure. */
 256   };
 257
 258 /* File writer plus a case map. */
 259 struct case_writer
 260   {
 261     struct any_writer *writer;  /* File writer. */
 262     struct case_map *map;       /* Map to output file dictionary
 263                                    (null pointer for identity mapping). */
 264     struct ccase bounce;        /* Bounce buffer for mapping (if needed). */
 265   };
 266
 267 /* Destroys AW. */
 268 static bool
 269 case_writer_destroy (struct case_writer *aw)
 270 {
 271   bool ok = true;
 272   if (aw != NULL)
 273     {
 274       ok = any_writer_close (aw->writer);
 275       destroy_case_map (aw->map);
 276       case_destroy (&aw->bounce);
 277       free (aw);
 278     }
 279   return ok;
 280 }
 281
 282 /* Parses SAVE or XSAVE or EXPORT or XEXPORT command.
 283    WRITER_TYPE identifies the type of file to write,
 284    and COMMAND_TYPE identifies the type of command.
 285
 286    On success, returns a writer.
 287    For procedures only, sets *RETAIN_UNSELECTED to true if cases
 288    that would otherwise be excluded by FILTER or USE should be
 289    included.
 290
 291    On failure, returns a null pointer. */
 292 static struct case_writer *
 293 parse_write_command (struct lexer *lexer, struct dataset *ds,
 294                      enum writer_type writer_type,
 295                      enum command_type command_type,
 296                      bool *retain_unselected)
 297 {
 298   /* Common data. */
 299   struct file_handle *handle; /* Output file. */
 300   struct dictionary *dict;    /* Dictionary for output file. */
 301   struct case_writer *aw;      /* Writer. */
 302
 303   /* Common options. */
 304   bool print_map;             /* Print map?  TODO. */
 305   bool print_short_names;     /* Print long-to-short name map.  TODO. */
 306   struct sfm_write_options sysfile_opts;
 307   struct pfm_write_options porfile_opts;
 308
 309   assert (writer_type == SYSFILE_WRITER || writer_type == PORFILE_WRITER);
 310   assert (command_type == XFORM_CMD || command_type == PROC_CMD);
 311   assert ((retain_unselected != NULL) == (command_type == PROC_CMD));
 312
 313   if (command_type == PROC_CMD)
 314     *retain_unselected = true;
 315
 316   handle = NULL;
 317   dict = dict_clone (dataset_dict (ds));
 318   aw = xmalloc (sizeof *aw);
 319   aw->writer = NULL;
 320   aw->map = NULL;
 321   case_nullify (&aw->bounce);
 322   print_map = false;
 323   print_short_names = false;
 324   sysfile_opts = sfm_writer_default_options ();
 325   porfile_opts = pfm_writer_default_options ();
 326
 327   start_case_map (dict);
 328   dict_delete_scratch_vars (dict);
 329
 330   lex_match (lexer, '/');
 331   for (;;)
 332     {
 333       if (lex_match_id (lexer, "OUTFILE"))
 334         {
 335           if (handle != NULL)
 336             {
 337               lex_sbc_only_once ("OUTFILE");
 338               goto error;
 339             }
 340
 341           lex_match (lexer, '=');
 342
 343           handle = fh_parse (lexer, FH_REF_FILE | FH_REF_SCRATCH);
 344           if (handle == NULL)
 345             goto error;
 346         }
 347       else if (lex_match_id (lexer, "NAMES"))
 348         print_short_names = true;
 349       else if (lex_match_id (lexer, "PERMISSIONS"))
 350         {
 351           bool cw;
 352
 353           lex_match (lexer, '=');
 354           if (lex_match_id (lexer, "READONLY"))
 355             cw = false;
 356           else if (lex_match_id (lexer, "WRITEABLE"))
 357             cw = true;
 358           else
 359             {
 360               lex_error (lexer, _("expecting %s or %s"), "READONLY", "WRITEABLE");
 361               goto error;
 362             }
 363           sysfile_opts.create_writeable = porfile_opts.create_writeable = cw;
 364         }
 365       else if (command_type == PROC_CMD && lex_match_id (lexer, "UNSELECTED"))
 366         {
 367           lex_match (lexer, '=');
 368           if (lex_match_id (lexer, "RETAIN"))
 369             *retain_unselected = true;
 370           else if (lex_match_id (lexer, "DELETE"))
 371             *retain_unselected = false;
 372           else
 373             {
 374               lex_error (lexer, _("expecting %s or %s"), "RETAIN", "DELETE");
 375               goto error;
 376             }
 377         }
 378       else if (writer_type == SYSFILE_WRITER && lex_match_id (lexer, "COMPRESSED"))
 379         sysfile_opts.compress = true;
 380       else if (writer_type == SYSFILE_WRITER && lex_match_id (lexer, "UNCOMPRESSED"))
 381         sysfile_opts.compress = false;
 382       else if (writer_type == SYSFILE_WRITER && lex_match_id (lexer, "VERSION"))
 383         {
 384           lex_match (lexer, '=');
 385           if (!lex_force_int (lexer))
 386             goto error;
 387           sysfile_opts.version = lex_integer (lexer);
 388           lex_get (lexer);
 389         }
 390       else if (writer_type == PORFILE_WRITER && lex_match_id (lexer, "TYPE"))
 391         {
 392           lex_match (lexer, '=');
 393           if (lex_match_id (lexer, "COMMUNICATIONS"))
 394             porfile_opts.type = PFM_COMM;
 395           else if (lex_match_id (lexer, "TAPE"))
 396             porfile_opts.type = PFM_TAPE;
 397           else
 398             {
 399               lex_error (lexer, _("expecting %s or %s"), "COMM", "TAPE");
 400               goto error;
 401             }
 402         }
 403       else if (writer_type == PORFILE_WRITER && lex_match_id (lexer, "DIGITS"))
 404         {
 405           lex_match (lexer, '=');
 406           if (!lex_force_int (lexer))
 407             goto error;
 408           porfile_opts.digits = lex_integer (lexer);
 409           lex_get (lexer);
 410         }
 411       else if (!parse_dict_trim (lexer, dict))
 412         goto error;
 413
 414       if (!lex_match (lexer, '/'))
 415         break;
 416     }
 417   if (lex_end_of_command (lexer) != CMD_SUCCESS)
 418     goto error;
 419
 420   if (handle == NULL)
 421     {
 422       lex_sbc_missing (lexer, "OUTFILE");
 423       goto error;
 424     }
 425
 426   dict_compact_values (dict);
 427   aw->map = finish_case_map (dict);
 428   if (aw->map != NULL)
 429     case_create (&aw->bounce, dict_get_next_value_idx (dict));
 430
 431   if (fh_get_referent (handle) == FH_REF_FILE)
 432     {
 433       switch (writer_type)
 434         {
 435         case SYSFILE_WRITER:
 436           aw->writer = any_writer_from_sfm_writer (
 437             sfm_open_writer (handle, dict, sysfile_opts));
 438           break;
 439         case PORFILE_WRITER:
 440           aw->writer = any_writer_from_pfm_writer (
 441             pfm_open_writer (handle, dict, porfile_opts));
 442           break;
 443         }
 444     }
 445   else
 446     aw->writer = any_writer_open (handle, dict);
 447   if (aw->writer == NULL)
 448     goto error;
 449   dict_destroy (dict);
 450
 451   return aw;
 452
 453  error:
 454   case_writer_destroy (aw);
 455   dict_destroy (dict);
 456   return NULL;
 457 }
 458
 459 /* Writes case C to writer AW. */
 460 static bool
 461 case_writer_write_case (struct case_writer *aw, const struct ccase *c)
 462 {
 463   if (aw->map != NULL)
 464     {
 465       map_case (aw->map, c, &aw->bounce);
 466       c = &aw->bounce;
 467     }
 468   return any_writer_write (aw->writer, c);
 469 }
 470 \f
 471 /* SAVE and EXPORT. */
 472
 473 static bool output_proc (const struct ccase *, void *, const struct dataset *);
 474
 475 /* Parses and performs the SAVE or EXPORT procedure. */
 476 static int
 477 parse_output_proc (struct lexer *lexer, struct dataset *ds, enum writer_type writer_type)
 478 {
 479   bool retain_unselected;
 480   struct variable *saved_filter_variable;
 481   struct case_writer *aw;
 482   bool ok;
 483
 484   aw = parse_write_command (lexer, ds, writer_type, PROC_CMD, &retain_unselected);
 485   if (aw == NULL)
 486     return CMD_CASCADING_FAILURE;
 487
 488   saved_filter_variable = dict_get_filter (dataset_dict (ds));
 489   if (retain_unselected)
 490     dict_set_filter (dataset_dict (ds), NULL);
 491   ok = procedure (ds, output_proc, aw);
 492   dict_set_filter (dataset_dict (ds), saved_filter_variable);
 493
 494   case_writer_destroy (aw);
 495   return ok ? CMD_SUCCESS : CMD_CASCADING_FAILURE;
 496 }
 497
 498 /* Writes case C to file. */
 499 static bool
 500 output_proc (const struct ccase *c, void *aw_, const struct dataset *ds UNUSED)
 501 {
 502   struct case_writer *aw = aw_;
 503   return case_writer_write_case (aw, c);
 504 }
 505
 506 int
 507 cmd_save (struct lexer *lexer, struct dataset *ds)
 508 {
 509   return parse_output_proc (lexer, ds, SYSFILE_WRITER);
 510 }
 511
 512 int
 513 cmd_export (struct lexer *lexer, struct dataset *ds)
 514 {
 515   return parse_output_proc (lexer, ds, PORFILE_WRITER);
 516 }
 517 \f
 518 /* XSAVE and XEXPORT. */
 519
 520 /* Transformation. */
 521 struct output_trns
 522   {
 523     struct case_writer *aw;      /* Writer. */
 524   };
 525
 526 static trns_proc_func output_trns_proc;
 527 static trns_free_func output_trns_free;
 528
 529 /* Parses the XSAVE or XEXPORT transformation command. */
 530 static int
 531 parse_output_trns (struct lexer *lexer, struct dataset *ds, enum writer_type writer_type)
 532 {
 533   struct output_trns *t = xmalloc (sizeof *t);
 534   t->aw = parse_write_command (lexer, ds, writer_type, XFORM_CMD, NULL);
 535   if (t->aw == NULL)
 536     {
 537       free (t);
 538       return CMD_CASCADING_FAILURE;
 539     }
 540
 541   add_transformation (ds, output_trns_proc, output_trns_free, t);
 542   return CMD_SUCCESS;
 543 }
 544
 545 /* Writes case C to the system file specified on XSAVE or XEXPORT. */
 546 static int
 547 output_trns_proc (void *trns_, struct ccase *c, casenumber case_num UNUSED)
 548 {
 549   struct output_trns *t = trns_;
 550   case_writer_write_case (t->aw, c);
 551   return TRNS_CONTINUE;
 552 }
 553
 554 /* Frees an XSAVE or XEXPORT transformation.
 555    Returns true if successful, false if an I/O error occurred. */
 556 static bool
 557 output_trns_free (void *trns_)
 558 {
 559   struct output_trns *t = trns_;
 560   bool ok = true;
 561
 562   if (t != NULL)
 563     {
 564       ok = case_writer_destroy (t->aw);
 565       free (t);
 566     }
 567   return ok;
 568 }
 569
 570 /* XSAVE command. */
 571 int
 572 cmd_xsave (struct lexer *lexer, struct dataset *ds)
 573 {
 574   return parse_output_trns (lexer, ds, SYSFILE_WRITER);
 575 }
 576
 577 /* XEXPORT command. */
 578 int
 579 cmd_xexport (struct lexer *lexer, struct dataset *ds)
 580 {
 581   return parse_output_trns (lexer, ds, PORFILE_WRITER);
 582 }
 583 \f
 584 static bool rename_variables (struct lexer *lexer, struct dictionary *dict);
 585 static bool drop_variables (struct lexer *, struct dictionary *dict);
 586 static bool keep_variables (struct lexer *, struct dictionary *dict);
 587
 588 /* Commands that read and write system files share a great deal
 589    of common syntactic structure for rearranging and dropping
 590    variables.  This function parses this syntax and modifies DICT
 591    appropriately.  Returns true on success, false on failure. */
 592 static bool
 593 parse_dict_trim (struct lexer *lexer, struct dictionary *dict)
 594 {
 595   if (lex_match_id (lexer, "MAP"))
 596     {
 597       /* FIXME. */
 598       return true;
 599     }
 600   else if (lex_match_id (lexer, "DROP"))
 601     return drop_variables (lexer, dict);
 602   else if (lex_match_id (lexer, "KEEP"))
 603     return keep_variables (lexer, dict);
 604   else if (lex_match_id (lexer, "RENAME"))
 605     return rename_variables (lexer, dict);
 606   else
 607     {
 608       lex_error (lexer, _("expecting a valid subcommand"));
 609       return false;
 610     }
 611 }
 612
 613 /* Parses and performs the RENAME subcommand of GET and SAVE. */
 614 static bool
 615 rename_variables (struct lexer *lexer, struct dictionary *dict)
 616 {
 617   size_t i;
 618
 619   int success = 0;
 620
 621   struct variable **v;
 622   char **new_names;
 623   size_t nv, nn;
 624   char *err_name;
 625
 626   int group;
 627
 628   lex_match (lexer, '=');
 629   if (lex_token (lexer) != '(')
 630     {
 631       struct variable *v;
 632
 633       v = parse_variable (lexer, dict);
 634       if (v == NULL)
 635         return 0;
 636       if (!lex_force_match (lexer, '=')
 637           || !lex_force_id (lexer))
 638         return 0;
 639       if (dict_lookup_var (dict, lex_tokid (lexer)) != NULL)
 640         {
 641           msg (SE, _("Cannot rename %s as %s because there already exists "
 642                      "a variable named %s.  To rename variables with "
 643                      "overlapping names, use a single RENAME subcommand "
 644                      "such as \"/RENAME (A=B)(B=C)(C=A)\", or equivalently, "
 645                      "\"/RENAME (A B C=B C A)\"."),
 646                var_get_name (v), lex_tokid (lexer), lex_tokid (lexer));
 647           return 0;
 648         }
 649
 650       dict_rename_var (dict, v, lex_tokid (lexer));
 651       lex_get (lexer);
 652       return 1;
 653     }
 654
 655   nv = nn = 0;
 656   v = NULL;
 657   new_names = 0;
 658   group = 1;
 659   while (lex_match (lexer, '('))
 660     {
 661       size_t old_nv = nv;
 662
 663       if (!parse_variables (lexer, dict, &v, &nv, PV_NO_DUPLICATE | PV_APPEND))
 664         goto done;
 665       if (!lex_match (lexer, '='))
 666         {
 667           msg (SE, _("`=' expected after variable list."));
 668           goto done;
 669         }
 670       if (!parse_DATA_LIST_vars (lexer, &new_names, &nn, PV_APPEND | PV_NO_SCRATCH))
 671         goto done;
 672       if (nn != nv)
 673         {
 674           msg (SE, _("Number of variables on left side of `=' (%d) does not "
 675                      "match number of variables on right side (%d), in "
 676                      "parenthesized group %d of RENAME subcommand."),
 677                (unsigned) (nv - old_nv), (unsigned) (nn - old_nv), group);
 678           goto done;
 679         }
 680       if (!lex_force_match (lexer, ')'))
 681         goto done;
 682       group++;
 683     }
 684
 685   if (!dict_rename_vars (dict, v, new_names, nv, &err_name))
 686     {
 687       msg (SE, _("Requested renaming duplicates variable name %s."), err_name);
 688       goto done;
 689     }
 690   success = 1;
 691
 692  done:
 693   for (i = 0; i < nn; i++)
 694     free (new_names[i]);
 695   free (new_names);
 696   free (v);
 697
 698   return success;
 699 }
 700
 701 /* Parses and performs the DROP subcommand of GET and SAVE.
 702    Returns true if successful, false on failure.*/
 703 static bool
 704 drop_variables (struct lexer *lexer, struct dictionary *dict)
 705 {
 706   struct variable **v;
 707   size_t nv;
 708
 709   lex_match (lexer, '=');
 710   if (!parse_variables (lexer, dict, &v, &nv, PV_NONE))
 711     return false;
 712   dict_delete_vars (dict, v, nv);
 713   free (v);
 714
 715   if (dict_get_var_cnt (dict) == 0)
 716     {
 717       msg (SE, _("Cannot DROP all variables from dictionary."));
 718       return false;
 719     }
 720   return true;
 721 }
 722
 723 /* Parses and performs the KEEP subcommand of GET and SAVE.
 724    Returns true if successful, false on failure.*/
 725 static bool
 726 keep_variables (struct lexer *lexer, struct dictionary *dict)
 727 {
 728   struct variable **v;
 729   size_t nv;
 730   size_t i;
 731
 732   lex_match (lexer, '=');
 733   if (!parse_variables (lexer, dict, &v, &nv, PV_NONE))
 734     return false;
 735
 736   /* Move the specified variables to the beginning. */
 737   dict_reorder_vars (dict, v, nv);
 738
 739   /* Delete the remaining variables. */
 740   v = xnrealloc (v, dict_get_var_cnt (dict) - nv, sizeof *v);
 741   for (i = nv; i < dict_get_var_cnt (dict); i++)
 742     v[i - nv] = dict_get_var (dict, i);
 743   dict_delete_vars (dict, v, dict_get_var_cnt (dict) - nv);
 744   free (v);
 745
 746   return true;
 747 }
 748 \f
 749 /* MATCH FILES. */
 750
 751 /* File types. */
 752 enum
 753   {
 754     MTF_FILE,                   /* Specified on FILE= subcommand. */
 755     MTF_TABLE                   /* Specified on TABLE= subcommand. */
 756   };
 757
 758 /* One of the files on MATCH FILES. */
 759 struct mtf_file
 760   {
 761     struct mtf_file *next, *prev; /* Next, previous in the list of files. */
 762     struct mtf_file *next_min;  /* Next in the chain of minimums. */
 763
 764     int type;                   /* One of MTF_*. */
 765     struct variable **by;       /* List of BY variables for this file. */
 766     struct file_handle *handle; /* File handle. */
 767     struct any_reader *reader;  /* File reader. */
 768     struct dictionary *dict;    /* Dictionary from system file. */
 769
 770     /* IN subcommand. */
 771     char *in_name;              /* Variable name. */
 772     struct variable *in_var;    /* Variable (in master dictionary). */
 773
 774     struct ccase input;         /* Input record. */
 775   };
 776
 777 /* MATCH FILES procedure. */
 778 struct mtf_proc
 779   {
 780     struct mtf_file *head;      /* First file mentioned on FILE or TABLE. */
 781     struct mtf_file *tail;      /* Last file mentioned on FILE or TABLE. */
 782
 783     bool ok;                    /* False if I/O error occurs. */
 784
 785     size_t by_cnt;              /* Number of variables on BY subcommand. */
 786
 787     /* Names of FIRST, LAST variables. */
 788     char first[LONG_NAME_LEN + 1], last[LONG_NAME_LEN + 1];
 789
 790     struct dictionary *dict;    /* Dictionary of output file. */
 791     struct casefile *output;    /* MATCH FILES output. */
 792     struct ccase mtf_case;      /* Case used for output. */
 793
 794     unsigned seq_num;           /* Have we initialized this variable? */
 795     unsigned *seq_nums;         /* Sequence numbers for each var in dict. */
 796   };
 797
 798 static bool mtf_free (struct mtf_proc *);
 799 static bool mtf_close_file (struct mtf_file *);
 800 static int mtf_merge_dictionary (struct dictionary *const, struct mtf_file *);
 801 static bool mtf_delete_file_in_place (struct mtf_proc *, struct mtf_file **);
 802
 803 static bool mtf_read_nonactive_records (void *);
 804 static bool mtf_processing_finish (void *, const struct dataset *);
 805 static bool mtf_processing (const struct ccase *, void *, const struct dataset *);
 806
 807 static char *var_type_description (struct variable *);
 808
 809 static void set_master (struct variable *, struct variable *master);
 810 static struct variable *get_master (struct variable *);
 811
 812 /* Parse and execute the MATCH FILES command. */
 813 int
 814 cmd_match_files (struct lexer *lexer, struct dataset *ds)
 815 {
 816   struct mtf_proc mtf;
 817   struct mtf_file *first_table = NULL;
 818   struct mtf_file *iter;
 819
 820   bool used_active_file = false;
 821   bool saw_table = false;
 822   bool saw_in = false;
 823
 824   bool ok;
 825
 826   mtf.head = mtf.tail = NULL;
 827   mtf.by_cnt = 0;
 828   mtf.first[0] = '\0';
 829   mtf.last[0] = '\0';
 830   mtf.dict = dict_create ();
 831   mtf.output = NULL;
 832   case_nullify (&mtf.mtf_case);
 833   mtf.seq_num = 0;
 834   mtf.seq_nums = NULL;
 835   dict_set_case_limit (mtf.dict, dict_get_case_limit (dataset_dict (ds)));
 836
 837   lex_match (lexer, '/');
 838   while (lex_token (lexer) == T_ID
 839          && (lex_id_match (ss_cstr ("FILE"), ss_cstr (lex_tokid (lexer)))
 840              || lex_id_match (ss_cstr ("TABLE"), ss_cstr (lex_tokid (lexer)))))
 841     {
 842       struct mtf_file *file = xmalloc (sizeof *file);
 843
 844       if (lex_match_id (lexer, "FILE"))
 845         file->type = MTF_FILE;
 846       else if (lex_match_id (lexer, "TABLE"))
 847         {
 848           file->type = MTF_TABLE;
 849           saw_table = true;
 850         }
 851       else
 852         NOT_REACHED ();
 853       lex_match (lexer, '=');
 854
 855       file->by = NULL;
 856       file->handle = NULL;
 857       file->reader = NULL;
 858       file->dict = NULL;
 859       file->in_name = NULL;
 860       file->in_var = NULL;
 861       case_nullify (&file->input);
 862
 863       /* FILEs go first, then TABLEs. */
 864       if (file->type == MTF_TABLE || first_table == NULL)
 865         {
 866           file->next = NULL;
 867           file->prev = mtf.tail;
 868           if (mtf.tail)
 869             mtf.tail->next = file;
 870           mtf.tail = file;
 871           if (mtf.head == NULL)
 872             mtf.head = file;
 873           if (file->type == MTF_TABLE && first_table == NULL)
 874             first_table = file;
 875         }
 876       else
 877         {
 878           assert (file->type == MTF_FILE);
 879           file->next = first_table;
 880           file->prev = first_table->prev;
 881           if (first_table->prev)
 882             first_table->prev->next = file;
 883           else
 884             mtf.head = file;
 885           first_table->prev = file;
 886         }
 887
 888       if (lex_match (lexer, '*'))
 889         {
 890           file->handle = NULL;
 891           file->reader = NULL;
 892
 893           if (used_active_file)
 894             {
 895               msg (SE, _("The active file may not be specified more "
 896                          "than once."));
 897               goto error;
 898             }
 899           used_active_file = true;
 900
 901           if (!proc_has_source (ds))
 902             {
 903               msg (SE, _("Cannot specify the active file since no active "
 904                          "file has been defined."));
 905               goto error;
 906             }
 907
 908           if (proc_make_temporary_transformations_permanent (ds))
 909             msg (SE,
 910                  _("MATCH FILES may not be used after TEMPORARY when "
 911                    "the active file is an input source.  "
 912                    "Temporary transformations will be made permanent."));
 913
 914           file->dict = dataset_dict (ds);
 915         }
 916       else
 917         {
 918           file->handle = fh_parse (lexer, FH_REF_FILE | FH_REF_SCRATCH);
 919           if (file->handle == NULL)
 920             goto error;
 921
 922           file->reader = any_reader_open (file->handle, &file->dict);
 923           if (file->reader == NULL)
 924             goto error;
 925
 926           case_create (&file->input, dict_get_next_value_idx (file->dict));
 927         }
 928
 929       while (lex_match (lexer, '/'))
 930         if (lex_match_id (lexer, "RENAME"))
 931           {
 932             if (!rename_variables (lexer, file->dict))
 933               goto error;
 934           }
 935         else if (lex_match_id (lexer, "IN"))
 936           {
 937             lex_match (lexer, '=');
 938             if (lex_token (lexer) != T_ID)
 939               {
 940                 lex_error (lexer, NULL);
 941                 goto error;
 942               }
 943
 944             if (file->in_name != NULL)
 945               {
 946                 msg (SE, _("Multiple IN subcommands for a single FILE or "
 947                            "TABLE."));
 948                 goto error;
 949               }
 950             file->in_name = xstrdup (lex_tokid (lexer));
 951             lex_get (lexer);
 952             saw_in = true;
 953           }
 954
 955       mtf_merge_dictionary (mtf.dict, file);
 956     }
 957
 958   while (lex_token (lexer) != '.')
 959     {
 960       if (lex_match (lexer, T_BY))
 961         {
 962           struct variable **by;
 963
 964           if (mtf.by_cnt)
 965             {
 966               msg (SE, _("BY may appear at most once."));
 967               goto error;
 968             }
 969
 970           lex_match (lexer, '=');
 971           if (!parse_variables (lexer, mtf.dict, &by, &mtf.by_cnt,
 972                                 PV_NO_DUPLICATE | PV_NO_SCRATCH))
 973             goto error;
 974
 975           for (iter = mtf.head; iter != NULL; iter = iter->next)
 976             {
 977               size_t i;
 978
 979               iter->by = xnmalloc (mtf.by_cnt, sizeof *iter->by);
 980
 981               for (i = 0; i < mtf.by_cnt; i++)
 982                 {
 983                   iter->by[i] = dict_lookup_var (iter->dict,
 984                                                  var_get_name (by[i]));
 985                   if (iter->by[i] == NULL)
 986                     {
 987                       msg (SE, _("File %s lacks BY variable %s."),
 988                            iter->handle ? fh_get_name (iter->handle) : "*",
 989                            var_get_name (by[i]));
 990                       free (by);
 991                       goto error;
 992                     }
 993                 }
 994             }
 995           free (by);
 996         }
 997       else if (lex_match_id (lexer, "FIRST"))
 998         {
 999           if (mtf.first[0] != '\0')
1000             {
1001               msg (SE, _("FIRST may appear at most once."));
1002               goto error;
1003             }
1004
1005           lex_match (lexer, '=');
1006           if (!lex_force_id (lexer))
1007             goto error;
1008           strcpy (mtf.first, lex_tokid (lexer));
1009           lex_get (lexer);
1010         }
1011       else if (lex_match_id (lexer, "LAST"))
1012         {
1013           if (mtf.last[0] != '\0')
1014             {
1015               msg (SE, _("LAST may appear at most once."));
1016               goto error;
1017             }
1018
1019           lex_match (lexer, '=');
1020           if (!lex_force_id (lexer))
1021             goto error;
1022           strcpy (mtf.last, lex_tokid (lexer));
1023           lex_get (lexer);
1024         }
1025       else if (lex_match_id (lexer, "MAP"))
1026         {
1027           /* FIXME. */
1028         }
1029       else if (lex_match_id (lexer, "DROP"))
1030         {
1031           if (!drop_variables (lexer, mtf.dict))
1032             goto error;
1033         }
1034       else if (lex_match_id (lexer, "KEEP"))
1035         {
1036           if (!keep_variables (lexer, mtf.dict))
1037             goto error;
1038         }
1039       else
1040         {
1041           lex_error (lexer, NULL);
1042           goto error;
1043         }
1044
1045       if (!lex_match (lexer, '/') && lex_token (lexer) != '.')
1046         {
1047           lex_end_of_command (lexer);
1048           goto error;
1049         }
1050     }
1051
1052   if (mtf.by_cnt == 0)
1053     {
1054       if (saw_table)
1055         {
1056           msg (SE, _("BY is required when TABLE is specified."));
1057           goto error;
1058         }
1059       if (saw_in)
1060         {
1061           msg (SE, _("BY is required when IN is specified."));
1062           goto error;
1063         }
1064     }
1065
1066   /* Set up mapping from each file's variables to master
1067      variables. */
1068   for (iter = mtf.head; iter != NULL; iter = iter->next)
1069     {
1070       struct dictionary *d = iter->dict;
1071       int i;
1072
1073       for (i = 0; i < dict_get_var_cnt (d); i++)
1074         {
1075           struct variable *v = dict_get_var (d, i);
1076           struct variable *mv = dict_lookup_var (mtf.dict, var_get_name (v));
1077           if (mv != NULL)
1078             set_master (v, mv);
1079         }
1080     }
1081
1082   /* Add IN variables to master dictionary. */
1083   for (iter = mtf.head; iter != NULL; iter = iter->next)
1084     if (iter->in_name != NULL)
1085       {
1086         struct fmt_spec format = fmt_for_output (FMT_F, 1, 0);
1087         iter->in_var = dict_create_var (mtf.dict, iter->in_name, 0);
1088         if (iter->in_var == NULL)
1089           {
1090             msg (SE, _("IN variable name %s duplicates an "
1091                        "existing variable name."),
1092                  var_get_name (iter->in_var));
1093             goto error;
1094           }
1095         var_set_both_formats (iter->in_var, &format);
1096       }
1097
1098   /* MATCH FILES performs an n-way merge on all its input files.
1099      Abstract algorithm:
1100
1101      1. Read one input record from every input FILE.
1102
1103      2. If no FILEs are left, stop.  Otherwise, proceed to step 3.
1104
1105      3. Find the FILE input record(s) that have minimum BY
1106      values.  Store all the values from these input records into
1107      the output record.
1108
1109      4. For every TABLE, read another record as long as the BY values
1110      on the TABLE's input record are less than the FILEs' BY values.
1111      If an exact match is found, store all the values from the TABLE
1112      input record into the output record.
1113
1114      5. Write the output record.
1115
1116      6. Read another record from each input file FILE and TABLE that
1117      we stored values from above.  If we come to the end of one of the
1118      input files, remove it from the list of input files.
1119
1120      7. Repeat from step 2.
1121
1122      Unfortunately, this algorithm can't be implemented in a
1123      straightforward way because there's no function to read a
1124      record from the active file.  Instead, it has to be written
1125      as a state machine.
1126
1127      FIXME: For merging large numbers of files (more than 10?) a
1128      better algorithm would use a heap for finding minimum
1129      values. */
1130
1131   if (!used_active_file)
1132     discard_variables (ds);
1133
1134   dict_compact_values (mtf.dict);
1135   mtf.output = fastfile_create (dict_get_next_value_idx (mtf.dict));
1136   mtf.seq_nums = xcalloc (dict_get_var_cnt (mtf.dict), sizeof *mtf.seq_nums);
1137   case_create (&mtf.mtf_case, dict_get_next_value_idx (mtf.dict));
1138
1139   if (!mtf_read_nonactive_records (&mtf))
1140     goto error;
1141
1142   if (used_active_file)
1143     {
1144       proc_set_sink (ds,
1145                      create_case_sink (&null_sink_class,
1146                                        dataset_dict (ds), NULL));
1147       ok =
1148         ( procedure (ds, mtf_processing, &mtf) &&
1149           mtf_processing_finish (&mtf, ds) );
1150     }
1151   else
1152     ok = mtf_processing_finish (&mtf, ds);
1153
1154   discard_variables (ds);
1155
1156   dict_destroy (dataset_dict (ds));
1157   dataset_set_dict (ds, mtf.dict);
1158   mtf.dict = NULL;
1159   proc_set_source (ds, storage_source_create (mtf.output));
1160   mtf.output = NULL;
1161
1162   if (!mtf_free (&mtf))
1163     ok = false;
1164   return ok ? CMD_SUCCESS : CMD_CASCADING_FAILURE;
1165
1166  error:
1167   mtf_free (&mtf);
1168   return CMD_CASCADING_FAILURE;
1169 }
1170
1171 /* Repeats 2...7 an arbitrary number of times. */
1172 static bool
1173 mtf_processing_finish (void *mtf_, const struct dataset *ds)
1174 {
1175   struct mtf_proc *mtf = mtf_;
1176   struct mtf_file *iter;
1177
1178   /* Find the active file and delete it. */
1179   for (iter = mtf->head; iter; iter = iter->next)
1180     if (iter->handle == NULL)
1181       {
1182         if (!mtf_delete_file_in_place (mtf, &iter))
1183           NOT_REACHED ();
1184         break;
1185       }
1186
1187   while (mtf->head && mtf->head->type == MTF_FILE)
1188     if (!mtf_processing (NULL, mtf, ds))
1189       return false;
1190
1191   return true;
1192 }
1193
1194 /* Return a string in a static buffer describing V's variable type and
1195    width. */
1196 static char *
1197 var_type_description (struct variable *v)
1198 {
1199   static char buf[2][32];
1200   static int x = 0;
1201   char *s;
1202
1203   x ^= 1;
1204   s = buf[x];
1205
1206   if (var_is_numeric (v))
1207     strcpy (s, "numeric");
1208   else
1209     sprintf (s, "string with width %d", var_get_width (v));
1210   return s;
1211 }
1212
1213 /* Closes FILE and frees its associated data.
1214    Returns true if successful, false if an I/O error
1215    occurred on FILE. */
1216 static bool
1217 mtf_close_file (struct mtf_file *file)
1218 {
1219   bool ok = file->reader == NULL || !any_reader_error (file->reader);
1220   free (file->by);
1221   any_reader_close (file->reader);
1222   if (file->handle != NULL)
1223     dict_destroy (file->dict);
1224   case_destroy (&file->input);
1225   free (file->in_name);
1226   free (file);
1227   return ok;
1228 }
1229
1230 /* Free all the data for the MATCH FILES procedure.
1231    Returns true if successful, false if an I/O error
1232    occurred. */
1233 static bool
1234 mtf_free (struct mtf_proc *mtf)
1235 {
1236   struct mtf_file *iter, *next;
1237   bool ok = true;
1238
1239   for (iter = mtf->head; iter; iter = next)
1240     {
1241       next = iter->next;
1242       assert (iter->dict != mtf->dict);
1243       if (!mtf_close_file (iter))
1244         ok = false;
1245     }
1246
1247   if (mtf->dict)
1248     dict_destroy (mtf->dict);
1249   case_destroy (&mtf->mtf_case);
1250   free (mtf->seq_nums);
1251
1252   return ok;
1253 }
1254
1255 /* Remove *FILE from the mtf_file chain.  Make *FILE point to the next
1256    file in the chain, or to NULL if was the last in the chain.
1257    Returns true if successful, false if an I/O error occurred. */
1258 static bool
1259 mtf_delete_file_in_place (struct mtf_proc *mtf, struct mtf_file **file)
1260 {
1261   struct mtf_file *f = *file;
1262   int i;
1263
1264   if (f->prev)
1265     f->prev->next = f->next;
1266   if (f->next)
1267     f->next->prev = f->prev;
1268   if (f == mtf->head)
1269     mtf->head = f->next;
1270   if (f == mtf->tail)
1271     mtf->tail = f->prev;
1272   *file = f->next;
1273
1274   if (f->in_var != NULL)
1275     case_data_rw (&mtf->mtf_case, f->in_var->fv)->f = 0.;
1276   for (i = 0; i < dict_get_var_cnt (f->dict); i++)
1277     {
1278       struct variable *v = dict_get_var (f->dict, i);
1279       struct variable *mv = get_master (v);
1280       if (mv != NULL)
1281         {
1282           union value *out = case_data_rw (&mtf->mtf_case, mv->fv);
1283
1284           if (var_is_numeric (v))
1285             out->f = SYSMIS;
1286           else
1287             memset (out->s, ' ', var_get_width (v));
1288         }
1289     }
1290
1291   return mtf_close_file (f);
1292 }
1293
1294 /* Read a record from every input file except the active file.
1295    Returns true if successful, false if an I/O error occurred. */
1296 static bool
1297 mtf_read_nonactive_records (void *mtf_)
1298 {
1299   struct mtf_proc *mtf = mtf_;
1300   struct mtf_file *iter, *next;
1301   bool ok = true;
1302
1303   for (iter = mtf->head; ok && iter != NULL; iter = next)
1304     {
1305       next = iter->next;
1306       if (iter->handle && !any_reader_read (iter->reader, &iter->input))
1307         if (!mtf_delete_file_in_place (mtf, &iter))
1308           ok = false;
1309     }
1310   return ok;
1311 }
1312
1313 /* Compare the BY variables for files A and B; return -1 if A < B, 0
1314    if A == B, 1 if A > B. */
1315 static inline int
1316 mtf_compare_BY_values (struct mtf_proc *mtf,
1317                        struct mtf_file *a, struct mtf_file *b,
1318                        const struct ccase *c)
1319 {
1320   const struct ccase *ca = case_is_null (&a->input) ? c : &a->input;
1321   const struct ccase *cb = case_is_null (&b->input) ? c : &b->input;
1322   assert ((a == NULL) + (b == NULL) + (c == NULL) <= 1);
1323   return case_compare_2dict (ca, cb, a->by, b->by, mtf->by_cnt);
1324 }
1325
1326 /* Perform one iteration of steps 3...7 above.
1327    Returns true if successful, false if an I/O error occurred. */
1328 static bool
1329 mtf_processing (const struct ccase *c, void *mtf_, const struct dataset *ds UNUSED)
1330 {
1331   struct mtf_proc *mtf = mtf_;
1332
1333   /* Do we need another record from the active file? */
1334   bool read_active_file;
1335
1336   assert (mtf->head != NULL);
1337   if (mtf->head->type == MTF_TABLE)
1338     return true;
1339
1340   do
1341     {
1342       struct mtf_file *min_head, *min_tail; /* Files with minimum BY values. */
1343       struct mtf_file *max_head, *max_tail; /* Files with non-minimum BYs. */
1344       struct mtf_file *iter, *next;
1345
1346       read_active_file = false;
1347
1348       /* 3. Find the FILE input record(s) that have minimum BY
1349          values.  Store all the values from these input records into
1350          the output record. */
1351       min_head = min_tail = mtf->head;
1352       max_head = max_tail = NULL;
1353       for (iter = mtf->head->next; iter && iter->type == MTF_FILE;
1354            iter = iter->next)
1355         {
1356           int cmp = mtf_compare_BY_values (mtf, min_head, iter, c);
1357           if (cmp < 0)
1358             {
1359               if (max_head)
1360                 max_tail = max_tail->next_min = iter;
1361               else
1362                 max_head = max_tail = iter;
1363             }
1364           else if (cmp == 0)
1365             min_tail = min_tail->next_min = iter;
1366           else /* cmp > 0 */
1367             {
1368               if (max_head)
1369                 {
1370                   max_tail->next_min = min_head;
1371                   max_tail = min_tail;
1372                 }
1373               else
1374                 {
1375                   max_head = min_head;
1376                   max_tail = min_tail;
1377                 }
1378               min_head = min_tail = iter;
1379             }
1380         }
1381
1382       /* 4. For every TABLE, read another record as long as the BY
1383          values on the TABLE's input record are less than the FILEs'
1384          BY values.  If an exact match is found, store all the values
1385          from the TABLE input record into the output record. */
1386       for (; iter != NULL; iter = next)
1387         {
1388           assert (iter->type == MTF_TABLE);
1389
1390           next = iter->next;
1391           for (;;)
1392             {
1393               int cmp = mtf_compare_BY_values (mtf, min_head, iter, c);
1394               if (cmp < 0)
1395                 {
1396                   if (max_head)
1397                     max_tail = max_tail->next_min = iter;
1398                   else
1399                     max_head = max_tail = iter;
1400                 }
1401               else if (cmp == 0)
1402                 min_tail = min_tail->next_min = iter;
1403               else /* cmp > 0 */
1404                 {
1405                   if (iter->handle == NULL)
1406                     return true;
1407                   if (any_reader_read (iter->reader, &iter->input))
1408                     continue;
1409                   if (!mtf_delete_file_in_place (mtf, &iter))
1410                     return false;
1411                 }
1412               break;
1413             }
1414         }
1415
1416       /* Next sequence number. */
1417       mtf->seq_num++;
1418
1419       /* Store data to all the records we are using. */
1420       if (min_tail)
1421         min_tail->next_min = NULL;
1422       for (iter = min_head; iter; iter = iter->next_min)
1423         {
1424           int i;
1425
1426           for (i = 0; i < dict_get_var_cnt (iter->dict); i++)
1427             {
1428               struct variable *v = dict_get_var (iter->dict, i);
1429               struct variable *mv = get_master (v);
1430
1431               if (mv != NULL && mtf->seq_nums[mv->index] != mtf->seq_num)
1432                 {
1433                   const struct ccase *record
1434                     = case_is_null (&iter->input) ? c : &iter->input;
1435                   union value *out = case_data_rw (&mtf->mtf_case, mv->fv);
1436
1437                   mtf->seq_nums[mv->index] = mtf->seq_num;
1438                   if (var_is_numeric (v))
1439                     out->f = case_num (record, v->fv);
1440                   else
1441                     memcpy (out->s, case_str (record, v->fv),
1442                             var_get_width (v));
1443                 }
1444             }
1445           if (iter->in_var != NULL)
1446             case_data_rw (&mtf->mtf_case, iter->in_var->fv)->f = 1.;
1447
1448           if (iter->type == MTF_FILE && iter->handle == NULL)
1449             read_active_file = true;
1450         }
1451
1452       /* Store missing values to all the records we're not
1453          using. */
1454       if (max_tail)
1455         max_tail->next_min = NULL;
1456       for (iter = max_head; iter; iter = iter->next_min)
1457         {
1458           int i;
1459
1460           for (i = 0; i < dict_get_var_cnt (iter->dict); i++)
1461             {
1462               struct variable *v = dict_get_var (iter->dict, i);
1463               struct variable *mv = get_master (v);
1464
1465               if (mv != NULL && mtf->seq_nums[mv->index] != mtf->seq_num)
1466                 {
1467                   union value *out = case_data_rw (&mtf->mtf_case, mv->fv);
1468                   mtf->seq_nums[mv->index] = mtf->seq_num;
1469
1470                   if (var_is_numeric (v))
1471                     out->f = SYSMIS;
1472                   else
1473                     memset (out->s, ' ', var_get_width (v));
1474                 }
1475             }
1476           if (iter->in_var != NULL)
1477             case_data_rw (&mtf->mtf_case, iter->in_var->fv)->f = 0.;
1478         }
1479
1480       /* 5. Write the output record. */
1481       casefile_append (mtf->output, &mtf->mtf_case);
1482
1483       /* 6. Read another record from each input file FILE and TABLE
1484          that we stored values from above.  If we come to the end of
1485          one of the input files, remove it from the list of input
1486          files. */
1487       for (iter = min_head; iter && iter->type == MTF_FILE; iter = next)
1488         {
1489           next = iter->next_min;
1490           if (iter->reader != NULL
1491               && !any_reader_read (iter->reader, &iter->input))
1492             if (!mtf_delete_file_in_place (mtf, &iter))
1493               return false;
1494         }
1495     }
1496   while (!read_active_file
1497          && mtf->head != NULL && mtf->head->type == MTF_FILE);
1498
1499   return true;
1500 }
1501
1502 /* Merge the dictionary for file F into master dictionary M. */
1503 static int
1504 mtf_merge_dictionary (struct dictionary *const m, struct mtf_file *f)
1505 {
1506   struct dictionary *d = f->dict;
1507   const char *d_docs, *m_docs;
1508   int i;
1509
1510   if (dict_get_label (m) == NULL)
1511     dict_set_label (m, dict_get_label (d));
1512
1513   d_docs = dict_get_documents (d);
1514   m_docs = dict_get_documents (m);
1515   if (d_docs != NULL)
1516     {
1517       if (m_docs == NULL)
1518         dict_set_documents (m, d_docs);
1519       else
1520         {
1521           char *new_docs;
1522           size_t new_len;
1523
1524           new_len = strlen (m_docs) + strlen (d_docs);
1525           new_docs = xmalloc (new_len + 1);
1526           strcpy (new_docs, m_docs);
1527           strcat (new_docs, d_docs);
1528           dict_set_documents (m, new_docs);
1529           free (new_docs);
1530         }
1531     }
1532
1533   for (i = 0; i < dict_get_var_cnt (d); i++)
1534     {
1535       struct variable *dv = dict_get_var (d, i);
1536       struct variable *mv = dict_lookup_var (m, var_get_name (dv));
1537
1538       if (dict_class_from_id (var_get_name (dv)) == DC_SCRATCH)
1539         continue;
1540
1541       if (mv != NULL)
1542         {
1543           if (var_get_width (mv) != var_get_width (dv))
1544             {
1545               msg (SE, _("Variable %s in file %s (%s) has different "
1546                          "type or width from the same variable in "
1547                          "earlier file (%s)."),
1548                    var_get_name (dv), fh_get_name (f->handle),
1549                    var_type_description (dv), var_type_description (mv));
1550               return 0;
1551             }
1552
1553           if (var_get_width (dv) == var_get_width (mv))
1554             {
1555               if (val_labs_count (dv->val_labs)
1556                   && !val_labs_count (mv->val_labs))
1557                 {
1558                   val_labs_destroy (mv->val_labs);
1559                   mv->val_labs = val_labs_copy (dv->val_labs);
1560                 }
1561               if (var_has_missing_values (dv) && !var_has_missing_values (mv))
1562                 var_set_missing_values (mv, var_get_missing_values (dv));
1563             }
1564
1565           if (var_get_label (dv) && !var_get_label (mv))
1566             var_set_label (mv, var_get_label (dv));
1567         }
1568       else
1569         mv = dict_clone_var_assert (m, dv, var_get_name (dv));
1570     }
1571
1572   return 1;
1573 }
1574
1575 /* Marks V's master variable as MASTER. */
1576 static void
1577 set_master (struct variable *v, struct variable *master)
1578 {
1579   var_attach_aux (v, master, NULL);
1580 }
1581
1582 /* Returns the master variable corresponding to V,
1583    as set with set_master(). */
1584 static struct variable *
1585 get_master (struct variable *v)
1586 {
1587   return v->aux;
1588 }
1589 \f
1590
1591 \f
1592 /* Case map.
1593
1594    A case map copies data from a case that corresponds for one
1595    dictionary to a case that corresponds to a second dictionary
1596    derived from the first by, optionally, deleting, reordering,
1597    or renaming variables.  (No new variables may be created.)
1598    */
1599
1600 /* A case map. */
1601 struct case_map
1602   {
1603     size_t value_cnt;   /* Number of values in map. */
1604     int *map;           /* For each destination index, the
1605                            corresponding source index. */
1606   };
1607
1608 /* Prepares dictionary D for producing a case map.  Afterward,
1609    the caller may delete, reorder, or rename variables within D
1610    at will before using finish_case_map() to produce the case
1611    map.
1612
1613    Uses D's aux members, which must otherwise not be in use. */
1614 static void
1615 start_case_map (struct dictionary *d)
1616 {
1617   size_t var_cnt = dict_get_var_cnt (d);
1618   size_t i;
1619
1620   for (i = 0; i < var_cnt; i++)
1621     {
1622       struct variable *v = dict_get_var (d, i);
1623       int *src_fv = xmalloc (sizeof *src_fv);
1624       *src_fv = v->fv;
1625       var_attach_aux (v, src_fv, var_dtor_free);
1626     }
1627 }
1628
1629 /* Produces a case map from dictionary D, which must have been
1630    previously prepared with start_case_map().
1631
1632    Does not retain any reference to D, and clears the aux members
1633    set up by start_case_map().
1634
1635    Returns the new case map, or a null pointer if no mapping is
1636    required (that is, no data has changed position). */
1637 static struct case_map *
1638 finish_case_map (struct dictionary *d)
1639 {
1640   struct case_map *map;
1641   size_t var_cnt = dict_get_var_cnt (d);
1642   size_t i;
1643   int identity_map;
1644
1645   map = xmalloc (sizeof *map);
1646   map->value_cnt = dict_get_next_value_idx (d);
1647   map->map = xnmalloc (map->value_cnt, sizeof *map->map);
1648   for (i = 0; i < map->value_cnt; i++)
1649     map->map[i] = -1;
1650
1651   identity_map = 1;
1652   for (i = 0; i < var_cnt; i++)
1653     {
1654       struct variable *v = dict_get_var (d, i);
1655       size_t value_cnt = var_get_value_cnt (v);
1656       int *src_fv = (int *) var_detach_aux (v);
1657       size_t idx;
1658
1659       if (v->fv != *src_fv)
1660         identity_map = 0;
1661
1662       for (idx = 0; idx < value_cnt; idx++)
1663         {
1664           int src_idx = *src_fv + idx;
1665           int dst_idx = v->fv + idx;
1666
1667           assert (map->map[dst_idx] == -1);
1668           map->map[dst_idx] = src_idx;
1669         }
1670       free (src_fv);
1671     }
1672
1673   if (identity_map)
1674     {
1675       destroy_case_map (map);
1676       return NULL;
1677     }
1678
1679   while (map->value_cnt > 0 && map->map[map->value_cnt - 1] == -1)
1680     map->value_cnt--;
1681
1682   return map;
1683 }
1684
1685 /* Maps from SRC to DST, applying case map MAP. */
1686 static void
1687 map_case (const struct case_map *map,
1688           const struct ccase *src, struct ccase *dst)
1689 {
1690   size_t dst_idx;
1691
1692   assert (map != NULL);
1693   assert (src != NULL);
1694   assert (dst != NULL);
1695   assert (src != dst);
1696
1697   for (dst_idx = 0; dst_idx < map->value_cnt; dst_idx++)
1698     {
1699       int src_idx = map->map[dst_idx];
1700       if (src_idx != -1)
1701         *case_data_rw (dst, dst_idx) = *case_data (src, src_idx);
1702     }
1703 }
1704
1705 /* Destroys case map MAP. */
1706 static void
1707 destroy_case_map (struct case_map *map)
1708 {
1709   if (map != NULL)
1710     {
1711       free (map->map);
1712       free (map);
1713     }
1714 }