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