* set.q (show_warranty): Use fputs to print lack-of-warranty
[pspp-builds.git] / src / data / casereader.c
1 /* PSPP - a program for statistical analysis.
2    Copyright (C) 2007 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 <data/casereader.h>
20 #include <data/casereader-provider.h>
21
22 #include <stdlib.h>
23
24 #include <data/casewindow.h>
25 #include <data/casewriter.h>
26 #include <data/settings.h>
27 #include <libpspp/assertion.h>
28 #include <libpspp/heap.h>
29 #include <libpspp/taint.h>
30
31 #include "xalloc.h"
32
33 /* A casereader. */
34 struct casereader
35   {
36     struct taint *taint;                  /* Corrupted? */
37     size_t value_cnt;                     /* Values per case. */
38     casenumber case_cnt;                  /* Number of cases,
39                                              CASENUMBER_MAX if unknown. */
40     const struct casereader_class *class; /* Class. */
41     void *aux;                            /* Auxiliary data for class. */
42   };
43
44 static void insert_shim (struct casereader *);
45
46 /* Creates a new case in C and reads the next case from READER
47    into it.  The caller owns C and must destroy C when its data
48    is no longer needed.  Return true if successful, false when
49    cases have been exhausted or upon detection of an I/O error.
50    In the latter case, C is set to the null case.
51
52    The case returned is effectively consumed: it can never be
53    read again through READER.  If this is inconvenient, READER
54    may be cloned in advance with casereader_clone, or
55    casereader_peek may be used instead. */
56 bool
57 casereader_read (struct casereader *reader, struct ccase *c)
58 {
59   if (reader->case_cnt != 0 && reader->class->read (reader, reader->aux, c))
60     {
61       assert (case_get_value_cnt (c) >= reader->value_cnt);
62       if (reader->case_cnt != CASENUMBER_MAX)
63         reader->case_cnt--;
64       return true;
65     }
66   else
67     {
68       reader->case_cnt = 0;
69       case_nullify (c);
70       return false;
71     }
72 }
73
74 /* Destroys READER.
75    Returns false if an I/O error was detected on READER, true
76    otherwise. */
77 bool
78 casereader_destroy (struct casereader *reader)
79 {
80   bool ok = true;
81   if (reader != NULL)
82     {
83       reader->class->destroy (reader, reader->aux);
84       ok = taint_destroy (reader->taint);
85       free (reader);
86     }
87   return ok;
88 }
89
90 /* Returns a clone of READER.  READER and its clone may be used
91    to read the same sequence of cases in the same order, barring
92    I/O errors. */
93 struct casereader *
94 casereader_clone (const struct casereader *reader_)
95 {
96   struct casereader *reader = (struct casereader *) reader_;
97   struct casereader *clone;
98   if ( reader == NULL ) 
99     return NULL;
100
101   if (reader->class->clone == NULL)
102     insert_shim (reader);
103   clone = reader->class->clone (reader, reader->aux);
104   assert (clone != NULL);
105   assert (clone != reader);
106   return clone;
107 }
108
109 /* Makes a copy of ORIGINAL into *NEW1 (if NEW1 is non-null) and
110    *NEW2 (if NEW2 is non-null), then destroys ORIGINAL. */
111 void
112 casereader_split (struct casereader *original,
113                   struct casereader **new1, struct casereader **new2)
114 {
115   if (new1 != NULL && new2 != NULL)
116     {
117       *new1 = casereader_rename (original);
118       *new2 = casereader_clone (*new1);
119     }
120   else if (new1 != NULL)
121     *new1 = casereader_rename (original);
122   else if (new2 != NULL)
123     *new2 = casereader_rename (original);
124   else
125     casereader_destroy (original);
126 }
127
128 /* Returns a copy of READER, which is itself destroyed.
129    Useful for taking over ownership of a casereader, to enforce
130    preventing the original owner from accessing the casereader
131    again. */
132 struct casereader *
133 casereader_rename (struct casereader *reader)
134 {
135   struct casereader *new = xmemdup (reader, sizeof *reader);
136   free (reader);
137   return new;
138 }
139
140 /* Exchanges the casereaders referred to by A and B. */
141 void
142 casereader_swap (struct casereader *a, struct casereader *b)
143 {
144   if (a != b)
145     {
146       struct casereader tmp = *a;
147       *a = *b;
148       *b = tmp;
149     }
150 }
151
152 /* Creates a new case in C and reads the (IDX + 1)'th case from
153    READER into it.  The caller owns C and must destroy C when its
154    data is no longer needed.  Return true if successful, false
155    when cases have been exhausted or upon detection of an I/O
156    error.  In the latter case, C is set to the null case. */
157 bool
158 casereader_peek (struct casereader *reader, casenumber idx, struct ccase *c)
159 {
160   if (idx < reader->case_cnt)
161     {
162       if (reader->class->peek == NULL)
163         insert_shim (reader);
164       if (reader->class->peek (reader, reader->aux, idx, c))
165         return true;
166       else if (casereader_error (reader))
167         reader->case_cnt = 0;
168     }
169   if (reader->case_cnt > idx)
170     reader->case_cnt = idx;
171   case_nullify (c);
172   return false;
173 }
174
175 /* Returns true if an I/O error or another hard error has
176    occurred on READER, a clone of READER, or on some object on
177    which READER's data has a dependency, false otherwise. */
178 bool
179 casereader_error (const struct casereader *reader)
180 {
181   return taint_is_tainted (reader->taint);
182 }
183
184 /* Marks READER as having encountered an error.
185
186    Ordinarily, this function should be called by the
187    implementation of a casereader, not by the casereader's
188    client.  Instead, casereader clients should usually ensure
189    that a casereader's error state is correct by using
190    taint_propagate to propagate to the casereader's taint
191    structure, which may be obtained via casereader_get_taint. */
192 void
193 casereader_force_error (struct casereader *reader)
194 {
195   taint_set_taint (reader->taint);
196 }
197
198 /* Returns READER's associate taint object, for use with
199    taint_propagate and other taint functions. */
200 const struct taint *
201 casereader_get_taint (const struct casereader *reader)
202 {
203   return reader->taint;
204 }
205
206 /* Returns the number of cases that will be read by successive
207    calls to casereader_read for READER, assuming that no errors
208    occur.  Upon an error condition, the case count drops to 0, so
209    that no more cases can be obtained.
210
211    Not all casereaders can predict the number of cases that they
212    will produce without actually reading all of them.  In that
213    case, this function returns CASENUMBER_MAX.  To obtain the
214    actual number of cases in such a casereader, use
215    casereader_count_cases. */
216 casenumber
217 casereader_get_case_cnt (struct casereader *reader)
218 {
219   return reader->case_cnt;
220 }
221
222 /* Returns the number of cases that will be read by successive
223    calls to casereader_read for READER, assuming that no errors
224    occur.  Upon an error condition, the case count drops to 0, so
225    that no more cases can be obtained.
226
227    For a casereader that cannot predict the number of cases it
228    will produce, this function actually reads (and discards) all
229    of the contents of a clone of READER.  Thus, the return value
230    is always correct in the absence of I/O errors. */
231 casenumber
232 casereader_count_cases (struct casereader *reader)
233 {
234   if (reader->case_cnt == CASENUMBER_MAX)
235     {
236       casenumber n_cases = 0;
237       struct ccase c;
238
239       struct casereader *clone = casereader_clone (reader);
240
241       for (; casereader_read (clone, &c); case_destroy (&c))
242         n_cases++;
243
244       casereader_destroy (clone);
245       reader->case_cnt = n_cases;
246     }
247
248   return reader->case_cnt;
249 }
250
251 /* Returns the number of struct values in each case in READER. */
252 size_t
253 casereader_get_value_cnt (struct casereader *reader)
254 {
255   return reader->value_cnt;
256 }
257
258 /* Copies all the cases in READER to WRITER, propagating errors
259    appropriately. */
260 void
261 casereader_transfer (struct casereader *reader, struct casewriter *writer)
262 {
263   struct ccase c;
264
265   taint_propagate (casereader_get_taint (reader),
266                    casewriter_get_taint (writer));
267   while (casereader_read (reader, &c))
268     casewriter_write (writer, &c);
269   casereader_destroy (reader);
270 }
271
272 /* Creates and returns a new casereader.  This function is
273    intended for use by casereader implementations, not by
274    casereader clients.
275
276    This function is most suited for creating a casereader for a
277    data source that is naturally sequential.
278    casereader_create_random may be more appropriate for a data
279    source that supports random access.
280
281    Ordinarily, specify a null pointer for TAINT, in which case
282    the new casereader will have a new, unique taint object.  If
283    the new casereader should have a clone of an existing taint
284    object, specify that object as TAINT.  (This is most commonly
285    useful in an implementation of the "clone" casereader_class
286    function, in which case the cloned casereader should have the
287    same taint object as the original casereader.)
288
289    VALUE_CNT must be the number of struct values per case read
290    from the casereader.
291
292    CASE_CNT is an upper limit on the number of cases that
293    casereader_read will return from the casereader in successive
294    calls.  Ordinarily, this is the actual number of cases in the
295    data source or CASENUMBER_MAX if the number of cases cannot be
296    predicted in advance.
297
298    CLASS and AUX are a set of casereader implementation-specific
299    member functions and auxiliary data to pass to those member
300    functions, respectively. */
301 struct casereader *
302 casereader_create_sequential (const struct taint *taint,
303                               size_t value_cnt, casenumber case_cnt,
304                               const struct casereader_class *class, void *aux)
305 {
306   struct casereader *reader = xmalloc (sizeof *reader);
307   reader->taint = taint != NULL ? taint_clone (taint) : taint_create ();
308   reader->value_cnt = value_cnt;
309   reader->case_cnt = case_cnt;
310   reader->class = class;
311   reader->aux = aux;
312   return reader;
313 }
314 \f
315 /* Random-access casereader implementation.
316
317    This is a set of wrappers around casereader_create_sequential
318    and struct casereader_class to make it easy to create
319    efficient casereaders for data sources that natively support
320    random access. */
321
322 /* One clone of a random reader. */
323 struct random_reader
324   {
325     struct random_reader_shared *shared; /* Data shared among clones. */
326     struct heap_node heap_node; /* Node in shared data's heap of readers. */
327     casenumber offset;          /* Number of cases already read. */
328   };
329
330 /* Returns the random_reader in which the given heap_node is
331    embedded. */
332 static struct random_reader *
333 random_reader_from_heap_node (const struct heap_node *node)
334 {
335   return heap_data (node, struct random_reader, heap_node);
336 }
337
338 /* Data shared among clones of a random reader. */
339 struct random_reader_shared
340   {
341     struct heap *readers;       /* Heap of struct random_readers. */
342     casenumber min_offset;      /* Smallest offset of any random_reader. */
343     const struct casereader_random_class *class;
344     void *aux;
345   };
346
347 static struct casereader_class random_reader_casereader_class;
348
349 /* Creates and returns a new random_reader with the given SHARED
350    data and OFFSET.  Inserts the new random reader into the
351    shared heap. */
352 static struct random_reader *
353 make_random_reader (struct random_reader_shared *shared, casenumber offset)
354 {
355   struct random_reader *br = xmalloc (sizeof *br);
356   br->offset = offset;
357   br->shared = shared;
358   heap_insert (shared->readers, &br->heap_node);
359   return br;
360 }
361
362 /* Compares random_readers A and B by offset and returns a
363    strcmp()-like result. */
364 static int
365 compare_random_readers_by_offset (const struct heap_node *a_,
366                                   const struct heap_node *b_,
367                                   const void *aux UNUSED)
368 {
369   const struct random_reader *a = random_reader_from_heap_node (a_);
370   const struct random_reader *b = random_reader_from_heap_node (b_);
371   return a->offset < b->offset ? -1 : a->offset > b->offset;
372 }
373
374 /* Creates and returns a new casereader.  This function is
375    intended for use by casereader implementations, not by
376    casereader clients.
377
378    This function is most suited for creating a casereader for a
379    data source that supports random access.
380    casereader_create_sequential is more appropriate for a data
381    source that is naturally sequential.
382
383    VALUE_CNT must be the number of struct values per case read
384    from the casereader.
385
386    CASE_CNT is an upper limit on the number of cases that
387    casereader_read will return from the casereader in successive
388    calls.  Ordinarily, this is the actual number of cases in the
389    data source or CASENUMBER_MAX if the number of cases cannot be
390    predicted in advance.
391
392    CLASS and AUX are a set of casereader implementation-specific
393    member functions and auxiliary data to pass to those member
394    functions, respectively. */
395 struct casereader *
396 casereader_create_random (size_t value_cnt, casenumber case_cnt,
397                           const struct casereader_random_class *class,
398                           void *aux)
399 {
400   struct random_reader_shared *shared = xmalloc (sizeof *shared);
401   shared->readers = heap_create (compare_random_readers_by_offset, NULL);
402   shared->class = class;
403   shared->aux = aux;
404   shared->min_offset = 0;
405   return casereader_create_sequential (NULL, value_cnt, case_cnt,
406                                        &random_reader_casereader_class,
407                                        make_random_reader (shared, 0));
408 }
409
410 /* Reassesses the min_offset in SHARED based on the minimum
411    offset in the heap.   */
412 static void
413 advance_random_reader (struct casereader *reader,
414                        struct random_reader_shared *shared)
415 {
416   casenumber old, new;
417
418   old = shared->min_offset;
419   new = random_reader_from_heap_node (heap_minimum (shared->readers))->offset;
420   assert (new >= old);
421   if (new > old)
422     {
423       shared->min_offset = new;
424       shared->class->advance (reader, shared->aux, new - old);
425     }
426 }
427
428 /* struct casereader_class "read" function for random reader. */
429 static bool
430 random_reader_read (struct casereader *reader, void *br_, struct ccase *c)
431 {
432   struct random_reader *br = br_;
433   struct random_reader_shared *shared = br->shared;
434
435   if (shared->class->read (reader, shared->aux,
436                            br->offset - shared->min_offset, c))
437     {
438       br->offset++;
439       heap_changed (shared->readers, &br->heap_node);
440       advance_random_reader (reader, shared);
441       return true;
442     }
443   else
444     return false;
445 }
446
447 /* struct casereader_class "destroy" function for random
448    reader. */
449 static void
450 random_reader_destroy (struct casereader *reader, void *br_)
451 {
452   struct random_reader *br = br_;
453   struct random_reader_shared *shared = br->shared;
454
455   heap_delete (shared->readers, &br->heap_node);
456   if (heap_is_empty (shared->readers))
457     {
458       heap_destroy (shared->readers);
459       shared->class->destroy (reader, shared->aux);
460       free (shared);
461     }
462   else
463     advance_random_reader (reader, shared);
464
465   free (br);
466 }
467
468 /* struct casereader_class "clone" function for random reader. */
469 static struct casereader *
470 random_reader_clone (struct casereader *reader, void *br_)
471 {
472   struct random_reader *br = br_;
473   struct random_reader_shared *shared = br->shared;
474   return casereader_create_sequential (casereader_get_taint (reader),
475                                        casereader_get_value_cnt (reader),
476                                        casereader_get_case_cnt (reader),
477                                        &random_reader_casereader_class,
478                                        make_random_reader (shared,
479                                                            br->offset));
480 }
481
482 /* struct casereader_class "peek" function for random reader. */
483 static bool
484 random_reader_peek (struct casereader *reader, void *br_,
485                     casenumber idx, struct ccase *c)
486 {
487   struct random_reader *br = br_;
488   struct random_reader_shared *shared = br->shared;
489
490   return shared->class->read (reader, shared->aux,
491                               br->offset - shared->min_offset + idx, c);
492 }
493
494 /* Casereader class for random reader. */
495 static struct casereader_class random_reader_casereader_class =
496   {
497     random_reader_read,
498     random_reader_destroy,
499     random_reader_clone,
500     random_reader_peek,
501   };
502 \f
503 /* Buffering shim for implementing clone and peek operations.
504
505    The "clone" and "peek" operations aren't implemented by all
506    types of casereaders, but we have to expose a uniform
507    interface anyhow.  We do this by interposing a buffering
508    casereader on top of the existing casereader on the first call
509    to "clone" or "peek".  The buffering casereader maintains a
510    window of cases that spans the positions of the original
511    casereader and all of its clones (the "clone set"), from the
512    position of the casereader that has read the fewest cases to
513    the position of the casereader that has read the most.
514
515    Thus, if all of the casereaders in the clone set are at
516    approximately the same position, only a few cases are buffered
517    and there is little inefficiency.  If, on the other hand, one
518    casereader is not used to read any cases at all, but another
519    one is used to read all of the cases, the entire contents of
520    the casereader is copied into the buffer.  This still might
521    not be so inefficient, given that case data in memory is
522    shared across multiple identical copies, but in the worst case
523    the window implementation will write cases to disk instead of
524    maintaining them in-memory. */
525
526 /* A buffering shim for a non-clonable or non-peekable
527    casereader. */
528 struct shim
529   {
530     struct casewindow *window;          /* Window of buffered cases. */
531     struct casereader *subreader;       /* Subordinate casereader. */
532   };
533
534 static struct casereader_random_class shim_class;
535
536 /* Interposes a buffering shim atop READER. */
537 static void
538 insert_shim (struct casereader *reader)
539 {
540   size_t value_cnt = casereader_get_value_cnt (reader);
541   casenumber case_cnt = casereader_get_case_cnt (reader);
542   struct shim *b = xmalloc (sizeof *b);
543   b->window = casewindow_create (value_cnt, get_workspace_cases (value_cnt));
544   b->subreader = casereader_create_random (value_cnt, case_cnt,
545                                            &shim_class, b);
546   casereader_swap (reader, b->subreader);
547   taint_propagate (casewindow_get_taint (b->window),
548                    casereader_get_taint (reader));
549   taint_propagate (casereader_get_taint (b->subreader),
550                    casereader_get_taint (reader));
551 }
552
553 /* Ensures that B's window contains at least CASE_CNT cases.
554    Return true if successful, false upon reaching the end of B's
555    subreader or an I/O error. */
556 static bool
557 prime_buffer (struct shim *b, casenumber case_cnt)
558 {
559   while (casewindow_get_case_cnt (b->window) < case_cnt)
560     {
561       struct ccase tmp;
562       if (!casereader_read (b->subreader, &tmp))
563         return false;
564       casewindow_push_head (b->window, &tmp);
565     }
566   return true;
567 }
568
569 /* Reads the case at the given 0-based OFFSET from the front of
570    the window into C.  Returns true if successful, false if
571    OFFSET is beyond the end of file or upon I/O error. */
572 static bool
573 shim_read (struct casereader *reader UNUSED, void *b_,
574            casenumber offset, struct ccase *c)
575 {
576   struct shim *b = b_;
577   return (prime_buffer (b, offset + 1)
578           && casewindow_get_case (b->window, offset, c));
579 }
580
581 /* Destroys B. */
582 static void
583 shim_destroy (struct casereader *reader UNUSED, void *b_)
584 {
585   struct shim *b = b_;
586   casewindow_destroy (b->window);
587   casereader_destroy (b->subreader);
588   free (b);
589 }
590
591 /* Discards CNT cases from the front of B's window. */
592 static void
593 shim_advance (struct casereader *reader UNUSED, void *b_, casenumber case_cnt)
594 {
595   struct shim *b = b_;
596   casewindow_pop_tail (b->window, case_cnt);
597 }
598
599 /* Class for the buffered reader. */
600 static struct casereader_random_class shim_class =
601   {
602     shim_read,
603     shim_destroy,
604     shim_advance,
605   };