pintos-os.org Git - pspp/blob - src/language/commands/data-parser.c

   1 /* PSPP - a program for statistical analysis.
   2    Copyright (C) 2007, 2009, 2010, 2011, 2012, 2013, 2016 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 "language/commands/data-parser.h"
  20
  21 #include <stdint.h>
  22 #include <stdlib.h>
  23
  24 #include "data/casereader-provider.h"
  25 #include "data/data-in.h"
  26 #include "data/dataset.h"
  27 #include "data/dictionary.h"
  28 #include "data/format.h"
  29 #include "data/file-handle-def.h"
  30 #include "data/settings.h"
  31 #include "language/commands/data-reader.h"
  32 #include "libpspp/intern.h"
  33 #include "libpspp/message.h"
  34 #include "libpspp/str.h"
  35 #include "libpspp/string-array.h"
  36 #include "output/pivot-table.h"
  37
  38 #include "gl/xalloc.h"
  39
  40 #include "gettext.h"
  41 #define N_(msgid) msgid
  42 #define _(msgid) gettext (msgid)
  43
  44 /* Data parser for textual data like that read by DATA LIST. */
  45 struct data_parser
  46   {
  47     enum data_parser_type type; /* Type of data to parse. */
  48     int skip_records;           /* Records to skip before first real data. */
  49
  50     struct field *fields;       /* Fields to parse. */
  51     size_t n_fields;            /* Number of fields. */
  52     size_t field_allocated;     /* Number of fields spaced allocated for. */
  53
  54     /* DP_DELIMITED parsers only. */
  55     bool span;                  /* May cases span multiple records? */
  56     bool empty_line_has_field;  /* Does an empty line have an (empty) field? */
  57     bool warn_missing_fields;   /* Should missing fields be considered errors? */
  58     struct substring quotes;    /* Characters that can quote separators. */
  59     bool quote_escape;          /* Doubled quote acts as escape? */
  60     struct substring soft_seps; /* Two soft separators act like just one. */
  61     struct substring hard_seps; /* Two hard separators yield empty fields. */
  62     struct string any_sep;      /* Concatenation of soft_seps and hard_seps. */
  63
  64     /* DP_FIXED parsers only. */
  65     int records_per_case;       /* Number of records in each case. */
  66   };
  67
  68 /* How to parse one variable. */
  69 struct field
  70   {
  71     struct fmt_spec format;     /* Input format of this field. */
  72     int case_idx;               /* First value in case. */
  73     char *name;                 /* Var name for error messages and tables. */
  74
  75     /* DP_FIXED only. */
  76     int record;                 /* Record number (1-based). */
  77     int first_column;           /* First column in record (1-based). */
  78   };
  79
  80 static void set_any_sep (struct data_parser *parser);
  81
  82 /* Creates and returns a new data parser. */
  83 struct data_parser *
  84 data_parser_create (void)
  85 {
  86   struct data_parser *parser = xmalloc (sizeof *parser);
  87   *parser = (struct data_parser) {
  88     .type = DP_FIXED,
  89     .span = true,
  90     .warn_missing_fields = true,
  91     .quotes = ss_clone (ss_cstr ("\"'")),
  92     .soft_seps = ss_clone (ss_cstr (CC_SPACES)),
  93     .hard_seps = ss_clone (ss_cstr (",")),
  94   };
  95   set_any_sep (parser);
  96
  97   return parser;
  98 }
  99
 100 /* Destroys PARSER. */
 101 void
 102 data_parser_destroy (struct data_parser *parser)
 103 {
 104   if (parser != NULL)
 105     {
 106       size_t i;
 107
 108       for (i = 0; i < parser->n_fields; i++)
 109         free (parser->fields[i].name);
 110       free (parser->fields);
 111       ss_dealloc (&parser->quotes);
 112       ss_dealloc (&parser->soft_seps);
 113       ss_dealloc (&parser->hard_seps);
 114       ds_destroy (&parser->any_sep);
 115       free (parser);
 116     }
 117 }
 118
 119 /* Returns the type of PARSER (either DP_DELIMITED or DP_FIXED). */
 120 enum data_parser_type
 121 data_parser_get_type (const struct data_parser *parser)
 122 {
 123   return parser->type;
 124 }
 125
 126 /* Sets the type of PARSER to TYPE (either DP_DELIMITED or
 127    DP_FIXED). */
 128 void
 129 data_parser_set_type (struct data_parser *parser, enum data_parser_type type)
 130 {
 131   assert (parser->n_fields == 0);
 132   assert (type == DP_FIXED || type == DP_DELIMITED);
 133   parser->type = type;
 134 }
 135
 136 /* Configures PARSER to skip the specified number of
 137    INITIAL_RECORDS_TO_SKIP before parsing any data.  By default,
 138    no records are skipped. */
 139 void
 140 data_parser_set_skip (struct data_parser *parser, int initial_records_to_skip)
 141 {
 142   assert (initial_records_to_skip >= 0);
 143   parser->skip_records = initial_records_to_skip;
 144 }
 145
 146 /* Returns true if PARSER is configured to allow cases to span
 147    multiple records. */
 148 bool
 149 data_parser_get_span (const struct data_parser *parser)
 150 {
 151   return parser->span;
 152 }
 153
 154 /* If MAY_CASES_SPAN_RECORDS is true, configures PARSER to allow
 155    a single case to span multiple records and multiple cases to
 156    occupy a single record.  If MAY_CASES_SPAN_RECORDS is false,
 157    configures PARSER to require each record to contain exactly
 158    one case.
 159
 160    This setting affects parsing of DP_DELIMITED files only. */
 161 void
 162 data_parser_set_span (struct data_parser *parser, bool may_cases_span_records)
 163 {
 164   parser->span = may_cases_span_records;
 165 }
 166
 167 /* If EMPTY_LINE_HAS_FIELD is true, configures PARSER to parse an
 168    empty line as an empty field and to treat a hard delimiter
 169    followed by end-of-line as an empty field.  If
 170    EMPTY_LINE_HAS_FIELD is false, PARSER will skip empty lines
 171    and hard delimiters at the end of lines without emitting empty
 172    fields.
 173
 174    This setting affects parsing of DP_DELIMITED files only. */
 175 void
 176 data_parser_set_empty_line_has_field (struct data_parser *parser,
 177                                       bool empty_line_has_field)
 178 {
 179   parser->empty_line_has_field = empty_line_has_field;
 180 }
 181
 182
 183 /* If WARN_MISSING_FIELDS is true, configures PARSER to emit a warning
 184    and cause an error condition when a missing field is encountered.
 185    If  WARN_MISSING_FIELDS is false, PARSER will silently fill such
 186    fields with the system missing value.
 187
 188    This setting affects parsing of DP_DELIMITED files only. */
 189 void
 190 data_parser_set_warn_missing_fields (struct data_parser *parser,
 191                                      bool warn_missing_fields)
 192 {
 193   parser->warn_missing_fields = warn_missing_fields;
 194 }
 195
 196
 197 /* Sets the characters that may be used for quoting field
 198    contents to QUOTES.  If QUOTES is empty, quoting will be
 199    disabled.
 200
 201    The caller retains ownership of QUOTES.
 202
 203    This setting affects parsing of DP_DELIMITED files only. */
 204 void
 205 data_parser_set_quotes (struct data_parser *parser, struct substring quotes)
 206 {
 207   ss_dealloc (&parser->quotes);
 208   parser->quotes = ss_clone (quotes);
 209 }
 210
 211 /* If ESCAPE is false (the default setting), a character used for
 212    quoting cannot itself be embedded within a quoted field.  If
 213    ESCAPE is true, then a quote character can be embedded within
 214    a quoted field by doubling it.
 215
 216    This setting affects parsing of DP_DELIMITED files only, and
 217    only when at least one quote character has been set (with
 218    data_parser_set_quotes). */
 219 void
 220 data_parser_set_quote_escape (struct data_parser *parser, bool escape)
 221 {
 222   parser->quote_escape = escape;
 223 }
 224
 225 /* Sets PARSER's soft delimiters to DELIMITERS.  Soft delimiters
 226    separate fields, but consecutive soft delimiters do not yield
 227    empty fields.  (Ordinarily, only white space characters are
 228    appropriate soft delimiters.)
 229
 230    The caller retains ownership of DELIMITERS.
 231
 232    This setting affects parsing of DP_DELIMITED files only. */
 233 void
 234 data_parser_set_soft_delimiters (struct data_parser *parser,
 235                                  struct substring delimiters)
 236 {
 237   ss_dealloc (&parser->soft_seps);
 238   parser->soft_seps = ss_clone (delimiters);
 239   set_any_sep (parser);
 240 }
 241
 242 /* Sets PARSER's hard delimiters to DELIMITERS.  Hard delimiters
 243    separate fields.  A consecutive pair of hard delimiters yield
 244    an empty field.
 245
 246    The caller retains ownership of DELIMITERS.
 247
 248    This setting affects parsing of DP_DELIMITED files only. */
 249 void
 250 data_parser_set_hard_delimiters (struct data_parser *parser,
 251                                  struct substring delimiters)
 252 {
 253   ss_dealloc (&parser->hard_seps);
 254   parser->hard_seps = ss_clone (delimiters);
 255   set_any_sep (parser);
 256 }
 257
 258 /* Returns the number of records per case. */
 259 int
 260 data_parser_get_records (const struct data_parser *parser)
 261 {
 262   return parser->records_per_case;
 263 }
 264
 265 /* Sets the number of records per case to RECORDS_PER_CASE.
 266
 267    This setting affects parsing of DP_FIXED files only. */
 268 void
 269 data_parser_set_records (struct data_parser *parser, int records_per_case)
 270 {
 271   assert (records_per_case >= 0);
 272   assert (records_per_case >= parser->records_per_case);
 273   parser->records_per_case = records_per_case;
 274 }
 275
 276 static void
 277 add_field (struct data_parser *p, struct fmt_spec format, int case_idx,
 278            const char *name, int record, int first_column)
 279 {
 280   struct field *field;
 281
 282   if (p->n_fields == p->field_allocated)
 283     p->fields = x2nrealloc (p->fields, &p->field_allocated, sizeof *p->fields);
 284   field = &p->fields[p->n_fields++];
 285   *field = (struct field) {
 286     .format = format,
 287     .case_idx = case_idx,
 288     .name = xstrdup (name),
 289     .record = record,
 290     .first_column = first_column,
 291   };
 292 }
 293
 294 /* Adds a delimited field to the field parsed by PARSER, which
 295    must be configured as a DP_DELIMITED parser.  The field is
 296    parsed as input format FORMAT.  Its data will be stored into case
 297    index CASE_INDEX.  Errors in input data will be reported
 298    against variable NAME. */
 299 void
 300 data_parser_add_delimited_field (struct data_parser *parser,
 301                                  struct fmt_spec format, int case_idx,
 302                                  const char *name)
 303 {
 304   assert (parser->type == DP_DELIMITED);
 305   add_field (parser, format, case_idx, name, 0, 0);
 306 }
 307
 308 /* Adds a fixed field to the field parsed by PARSER, which
 309    must be configured as a DP_FIXED parser.  The field is
 310    parsed as input format FORMAT.  Its data will be stored into case
 311    index CASE_INDEX.  Errors in input data will be reported
 312    against variable NAME.  The field will be drawn from the
 313    FORMAT->w columns in 1-based RECORD starting at 1-based
 314    column FIRST_COLUMN.
 315
 316    RECORD must be at least as great as that of any field already
 317    added; that is, fields must be added in increasing order of
 318    record number.  If RECORD is greater than the current number
 319    of records per case, the number of records per case are
 320    increased as needed.  */
 321 void
 322 data_parser_add_fixed_field (struct data_parser *parser,
 323                              struct fmt_spec format, int case_idx,
 324                              const char *name,
 325                              int record, int first_column)
 326 {
 327   assert (parser->type == DP_FIXED);
 328   assert (parser->n_fields == 0
 329           || record >= parser->fields[parser->n_fields - 1].record);
 330   if (record > parser->records_per_case)
 331     parser->records_per_case = record;
 332   add_field (parser, format, case_idx, name, record, first_column);
 333 }
 334
 335 /* Returns true if any fields have been added to PARSER, false
 336    otherwise. */
 337 bool
 338 data_parser_any_fields (const struct data_parser *parser)
 339 {
 340   return parser->n_fields > 0;
 341 }
 342
 343 static void
 344 set_any_sep (struct data_parser *parser)
 345 {
 346   ds_assign_substring (&parser->any_sep, parser->soft_seps);
 347   ds_put_substring (&parser->any_sep, parser->hard_seps);
 348 }
 349 \f
 350 static bool parse_delimited_span (const struct data_parser *,
 351                                   struct dfm_reader *,
 352                                   struct dictionary *, struct ccase *);
 353 static bool parse_delimited_no_span (const struct data_parser *,
 354                                      struct dfm_reader *,
 355                                      struct dictionary *, struct ccase *);
 356 static bool parse_fixed (const struct data_parser *, struct dfm_reader *,
 357                          struct dictionary *, struct ccase *);
 358
 359 /* Reads a case from DFM into C, which matches dictionary DICT, parsing it with
 360    PARSER.  Returns true if successful, false at end of file or on I/O error.
 361
 362    Case C must not be shared. */
 363 bool
 364 data_parser_parse (struct data_parser *parser, struct dfm_reader *reader,
 365                    struct dictionary *dict, struct ccase *c)
 366 {
 367   bool retval;
 368
 369   assert (!case_is_shared (c));
 370   assert (data_parser_any_fields (parser));
 371
 372   /* Skip the requested number of records before reading the
 373      first case. */
 374   for (; parser->skip_records > 0; parser->skip_records--)
 375     {
 376       if (dfm_eof (reader))
 377         return false;
 378       dfm_forward_record (reader);
 379     }
 380
 381   /* Limit cases. */
 382   if (parser->type == DP_DELIMITED)
 383     {
 384       if (parser->span)
 385         retval = parse_delimited_span (parser, reader, dict, c);
 386       else
 387         retval = parse_delimited_no_span (parser, reader, dict, c);
 388     }
 389   else
 390     retval = parse_fixed (parser, reader, dict, c);
 391
 392   return retval;
 393 }
 394
 395 static void
 396 cut_field__ (const struct data_parser *parser, const struct substring *line,
 397              struct substring *p, size_t *n_columns,
 398              struct string *tmp, struct substring *field)
 399 {
 400   bool quoted = ss_find_byte (parser->quotes, ss_first (*p)) != SIZE_MAX;
 401   if (quoted)
 402     {
 403       /* Quoted field. */
 404       int quote = ss_get_byte (p);
 405       if (!ss_get_until (p, quote, field))
 406         msg (DW, _("Quoted string extends beyond end of line."));
 407       if (parser->quote_escape && ss_first (*p) == quote)
 408         {
 409           ds_assign_substring (tmp, *field);
 410           while (ss_match_byte (p, quote))
 411             {
 412               struct substring ss;
 413               ds_put_byte (tmp, quote);
 414               if (!ss_get_until (p, quote, &ss))
 415                 msg (DW, _("Quoted string extends beyond end of line."));
 416               ds_put_substring (tmp, ss);
 417             }
 418           *field = ds_ss (tmp);
 419         }
 420       *n_columns = ss_length (*line) - ss_length (*p);
 421     }
 422   else
 423     {
 424       /* Regular field. */
 425       ss_get_bytes (p, ss_cspan (*p, ds_ss (&parser->any_sep)), field);
 426       *n_columns = ss_length (*field);
 427     }
 428
 429   /* Skip trailing soft separator and a single hard separator if present. */
 430   size_t length_before_separators = ss_length (*p);
 431   ss_ltrim (p, parser->soft_seps);
 432   if (!ss_is_empty (*p)
 433       && ss_find_byte (parser->hard_seps, ss_first (*p)) != SIZE_MAX)
 434     {
 435       ss_advance (p, 1);
 436       ss_ltrim (p, parser->soft_seps);
 437     }
 438
 439   if (!ss_is_empty (*p) && quoted && length_before_separators == ss_length (*p))
 440     msg (DW, _("Missing delimiter following quoted string."));
 441 }
 442
 443 /* Extracts a delimited field from the current position in the
 444    current record according to PARSER, reading data from READER.
 445
 446    *FIELD is set to the field content.  The caller must not or
 447    destroy this constant string.
 448
 449    Sets *FIRST_COLUMN to the 1-based column number of the start of
 450    the extracted field, and *LAST_COLUMN to the end of the extracted
 451    field.
 452
 453    Returns true on success, false on failure. */
 454 static bool
 455 cut_field (const struct data_parser *parser, struct dfm_reader *reader,
 456            int *first_column, int *last_column, struct string *tmp,
 457            struct substring *field)
 458 {
 459   struct substring line, p;
 460
 461   if (dfm_eof (reader))
 462     return false;
 463   if (ss_is_empty (parser->hard_seps))
 464     dfm_expand_tabs (reader);
 465   line = p = dfm_get_record (reader);
 466
 467   /* Skip leading soft separators. */
 468   ss_ltrim (&p, parser->soft_seps);
 469
 470   /* Handle empty or completely consumed lines. */
 471   if (ss_is_empty (p))
 472     {
 473       if (!parser->empty_line_has_field || dfm_columns_past_end (reader) > 0)
 474         return false;
 475       else
 476         {
 477           *field = p;
 478           *first_column = dfm_column_start (reader);
 479           *last_column = *first_column + 1;
 480           dfm_forward_columns (reader, 1);
 481           return true;
 482         }
 483     }
 484
 485   size_t n_columns;
 486   cut_field__ (parser, &line, &p, &n_columns, tmp, field);
 487   *first_column = dfm_column_start (reader);
 488   *last_column = *first_column + n_columns;
 489
 490   if (ss_is_empty (p))
 491     dfm_forward_columns (reader, 1);
 492   dfm_forward_columns (reader, ss_length (line) - ss_length (p));
 493
 494   return true;
 495 }
 496
 497 static void
 498 parse_error (const struct dfm_reader *reader, const struct field *field,
 499              int first_column, int last_column, char *error)
 500 {
 501   int line_number = dfm_get_line_number (reader);
 502   struct msg_location *location = xmalloc (sizeof *location);
 503   *location = (struct msg_location) {
 504     .file_name = intern_new (dfm_get_file_name (reader)),
 505     .start = { .line = line_number, .column = first_column },
 506     .end = { .line = line_number, .column = last_column - 1 },
 507   };
 508   struct msg *m = xmalloc (sizeof *m);
 509   *m = (struct msg) {
 510     .category = MSG_C_DATA,
 511     .severity = MSG_S_WARNING,
 512     .location = location,
 513     .text = xasprintf (_("Data for variable %s is not valid as format %s: %s"),
 514                        field->name, fmt_name (field->format.type), error),
 515   };
 516   msg_emit (m);
 517
 518   free (error);
 519 }
 520
 521 /* Reads a case from READER into C, which matches DICT, parsing it according to
 522    fixed-format syntax rules in PARSER.  Returns true if successful, false at
 523    end of file or on I/O error. */
 524 static bool
 525 parse_fixed (const struct data_parser *parser, struct dfm_reader *reader,
 526              struct dictionary *dict, struct ccase *c)
 527 {
 528   const char *input_encoding = dfm_reader_get_encoding (reader);
 529   const char *output_encoding = dict_get_encoding (dict);
 530   struct field *f;
 531   int row;
 532
 533   if (dfm_eof (reader))
 534     return false;
 535
 536   f = parser->fields;
 537   for (row = 1; row <= parser->records_per_case; row++)
 538     {
 539       struct substring line;
 540
 541       if (dfm_eof (reader))
 542         {
 543           msg (DW, _("Partial case of %d of %d records discarded."),
 544                row - 1, parser->records_per_case);
 545           return false;
 546         }
 547       dfm_expand_tabs (reader);
 548       line = dfm_get_record (reader);
 549
 550       for (; f < &parser->fields[parser->n_fields] && f->record == row; f++)
 551         {
 552           struct substring s = ss_substr (line, f->first_column - 1,
 553                                           f->format.w);
 554           union value *value = case_data_rw_idx (c, f->case_idx);
 555           char *error = data_in (s, input_encoding, f->format.type,
 556                                  settings_get_fmt_settings (),
 557                                  value, fmt_var_width (f->format),
 558                                  output_encoding);
 559
 560           if (error == NULL)
 561             data_in_imply_decimals (s, input_encoding, f->format.type,
 562                                     f->format.d, settings_get_fmt_settings (),
 563                                     value);
 564           else
 565             parse_error (reader, f, f->first_column,
 566                          f->first_column + f->format.w, error);
 567         }
 568
 569       dfm_forward_record (reader);
 570     }
 571
 572   return true;
 573 }
 574
 575 /* Splits the data line in LINE into individual text fields and returns the
 576    number of fields.  If SA is nonnull, appends each field to SA; the caller
 577    retains ownership of SA and its contents.  */
 578 size_t
 579 data_parser_split (const struct data_parser *parser,
 580                    struct substring line, struct string_array *sa)
 581 {
 582   size_t n = 0;
 583
 584   struct string tmp = DS_EMPTY_INITIALIZER;
 585   for (;;)
 586     {
 587       struct substring p = line;
 588       ss_ltrim (&p, parser->soft_seps);
 589       if (ss_is_empty (p))
 590         {
 591           ds_destroy (&tmp);
 592           return n;
 593         }
 594
 595       size_t n_columns;
 596       struct substring field;
 597
 598       msg_disable ();
 599       cut_field__ (parser, &line, &p, &n_columns, &tmp, &field);
 600       msg_enable ();
 601
 602       if (sa)
 603         string_array_append_nocopy (sa, ss_xstrdup (field));
 604       n++;
 605       line = p;
 606     }
 607 }
 608
 609 /* Reads a case from READER into C, which matches dictionary DICT, parsing it
 610    according to free-format syntax rules in PARSER.  Returns true if
 611    successful, false at end of file or on I/O error. */
 612 static bool
 613 parse_delimited_span (const struct data_parser *parser,
 614                       struct dfm_reader *reader,
 615                       struct dictionary *dict, struct ccase *c)
 616 {
 617   const char *output_encoding = dict_get_encoding (dict);
 618   struct string tmp = DS_EMPTY_INITIALIZER;
 619   struct field *f;
 620
 621   for (f = parser->fields; f < &parser->fields[parser->n_fields]; f++)
 622     {
 623       struct substring s;
 624       int first_column, last_column;
 625       char *error;
 626
 627       /* Cut out a field and read in a new record if necessary. */
 628       while (!cut_field (parser, reader,
 629                          &first_column, &last_column, &tmp, &s))
 630         {
 631           if (!dfm_eof (reader))
 632             dfm_forward_record (reader);
 633           if (dfm_eof (reader))
 634             {
 635               if (f > parser->fields)
 636                 msg (DW, _("Partial case discarded.  The first variable "
 637                            "missing was %s."), f->name);
 638               ds_destroy (&tmp);
 639               return false;
 640             }
 641         }
 642
 643       const char *input_encoding = dfm_reader_get_encoding (reader);
 644       error = data_in (s, input_encoding, f->format.type,
 645                        settings_get_fmt_settings (),
 646                        case_data_rw_idx (c, f->case_idx),
 647                        fmt_var_width (f->format), output_encoding);
 648       if (error != NULL)
 649         parse_error (reader, f, first_column, last_column, error);
 650     }
 651   ds_destroy (&tmp);
 652   return true;
 653 }
 654
 655 /* Reads a case from READER into C, which matches dictionary DICT, parsing it
 656    according to delimited syntax rules with one case per record in PARSER.
 657    Returns true if successful, false at end of file or on I/O error. */
 658 static bool
 659 parse_delimited_no_span (const struct data_parser *parser,
 660                          struct dfm_reader *reader,
 661                          struct dictionary *dict, struct ccase *c)
 662 {
 663   const char *output_encoding = dict_get_encoding (dict);
 664   struct string tmp = DS_EMPTY_INITIALIZER;
 665   struct substring s;
 666   struct field *f, *end;
 667
 668   if (dfm_eof (reader))
 669     return false;
 670
 671   end = &parser->fields[parser->n_fields];
 672   for (f = parser->fields; f < end; f++)
 673     {
 674       int first_column, last_column;
 675       char *error;
 676
 677       if (!cut_field (parser, reader, &first_column, &last_column, &tmp, &s))
 678         {
 679           if (f < end - 1 && settings_get_undefined () && parser->warn_missing_fields)
 680             msg (DW, _("Missing value(s) for all variables from %s onward.  "
 681                        "These will be filled with the system-missing value "
 682                        "or blanks, as appropriate."),
 683                  f->name);
 684           for (; f < end; f++)
 685             value_set_missing (case_data_rw_idx (c, f->case_idx),
 686                                fmt_var_width (f->format));
 687           goto exit;
 688         }
 689
 690       const char *input_encoding = dfm_reader_get_encoding (reader);
 691       error = data_in (s, input_encoding, f->format.type,
 692                        settings_get_fmt_settings (),
 693                        case_data_rw_idx (c, f->case_idx),
 694                        fmt_var_width (f->format), output_encoding);
 695       if (error != NULL)
 696         parse_error (reader, f, first_column, last_column, error);
 697     }
 698
 699   s = dfm_get_record (reader);
 700   ss_ltrim (&s, parser->soft_seps);
 701   if (!ss_is_empty (s))
 702     msg (DW, _("Record ends in data not part of any field."));
 703
 704 exit:
 705   dfm_forward_record (reader);
 706   ds_destroy (&tmp);
 707   return true;
 708 }
 709 \f
 710 /* Displays a table giving information on fixed-format variable
 711    parsing on DATA LIST. */
 712 static void
 713 dump_fixed_table (const struct data_parser *parser,
 714                   const struct file_handle *fh)
 715 {
 716   /* XXX This should not be preformatted. */
 717   char *title = xasprintf (ngettext ("Reading %d record from %s.",
 718                                      "Reading %d records from %s.",
 719                                      parser->records_per_case),
 720                            parser->records_per_case, fh_get_name (fh));
 721   struct pivot_table *table = pivot_table_create__ (
 722     pivot_value_new_user_text (title, -1), "Fixed Data Records");
 723   free (title);
 724
 725   pivot_dimension_create (
 726     table, PIVOT_AXIS_COLUMN, N_("Attributes"),
 727     N_("Record"), N_("Columns"), N_("Format"));
 728
 729   struct pivot_dimension *variables = pivot_dimension_create (
 730     table, PIVOT_AXIS_ROW, N_("Variable"));
 731   variables->root->show_label = true;
 732   for (size_t i = 0; i < parser->n_fields; i++)
 733     {
 734       struct field *f = &parser->fields[i];
 735
 736       /* XXX It would be better to have the actual variable here. */
 737       int variable_idx = pivot_category_create_leaf (
 738         variables->root, pivot_value_new_user_text (f->name, -1));
 739
 740       pivot_table_put2 (table, 0, variable_idx,
 741                         pivot_value_new_integer (f->record));
 742
 743       int first_column = f->first_column;
 744       int last_column = f->first_column + f->format.w - 1;
 745       char *columns = xasprintf ("%d-%d", first_column, last_column);
 746       pivot_table_put2 (table, 1, variable_idx,
 747                         pivot_value_new_user_text (columns, -1));
 748       free (columns);
 749
 750       char str[FMT_STRING_LEN_MAX + 1];
 751       pivot_table_put2 (table, 2, variable_idx,
 752                         pivot_value_new_user_text (
 753                           fmt_to_string (f->format, str), -1));
 754
 755     }
 756
 757   pivot_table_submit (table);
 758 }
 759
 760 /* Displays a table giving information on free-format variable parsing
 761    on DATA LIST. */
 762 static void
 763 dump_delimited_table (const struct data_parser *parser,
 764                       const struct file_handle *fh)
 765 {
 766   struct pivot_table *table = pivot_table_create__ (
 767     pivot_value_new_text_format (N_("Reading free-form data from %s."),
 768                                  fh_get_name (fh)),
 769     "Free-Form Data Records");
 770
 771   pivot_dimension_create (
 772     table, PIVOT_AXIS_COLUMN, N_("Attributes"), N_("Format"));
 773
 774   struct pivot_dimension *variables = pivot_dimension_create (
 775     table, PIVOT_AXIS_ROW, N_("Variable"));
 776   variables->root->show_label = true;
 777   for (size_t i = 0; i < parser->n_fields; i++)
 778     {
 779       struct field *f = &parser->fields[i];
 780
 781       /* XXX It would be better to have the actual variable here. */
 782       int variable_idx = pivot_category_create_leaf (
 783         variables->root, pivot_value_new_user_text (f->name, -1));
 784
 785       char str[FMT_STRING_LEN_MAX + 1];
 786       pivot_table_put2 (table, 0, variable_idx,
 787                         pivot_value_new_user_text (
 788                           fmt_to_string (f->format, str), -1));
 789     }
 790
 791   pivot_table_submit (table);
 792 }
 793
 794 /* Displays a table giving information on how PARSER will read
 795    data from FH. */
 796 void
 797 data_parser_output_description (struct data_parser *parser,
 798                                 const struct file_handle *fh)
 799 {
 800   if (parser->type == DP_FIXED)
 801     dump_fixed_table (parser, fh);
 802   else
 803     dump_delimited_table (parser, fh);
 804 }
 805 \f
 806 /* Data parser input program. */
 807 struct data_parser_casereader
 808   {
 809     struct data_parser *parser; /* Parser. */
 810     struct dictionary *dict;    /* Dictionary. */
 811     struct dfm_reader *reader;  /* Data file reader. */
 812     struct caseproto *proto;    /* Format of cases. */
 813   };
 814
 815 static const struct casereader_class data_parser_casereader_class;
 816
 817 /* Replaces DS's active dataset by an input program that reads data
 818    from READER according to the rules in PARSER, using DICT as
 819    the underlying dictionary.  Ownership of PARSER and READER is
 820    transferred to the input program, and ownership of DICT is
 821    transferred to the dataset. */
 822 void
 823 data_parser_make_active_file (struct data_parser *parser, struct dataset *ds,
 824                                struct dfm_reader *reader,
 825                               struct dictionary *dict,
 826                                struct casereader* (*func)(struct casereader *,
 827                                                           const struct dictionary *,
 828                                                           void *),
 829                                void *ud)
 830 {
 831   struct data_parser_casereader *r;
 832   struct casereader *casereader0;
 833   struct casereader *casereader1;
 834
 835   r = xmalloc (sizeof *r);
 836   r->parser = parser;
 837   r->dict = dict_ref (dict);
 838   r->reader = reader;
 839   r->proto = caseproto_ref (dict_get_proto (dict));
 840   casereader0 = casereader_create_sequential (NULL, r->proto,
 841                                              CASENUMBER_MAX,
 842                                              &data_parser_casereader_class, r);
 843
 844   if (func)
 845     casereader1 = func (casereader0, dict, ud);
 846   else
 847     casereader1 = casereader0;
 848
 849   dataset_set_dict (ds, dict);
 850   dataset_set_source (ds, casereader1);
 851 }
 852
 853
 854 static struct ccase *
 855 data_parser_casereader_read (struct casereader *reader UNUSED, void *r_)
 856 {
 857   struct data_parser_casereader *r = r_;
 858   struct ccase *c = case_create (r->proto);
 859   if (data_parser_parse (r->parser, r->reader, r->dict, c))
 860     return c;
 861   else
 862     {
 863       case_unref (c);
 864       return NULL;
 865     }
 866 }
 867
 868 static void
 869 data_parser_casereader_destroy (struct casereader *reader, void *r_)
 870 {
 871   struct data_parser_casereader *r = r_;
 872   if (dfm_reader_error (r->reader))
 873     casereader_force_error (reader);
 874   dfm_close_reader (r->reader);
 875   caseproto_unref (r->proto);
 876   dict_unref (r->dict);
 877   data_parser_destroy (r->parser);
 878   free (r);
 879 }
 880
 881 static const struct casereader_class data_parser_casereader_class =
 882   {
 883     data_parser_casereader_read,
 884     data_parser_casereader_destroy,
 885     NULL,
 886     NULL,
 887   };