src/language/data-io/combine-files.c

   1 /* PSPP - a program for statistical analysis.
   2    Copyright (C) 1997-9, 2000, 2006, 2007, 2008, 2009 Free Software Foundation, Inc.
   3
   4    This program is free software: you can redistribute it and/or modify
   5    it under the terms of the GNU General Public License as published by
   6    the Free Software Foundation, either version 3 of the License, or
   7    (at your option) any later version.
   8
   9    This program is distributed in the hope that it will be useful,
  10    but WITHOUT ANY WARRANTY; without even the implied warranty of
  11    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12    GNU General Public License for more details.
  13
  14    You should have received a copy of the GNU General Public License
  15    along with this program.  If not, see <http://www.gnu.org/licenses/>. */
  16
  17 #include <config.h>
  18
  19 #include <stdlib.h>
  20
  21 #include <data/any-reader.h>
  22 #include <data/case-matcher.h>
  23 #include <data/case.h>
  24 #include <data/casereader.h>
  25 #include <data/casewriter.h>
  26 #include <data/dictionary.h>
  27 #include <data/format.h>
  28 #include <data/procedure.h>
  29 #include <data/subcase.h>
  30 #include <data/variable.h>
  31 #include <language/command.h>
  32 #include <language/data-io/file-handle.h>
  33 #include <language/data-io/trim.h>
  34 #include <language/lexer/lexer.h>
  35 #include <language/lexer/variable-parser.h>
  36 #include <language/stats/sort-criteria.h>
  37 #include <libpspp/assertion.h>
  38 #include <libpspp/message.h>
  39 #include <libpspp/taint.h>
  40 #include <math/sort.h>
  41
  42 #include "xalloc.h"
  43
  44 #include "gettext.h"
  45 #define _(msgid) gettext (msgid)
  46
  47 enum comb_command_type
  48   {
  49     COMB_ADD,
  50     COMB_MATCH,
  51     COMB_UPDATE
  52   };
  53
  54 /* File types. */
  55 enum comb_file_type
  56   {
  57     COMB_FILE,                  /* Specified on FILE= subcommand. */
  58     COMB_TABLE                  /* Specified on TABLE= subcommand. */
  59   };
  60
  61 /* One FILE or TABLE subcommand. */
  62 struct comb_file
  63   {
  64     /* Basics. */
  65     enum comb_file_type type;   /* COMB_FILE or COMB_TABLE. */
  66
  67     /* Variables. */
  68     struct subcase by_vars;     /* BY variables in this input file. */
  69     struct subcase src, dst;    /* Data to copy to output; where to put it. */
  70
  71     /* Input files. */
  72     struct file_handle *handle; /* Input file handle. */
  73     struct dictionary *dict;    /* Input file dictionary. */
  74     struct casereader *reader;  /* Input data source. */
  75     struct ccase *data;         /* The current input case. */
  76     bool is_minimal;            /* Does 'data' have minimum BY values across
  77                                    all input files? */
  78     bool is_sorted;             /* Is file presorted on the BY variables? */
  79
  80     /* IN subcommand. */
  81     char in_name[VAR_NAME_LEN + 1];
  82     struct variable *in_var;
  83   };
  84
  85 struct comb_proc
  86   {
  87     struct comb_file *files;    /* All the files being merged. */
  88     size_t n_files;             /* Number of files. */
  89
  90     struct dictionary *dict;    /* Dictionary of output file. */
  91     struct subcase by_vars;     /* BY variables in the output. */
  92     struct casewriter *output;  /* Destination for output. */
  93
  94     struct case_matcher *matcher;
  95
  96     /* FIRST, LAST.
  97        Only if "first" or "last" is nonnull are the remaining
  98        members used. */
  99     struct variable *first;     /* Variable specified on FIRST (if any). */
 100     struct variable *last;      /* Variable specified on LAST (if any). */
 101     struct ccase *buffered_case; /* Case ready for output except that we don't
 102                                     know the value for the LAST var yet. */
 103     union value *prev_BY;       /* Values of BY vars in buffered_case. */
 104   };
 105
 106 static int combine_files (enum comb_command_type, struct lexer *,
 107                           struct dataset *);
 108 static void free_comb_proc (struct comb_proc *);
 109
 110 static void close_all_comb_files (struct comb_proc *);
 111 static bool merge_dictionary (struct dictionary *const, struct comb_file *);
 112
 113 static void execute_update (struct comb_proc *);
 114 static void execute_match_files (struct comb_proc *);
 115 static void execute_add_files (struct comb_proc *);
 116
 117 static bool create_flag_var (const char *subcommand_name, const char *var_name,
 118                              struct dictionary *, struct variable **);
 119 static void output_case (struct comb_proc *, struct ccase *, union value *by);
 120 static void output_buffered_case (struct comb_proc *);
 121
 122 int
 123 cmd_add_files (struct lexer *lexer, struct dataset *ds)
 124 {
 125   return combine_files (COMB_ADD, lexer, ds);
 126 }
 127
 128 int
 129 cmd_match_files (struct lexer *lexer, struct dataset *ds)
 130 {
 131   return combine_files (COMB_MATCH, lexer, ds);
 132 }
 133
 134 int
 135 cmd_update (struct lexer *lexer, struct dataset *ds)
 136 {
 137   return combine_files (COMB_UPDATE, lexer, ds);
 138 }
 139
 140 static int
 141 combine_files (enum comb_command_type command,
 142                struct lexer *lexer, struct dataset *ds)
 143 {
 144   struct comb_proc proc;
 145
 146   bool saw_by = false;
 147   bool saw_sort = false;
 148   struct casereader *active_file = NULL;
 149
 150   char first_name[VAR_NAME_LEN + 1] = "";
 151   char last_name[VAR_NAME_LEN + 1] = "";
 152
 153   struct taint *taint = NULL;
 154
 155   size_t n_tables = 0;
 156   size_t allocated_files = 0;
 157
 158   size_t i;
 159
 160   proc.files = NULL;
 161   proc.n_files = 0;
 162   proc.dict = dict_create ();
 163   proc.output = NULL;
 164   proc.matcher = NULL;
 165   subcase_init_empty (&proc.by_vars);
 166   proc.first = NULL;
 167   proc.last = NULL;
 168   proc.buffered_case = NULL;
 169   proc.prev_BY = NULL;
 170
 171   dict_set_case_limit (proc.dict, dict_get_case_limit (dataset_dict (ds)));
 172
 173   lex_match (lexer, '/');
 174   for (;;)
 175     {
 176       struct comb_file *file;
 177       enum comb_file_type type;
 178
 179       if (lex_match_id (lexer, "FILE"))
 180         type = COMB_FILE;
 181       else if (command == COMB_MATCH && lex_match_id (lexer, "TABLE"))
 182         {
 183           type = COMB_TABLE;
 184           n_tables++;
 185         }
 186       else
 187         break;
 188       lex_match (lexer, '=');
 189
 190       if (proc.n_files >= allocated_files)
 191         proc.files = x2nrealloc (proc.files, &allocated_files,
 192                                 sizeof *proc.files);
 193       file = &proc.files[proc.n_files++];
 194       file->type = type;
 195       subcase_init_empty (&file->by_vars);
 196       subcase_init_empty (&file->src);
 197       subcase_init_empty (&file->dst);
 198       file->handle = NULL;
 199       file->dict = NULL;
 200       file->reader = NULL;
 201       file->data = NULL;
 202       file->is_sorted = true;
 203       file->in_name[0] = '\0';
 204       file->in_var = NULL;
 205
 206       if (lex_match (lexer, '*'))
 207         {
 208           if (!proc_has_active_file (ds))
 209             {
 210               msg (SE, _("Cannot specify the active file since no active "
 211                          "file has been defined."));
 212               goto error;
 213             }
 214
 215           if (proc_make_temporary_transformations_permanent (ds))
 216             msg (SE, _("This command may not be used after TEMPORARY when "
 217                        "the active file is an input source.  "
 218                        "Temporary transformations will be made permanent."));
 219
 220           file->dict = dict_clone (dataset_dict (ds));
 221         }
 222       else
 223         {
 224           file->handle = fh_parse (lexer, FH_REF_FILE | FH_REF_SCRATCH);
 225           if (file->handle == NULL)
 226             goto error;
 227
 228           file->reader = any_reader_open (file->handle, &file->dict);
 229           if (file->reader == NULL)
 230             goto error;
 231         }
 232
 233       while (lex_match (lexer, '/'))
 234         if (lex_match_id (lexer, "RENAME"))
 235           {
 236             if (!parse_dict_rename (lexer, file->dict))
 237               goto error;
 238           }
 239         else if (lex_match_id (lexer, "IN"))
 240           {
 241             lex_match (lexer, '=');
 242             if (lex_token (lexer) != T_ID)
 243               {
 244                 lex_error (lexer, NULL);
 245                 goto error;
 246               }
 247
 248             if (file->in_name[0])
 249               {
 250                 msg (SE, _("Multiple IN subcommands for a single FILE or "
 251                            "TABLE."));
 252                 goto error;
 253               }
 254             strcpy (file->in_name, lex_tokid (lexer));
 255             lex_get (lexer);
 256           }
 257         else if (lex_match_id (lexer, "SORT"))
 258           {
 259             file->is_sorted = false;
 260             saw_sort = true;
 261           }
 262
 263       merge_dictionary (proc.dict, file);
 264     }
 265
 266   while (lex_token (lexer) != '.')
 267     {
 268       if (lex_match (lexer, T_BY))
 269         {
 270           const struct variable **by_vars;
 271           size_t i;
 272           bool ok;
 273
 274           if (saw_by)
 275             {
 276               lex_sbc_only_once ("BY");
 277               goto error;
 278             }
 279           saw_by = true;
 280
 281           lex_match (lexer, '=');
 282           if (!parse_sort_criteria (lexer, proc.dict, &proc.by_vars,
 283                                     &by_vars, NULL))
 284             goto error;
 285
 286           ok = true;
 287           for (i = 0; i < proc.n_files; i++)
 288             {
 289               struct comb_file *file = &proc.files[i];
 290               size_t j;
 291
 292               for (j = 0; j < subcase_get_n_values (&proc.by_vars); j++)
 293                 {
 294                   const char *name = var_get_name (by_vars[j]);
 295                   struct variable *var = dict_lookup_var (file->dict, name);
 296                   if (var != NULL)
 297                     subcase_add_var (&file->by_vars, var,
 298                                      subcase_get_direction (&proc.by_vars, j));
 299                   else
 300                     {
 301                       if (file->handle != NULL)
 302                         msg (SE, _("File %s lacks BY variable %s."),
 303                              fh_get_name (file->handle), name);
 304                       else
 305                         msg (SE, _("Active file lacks BY variable %s."), name);
 306                       ok = false;
 307                     }
 308                 }
 309               assert (!ok || subcase_conformable (&file->by_vars,
 310                                                   &proc.files[0].by_vars));
 311             }
 312           free (by_vars);
 313
 314           if (!ok)
 315             goto error;
 316         }
 317       else if (command != COMB_UPDATE && lex_match_id (lexer, "FIRST"))
 318         {
 319           if (first_name[0] != '\0')
 320             {
 321               lex_sbc_only_once ("FIRST");
 322               goto error;
 323             }
 324
 325           lex_match (lexer, '=');
 326           if (!lex_force_id (lexer))
 327             goto error;
 328           strcpy (first_name, lex_tokid (lexer));
 329           lex_get (lexer);
 330         }
 331       else if (command != COMB_UPDATE && lex_match_id (lexer, "LAST"))
 332         {
 333           if (last_name[0] != '\0')
 334             {
 335               lex_sbc_only_once ("LAST");
 336               goto error;
 337             }
 338
 339           lex_match (lexer, '=');
 340           if (!lex_force_id (lexer))
 341             goto error;
 342           strcpy (last_name, lex_tokid (lexer));
 343           lex_get (lexer);
 344         }
 345       else if (lex_match_id (lexer, "MAP"))
 346         {
 347           /* FIXME. */
 348         }
 349       else if (lex_match_id (lexer, "DROP"))
 350         {
 351           if (!parse_dict_drop (lexer, proc.dict))
 352             goto error;
 353         }
 354       else if (lex_match_id (lexer, "KEEP"))
 355         {
 356           if (!parse_dict_keep (lexer, proc.dict))
 357             goto error;
 358         }
 359       else
 360         {
 361           lex_error (lexer, NULL);
 362           goto error;
 363         }
 364
 365       if (!lex_match (lexer, '/') && lex_token (lexer) != '.')
 366         {
 367           lex_end_of_command (lexer);
 368           goto error;
 369         }
 370     }
 371
 372   if (!saw_by)
 373     {
 374       if (command == COMB_UPDATE)
 375         {
 376           msg (SE, _("The BY subcommand is required."));
 377           goto error;
 378         }
 379       if (n_tables)
 380         {
 381           msg (SE, _("BY is required when TABLE is specified."));
 382           goto error;
 383         }
 384       if (saw_sort)
 385         {
 386           msg (SE, _("BY is required when SORT is specified."));
 387           goto error;
 388         }
 389     }
 390
 391   /* Add IN, FIRST, and LAST variables to master dictionary. */
 392   for (i = 0; i < proc.n_files; i++)
 393     {
 394       struct comb_file *file = &proc.files[i];
 395       if (!create_flag_var ("IN", file->in_name, proc.dict, &file->in_var))
 396         goto error;
 397     }
 398   if (!create_flag_var ("FIRST", first_name, proc.dict, &proc.first)
 399       || !create_flag_var ("LAST", last_name, proc.dict, &proc.last))
 400     goto error;
 401
 402   dict_delete_scratch_vars (proc.dict);
 403   dict_compact_values (proc.dict);
 404
 405   /* Set up mapping from each file's variables to master
 406      variables. */
 407   for (i = 0; i < proc.n_files; i++)
 408     {
 409       struct comb_file *file = &proc.files[i];
 410       size_t src_var_cnt = dict_get_var_cnt (file->dict);
 411       size_t j;
 412
 413       for (j = 0; j < src_var_cnt; j++)
 414         {
 415           struct variable *src_var = dict_get_var (file->dict, j);
 416           struct variable *dst_var = dict_lookup_var (proc.dict,
 417                                                       var_get_name (src_var));
 418           if (dst_var != NULL)
 419             {
 420               subcase_add_var (&file->src, src_var, SC_ASCEND);
 421               subcase_add_var (&file->dst, dst_var, SC_ASCEND);
 422             }
 423         }
 424     }
 425
 426   proc.output = autopaging_writer_create (dict_get_next_value_idx (proc.dict));
 427   taint = taint_clone (casewriter_get_taint (proc.output));
 428
 429   /* Set up case matcher. */
 430   proc.matcher = case_matcher_create ();
 431   for (i = 0; i < proc.n_files; i++)
 432     {
 433       struct comb_file *file = &proc.files[i];
 434       if (file->reader == NULL)
 435         {
 436           if (active_file == NULL)
 437             {
 438               proc_discard_output (ds);
 439               file->reader = active_file = proc_open (ds);
 440             }
 441           else
 442             file->reader = casereader_clone (active_file);
 443         }
 444       if (!file->is_sorted)
 445         file->reader = sort_execute (file->reader, &file->by_vars);
 446       taint_propagate (casereader_get_taint (file->reader), taint);
 447       file->data = casereader_read (file->reader);
 448       if (file->type == COMB_FILE)
 449         case_matcher_add_input (proc.matcher, &file->by_vars,
 450                                 &file->data, &file->is_minimal);
 451     }
 452
 453   if (command == COMB_ADD)
 454     execute_add_files (&proc);
 455   else if (command == COMB_MATCH)
 456     execute_match_files (&proc);
 457   else if (command == COMB_UPDATE)
 458     execute_update (&proc);
 459   else
 460     NOT_REACHED ();
 461
 462   case_matcher_destroy (proc.matcher);
 463   proc.matcher = NULL;
 464   close_all_comb_files (&proc);
 465   if (active_file != NULL)
 466     proc_commit (ds);
 467
 468   proc_set_active_file (ds, casewriter_make_reader (proc.output), proc.dict);
 469   proc.dict = NULL;
 470   proc.output = NULL;
 471
 472   free_comb_proc (&proc);
 473
 474   return taint_destroy (taint) ? CMD_SUCCESS : CMD_CASCADING_FAILURE;
 475
 476  error:
 477   if (active_file != NULL)
 478     proc_commit (ds);
 479   free_comb_proc (&proc);
 480   taint_destroy (taint);
 481   return CMD_CASCADING_FAILURE;
 482 }
 483
 484 /* Merge the dictionary for file F into master dictionary M. */
 485 static bool
 486 merge_dictionary (struct dictionary *const m, struct comb_file *f)
 487 {
 488   struct dictionary *d = f->dict;
 489   const char *d_docs, *m_docs;
 490   int i;
 491   const char *file_encoding;
 492
 493   if (dict_get_label (m) == NULL)
 494     dict_set_label (m, dict_get_label (d));
 495
 496   d_docs = dict_get_documents (d);
 497   m_docs = dict_get_documents (m);
 498
 499
 500   /* FIXME: If the input files have different encodings, then
 501      the result is undefined.
 502      The correct thing to do would be to convert to an encoding
 503      which can cope with all the input files (eg UTF-8).
 504    */
 505   file_encoding = dict_get_encoding (f->dict);
 506   if ( file_encoding != NULL)
 507     {
 508       if ( dict_get_encoding (m) == NULL)
 509         dict_set_encoding (m, file_encoding);
 510       else if ( 0 != strcmp (file_encoding, dict_get_encoding (m)))
 511         {
 512           msg (MW,
 513                _("Combining files with incompatible encodings. String data may not be represented correctly."));
 514         }
 515     }
 516
 517   if (d_docs != NULL)
 518     {
 519       if (m_docs == NULL)
 520         dict_set_documents (m, d_docs);
 521       else
 522         {
 523           char *new_docs = xasprintf ("%s%s", m_docs, d_docs);
 524           dict_set_documents (m, new_docs);
 525           free (new_docs);
 526         }
 527     }
 528
 529   for (i = 0; i < dict_get_var_cnt (d); i++)
 530     {
 531       struct variable *dv = dict_get_var (d, i);
 532       struct variable *mv = dict_lookup_var (m, var_get_name (dv));
 533
 534       if (dict_class_from_id (var_get_name (dv)) == DC_SCRATCH)
 535         continue;
 536
 537       if (mv != NULL)
 538         {
 539           if (var_get_width (mv) != var_get_width (dv))
 540             {
 541               const char *var_name = var_get_name (dv);
 542               const char *file_name = fh_get_name (f->handle);
 543               struct string s = DS_EMPTY_INITIALIZER;
 544               ds_put_format (&s,
 545                              _("Variable %s in file %s has different "
 546                                "type or width from the same variable in "
 547                                "earlier file."),
 548                              var_name, file_name);
 549               ds_put_cstr (&s, "  ");
 550               if (var_is_numeric (dv))
 551                 ds_put_format (&s, _("In file %s, %s is numeric."),
 552                                file_name, var_name);
 553               else
 554                 ds_put_format (&s, _("In file %s, %s is a string variable "
 555                                      "with width %d."),
 556                                file_name, var_name, var_get_width (dv));
 557               ds_put_cstr (&s, "  ");
 558               if (var_is_numeric (mv))
 559                 ds_put_format (&s, _("In an earlier file, %s was numeric."),
 560                                var_name);
 561               else
 562                 ds_put_format (&s, _("In an earlier file, %s was a string "
 563                                      "variable with width %d."),
 564                                var_name, var_get_width (mv));
 565               msg (SE, ds_cstr (&s));
 566               ds_destroy (&s);
 567               return false;
 568             }
 569
 570           if (var_has_value_labels (dv) && !var_has_value_labels (mv))
 571             var_set_value_labels (mv, var_get_value_labels (dv));
 572           if (var_has_missing_values (dv) && !var_has_missing_values (mv))
 573             var_set_missing_values (mv, var_get_missing_values (dv));
 574           if (var_get_label (dv) && !var_get_label (mv))
 575             var_set_label (mv, var_get_label (dv));
 576         }
 577       else
 578         mv = dict_clone_var_assert (m, dv, var_get_name (dv));
 579     }
 580
 581   return true;
 582 }
 583
 584 /* If VAR_NAME is a non-empty string, attempts to create a
 585    variable named VAR_NAME, with format F1.0, in DICT, and stores
 586    a pointer to the variable in *VAR.  Returns true if
 587    successful, false if the variable name is a duplicate (in
 588    which case a message saying that the variable specified on the
 589    given SUBCOMMAND is a duplicate is emitted).  Also returns
 590    true, without doing anything, if VAR_NAME is null or empty. */
 591 static bool
 592 create_flag_var (const char *subcommand, const char *var_name,
 593                  struct dictionary *dict, struct variable **var)
 594 {
 595   if (var_name[0] != '\0')
 596     {
 597       struct fmt_spec format = fmt_for_output (FMT_F, 1, 0);
 598       *var = dict_create_var (dict, var_name, 0);
 599       if (*var == NULL)
 600         {
 601           msg (SE, _("Variable name %s specified on %s subcommand "
 602                      "duplicates an existing variable name."),
 603                subcommand, var_name);
 604           return false;
 605         }
 606       var_set_both_formats (*var, &format);
 607     }
 608   else
 609     *var = NULL;
 610   return true;
 611 }
 612
 613 /* Closes all the files in PROC and frees their associated data. */
 614 static void
 615 close_all_comb_files (struct comb_proc *proc)
 616 {
 617   size_t i;
 618
 619   for (i = 0; i < proc->n_files; i++)
 620     {
 621       struct comb_file *file = &proc->files[i];
 622       subcase_destroy (&file->by_vars);
 623       subcase_destroy (&file->src);
 624       subcase_destroy (&file->dst);
 625       fh_unref (file->handle);
 626       dict_destroy (file->dict);
 627       casereader_destroy (file->reader);
 628       case_unref (file->data);
 629     }
 630   free (proc->files);
 631   proc->files = NULL;
 632   proc->n_files = 0;
 633 }
 634
 635 /* Frees all the data for the procedure. */
 636 static void
 637 free_comb_proc (struct comb_proc *proc)
 638 {
 639   close_all_comb_files (proc);
 640   dict_destroy (proc->dict);
 641   casewriter_destroy (proc->output);
 642   case_matcher_destroy (proc->matcher);
 643   subcase_destroy (&proc->by_vars);
 644   case_unref (proc->buffered_case);
 645   free (proc->prev_BY);
 646 }
 647 \f
 648 static bool scan_table (struct comb_file *, union value by[]);
 649 static struct ccase *create_output_case (const struct comb_proc *);
 650 static void apply_case (const struct comb_file *, struct ccase *);
 651 static void apply_file_case_and_advance (struct comb_file *, struct ccase *,
 652                                          union value by[]);
 653 static void output_case (struct comb_proc *, struct ccase *, union value by[]);
 654 static void output_buffered_case (struct comb_proc *);
 655
 656 /* Executes the ADD FILES command. */
 657 static void
 658 execute_add_files (struct comb_proc *proc)
 659 {
 660   union value *by;
 661
 662   while (case_matcher_match (proc->matcher, &by))
 663     {
 664       size_t i;
 665
 666       for (i = 0; i < proc->n_files; i++)
 667         {
 668           struct comb_file *file = &proc->files[i];
 669           while (file->is_minimal)
 670             {
 671               struct ccase *output = create_output_case (proc);
 672               apply_file_case_and_advance (file, output, by);
 673               output_case (proc, output, by);
 674             }
 675         }
 676     }
 677   output_buffered_case (proc);
 678 }
 679
 680 /* Executes the MATCH FILES command. */
 681 static void
 682 execute_match_files (struct comb_proc *proc)
 683 {
 684   union value *by;
 685
 686   while (case_matcher_match (proc->matcher, &by))
 687     {
 688       struct ccase *output;
 689       size_t i;
 690
 691       output = create_output_case (proc);
 692       for (i = proc->n_files; i-- > 0; )
 693         {
 694           struct comb_file *file = &proc->files[i];
 695           if (file->type == COMB_FILE)
 696             {
 697               if (file->is_minimal)
 698                 apply_file_case_and_advance (file, output, NULL);
 699             }
 700           else
 701             {
 702               if (scan_table (file, by))
 703                 apply_case (file, output);
 704             }
 705         }
 706       output_case (proc, output, by);
 707     }
 708   output_buffered_case (proc);
 709 }
 710
 711 /* Executes the UPDATE command. */
 712 static void
 713 execute_update (struct comb_proc *proc)
 714 {
 715   union value *by;
 716   size_t n_duplicates = 0;
 717
 718   while (case_matcher_match (proc->matcher, &by))
 719     {
 720       struct comb_file *first, *file;
 721       struct ccase *output;
 722
 723       /* Find first nonnull case in array and make an output case
 724          from it. */
 725       output = create_output_case (proc);
 726       for (first = &proc->files[0]; ; first++)
 727         if (first->is_minimal)
 728           break;
 729       apply_file_case_and_advance (first, output, by);
 730
 731       /* Read additional cases and update the output case from
 732          them.  (Don't update the output case from any duplicate
 733          cases in the master file.) */
 734       for (file = first + (first == proc->files);
 735            file < &proc->files[proc->n_files]; file++)
 736         {
 737           while (file->is_minimal)
 738             apply_file_case_and_advance (file, output, by);
 739         }
 740       casewriter_write (proc->output, output);
 741
 742       /* Write duplicate cases in the master file directly to the
 743          output.  */
 744       if (first == proc->files && first->is_minimal)
 745         {
 746           n_duplicates++;
 747           while (first->is_minimal)
 748             {
 749               output = create_output_case (proc);
 750               apply_file_case_and_advance (first, output, by);
 751               casewriter_write (proc->output, output);
 752             }
 753         }
 754     }
 755
 756   if (n_duplicates)
 757     msg (SW, _("Encountered %zu sets of duplicate cases in the master file."),
 758          n_duplicates);
 759 }
 760
 761 /* Reads FILE, which must be of type COMB_TABLE, until it
 762    encounters a case with BY or greater for its BY variables.
 763    Returns true if a case with exactly BY for its BY variables
 764    was found, otherwise false. */
 765 static bool
 766 scan_table (struct comb_file *file, union value by[])
 767 {
 768   while (file->data != NULL)
 769     {
 770       int cmp = subcase_compare_3way_xc (&file->by_vars, by, file->data);
 771       if (cmp > 0)
 772         {
 773           case_unref (file->data);
 774           file->data = casereader_read (file->reader);
 775         }
 776       else
 777         return cmp == 0;
 778     }
 779   return false;
 780 }
 781
 782 /* Creates and returns an output case for PROC, initializing each
 783    of its values to system-missing or blanks, except that the
 784    values of IN variables are set to 0. */
 785 static struct ccase *
 786 create_output_case (const struct comb_proc *proc)
 787 {
 788   size_t n_vars = dict_get_var_cnt (proc->dict);
 789   struct ccase *output;
 790   size_t i;
 791
 792   output = case_create (dict_get_next_value_idx (proc->dict));
 793   for (i = 0; i < n_vars; i++)
 794     {
 795       struct variable *v = dict_get_var (proc->dict, i);
 796       value_set_missing (case_data_rw (output, v), var_get_width (v));
 797     }
 798   for (i = 0; i < proc->n_files; i++)
 799     {
 800       struct comb_file *file = &proc->files[i];
 801       if (file->in_var != NULL)
 802         case_data_rw (output, file->in_var)->f = false;
 803     }
 804   return output;
 805 }
 806
 807 /* Copies the data from FILE's case into output case OUTPUT.
 808    If FILE has an IN variable, then it is set to 1 in OUTPUT. */
 809 static void
 810 apply_case (const struct comb_file *file, struct ccase *output)
 811 {
 812   subcase_copy (&file->src, file->data, &file->dst, output);
 813   if (file->in_var != NULL)
 814     case_data_rw (output, file->in_var)->f = true;
 815 }
 816
 817 /* Like apply_case() above, but also advances FILE to its next
 818    case.  Also, if BY is nonnull, then FILE's is_minimal member
 819    is updated based on whether the new case's BY values still
 820    match those in BY. */
 821 static void
 822 apply_file_case_and_advance (struct comb_file *file, struct ccase *output,
 823                              union value by[])
 824 {
 825   apply_case (file, output);
 826   case_unref (file->data);
 827   file->data = casereader_read (file->reader);
 828   if (by)
 829     file->is_minimal = (file->data != NULL
 830                         && subcase_equal_cx (&file->by_vars, file->data, by));
 831 }
 832
 833 /* Writes OUTPUT, whose BY values has been extracted into BY, to
 834    PROC's output file, first initializing any FIRST or LAST
 835    variables in OUTPUT to the correct values. */
 836 static void
 837 output_case (struct comb_proc *proc, struct ccase *output, union value by[])
 838 {
 839   if (proc->first == NULL && proc->last == NULL)
 840     casewriter_write (proc->output, output);
 841   else
 842     {
 843       /* It's harder with LAST, because we can't know whether
 844          this case is the last in a group until we've prepared
 845          the *next* case also.  Thus, we buffer the previous
 846          output case until the next one is ready. */
 847       bool new_BY;
 848       if (proc->prev_BY != NULL)
 849         {
 850           new_BY = !subcase_equal_xx (&proc->by_vars, proc->prev_BY, by);
 851           if (proc->last != NULL)
 852             case_data_rw (proc->buffered_case, proc->last)->f = new_BY;
 853           casewriter_write (proc->output, proc->buffered_case);
 854         }
 855       else
 856         new_BY = true;
 857
 858       proc->buffered_case = output;
 859       if (proc->first != NULL)
 860         case_data_rw (proc->buffered_case, proc->first)->f = new_BY;
 861
 862       if (new_BY)
 863         {
 864           size_t n = (subcase_get_n_values (&proc->by_vars)
 865                       * sizeof (union value));
 866           if (proc->prev_BY == NULL)
 867             proc->prev_BY = xmalloc (n);
 868           memcpy (proc->prev_BY, by, n);
 869         }
 870     }
 871 }
 872
 873 /* Writes a trailing buffered case to the output, if FIRST or
 874    LAST is in use. */
 875 static void
 876 output_buffered_case (struct comb_proc *proc)
 877 {
 878   if (proc->prev_BY != NULL)
 879     {
 880       if (proc->last != NULL)
 881         case_data_rw (proc->buffered_case, proc->last)->f = 1.0;
 882       casewriter_write (proc->output, proc->buffered_case);
 883       proc->buffered_case = NULL;
 884     }
 885 }