Open-ILS/src/sql/Pg/upgrade/0936.schema.fix_ingest_of_xpath_based_attrs.sql

   1 -- make record attributes definitions that use xpath to extract values
   2 -- work
   3
   4 BEGIN;
   5
   6 SELECT evergreen.upgrade_deps_block_check('0936', :eg_version);
   7
   8 CREATE OR REPLACE FUNCTION metabib.reingest_record_attributes (rid BIGINT, pattr_list TEXT[] DEFAULT NULL, prmarc TEXT DEFAULT NULL, rdeleted BOOL DEFAULT TRUE) RETURNS VOID AS $func$
   9 DECLARE
  10     transformed_xml TEXT;
  11     rmarc           TEXT := prmarc;
  12     tmp_val         TEXT;
  13     prev_xfrm       TEXT;
  14     normalizer      RECORD;
  15     xfrm            config.xml_transform%ROWTYPE;
  16     attr_vector     INT[] := '{}'::INT[];
  17     attr_vector_tmp INT[];
  18     attr_list       TEXT[] := pattr_list;
  19     attr_value      TEXT[];
  20     norm_attr_value TEXT[];
  21     tmp_xml         TEXT;
  22     attr_def        config.record_attr_definition%ROWTYPE;
  23     ccvm_row        config.coded_value_map%ROWTYPE;
  24 BEGIN
  25
  26     IF attr_list IS NULL OR rdeleted THEN -- need to do the full dance on INSERT or undelete
  27         SELECT ARRAY_AGG(name) INTO attr_list FROM config.record_attr_definition;
  28     END IF;
  29
  30     IF rmarc IS NULL THEN
  31         SELECT marc INTO rmarc FROM biblio.record_entry WHERE id = rid;
  32     END IF;
  33
  34     FOR attr_def IN SELECT * FROM config.record_attr_definition WHERE NOT composite AND name = ANY( attr_list ) ORDER BY format LOOP
  35
  36         attr_value := '{}'::TEXT[];
  37         norm_attr_value := '{}'::TEXT[];
  38         attr_vector_tmp := '{}'::INT[];
  39
  40         SELECT * INTO ccvm_row FROM config.coded_value_map c WHERE c.ctype = attr_def.name LIMIT 1;
  41
  42         -- tag+sf attrs only support SVF
  43         IF attr_def.tag IS NOT NULL THEN -- tag (and optional subfield list) selection
  44             SELECT  ARRAY[ARRAY_TO_STRING(ARRAY_AGG(value), COALESCE(attr_def.joiner,' '))] INTO attr_value
  45               FROM  (SELECT * FROM metabib.full_rec ORDER BY tag, subfield) AS x
  46               WHERE record = rid
  47                     AND tag LIKE attr_def.tag
  48                     AND CASE
  49                         WHEN attr_def.sf_list IS NOT NULL
  50                             THEN POSITION(subfield IN attr_def.sf_list) > 0
  51                         ELSE TRUE
  52                     END
  53               GROUP BY tag
  54               ORDER BY tag
  55               LIMIT 1;
  56
  57         ELSIF attr_def.fixed_field IS NOT NULL THEN -- a named fixed field, see config.marc21_ff_pos_map.fixed_field
  58             attr_value := vandelay.marc21_extract_fixed_field_list(rmarc, attr_def.fixed_field);
  59
  60             IF NOT attr_def.multi THEN
  61                 attr_value := ARRAY[attr_value[1]];
  62             END IF;
  63
  64         ELSIF attr_def.xpath IS NOT NULL THEN -- and xpath expression
  65
  66             SELECT INTO xfrm * FROM config.xml_transform WHERE name = attr_def.format;
  67
  68             -- See if we can skip the XSLT ... it's expensive
  69             IF prev_xfrm IS NULL OR prev_xfrm <> xfrm.name THEN
  70                 -- Can't skip the transform
  71                 IF xfrm.xslt <> '---' THEN
  72                     transformed_xml := oils_xslt_process(rmarc,xfrm.xslt);
  73                 ELSE
  74                     transformed_xml := rmarc;
  75                 END IF;
  76
  77                 prev_xfrm := xfrm.name;
  78             END IF;
  79
  80             IF xfrm.name IS NULL THEN
  81                 -- just grab the marcxml (empty) transform
  82                 SELECT INTO xfrm * FROM config.xml_transform WHERE xslt = '---' LIMIT 1;
  83                 prev_xfrm := xfrm.name;
  84             END IF;
  85
  86             FOR tmp_xml IN SELECT UNNEST(oils_xpath(attr_def.xpath, transformed_xml, ARRAY[ARRAY[xfrm.prefix, xfrm.namespace_uri]])) LOOP
  87                 tmp_val := oils_xpath_string(
  88                                 '//*',
  89                                 tmp_xml,
  90                                 COALESCE(attr_def.joiner,' '),
  91                                 ARRAY[ARRAY[xfrm.prefix, xfrm.namespace_uri]]
  92                             );
  93                 IF tmp_val IS NOT NULL AND BTRIM(tmp_val) <> '' THEN
  94                     attr_value := attr_value || tmp_val;
  95                     EXIT WHEN NOT attr_def.multi;
  96                 END IF;
  97             END LOOP;
  98
  99         ELSIF attr_def.phys_char_sf IS NOT NULL THEN -- a named Physical Characteristic, see config.marc21_physical_characteristic_*_map
 100             SELECT  ARRAY_AGG(m.value) INTO attr_value
 101               FROM  vandelay.marc21_physical_characteristics(rmarc) v
 102                     LEFT JOIN config.marc21_physical_characteristic_value_map m ON (m.id = v.value)
 103               WHERE v.subfield = attr_def.phys_char_sf AND (m.value IS NOT NULL AND BTRIM(m.value) <> '')
 104                     AND ( ccvm_row.id IS NULL OR ( ccvm_row.id IS NOT NULL AND v.id IS NOT NULL) );
 105
 106             IF NOT attr_def.multi THEN
 107                 attr_value := ARRAY[attr_value[1]];
 108             END IF;
 109
 110         END IF;
 111
 112                 -- apply index normalizers to attr_value
 113         FOR tmp_val IN SELECT value FROM UNNEST(attr_value) x(value) LOOP
 114             FOR normalizer IN
 115                 SELECT  n.func AS func,
 116                         n.param_count AS param_count,
 117                         m.params AS params
 118                   FROM  config.index_normalizer n
 119                         JOIN config.record_attr_index_norm_map m ON (m.norm = n.id)
 120                   WHERE attr = attr_def.name
 121                   ORDER BY m.pos LOOP
 122                     EXECUTE 'SELECT ' || normalizer.func || '(' ||
 123                     COALESCE( quote_literal( tmp_val ), 'NULL' ) ||
 124                         CASE
 125                             WHEN normalizer.param_count > 0
 126                                 THEN ',' || REPLACE(REPLACE(BTRIM(normalizer.params,'[]'),E'\'',E'\\\''),E'"',E'\'')
 127                                 ELSE ''
 128                             END ||
 129                     ')' INTO tmp_val;
 130
 131             END LOOP;
 132             IF tmp_val IS NOT NULL AND tmp_val <> '' THEN
 133                 -- note that a string that contains only blanks
 134                 -- is a valid value for some attributes
 135                 norm_attr_value := norm_attr_value || tmp_val;
 136             END IF;
 137         END LOOP;
 138
 139         IF attr_def.filter THEN
 140             -- Create unknown uncontrolled values and find the IDs of the values
 141             IF ccvm_row.id IS NULL THEN
 142                 FOR tmp_val IN SELECT value FROM UNNEST(norm_attr_value) x(value) LOOP
 143                     IF tmp_val IS NOT NULL AND BTRIM(tmp_val) <> '' THEN
 144                         BEGIN -- use subtransaction to isolate unique constraint violations
 145                             INSERT INTO metabib.uncontrolled_record_attr_value ( attr, value ) VALUES ( attr_def.name, tmp_val );
 146                         EXCEPTION WHEN unique_violation THEN END;
 147                     END IF;
 148                 END LOOP;
 149
 150                 SELECT ARRAY_AGG(id) INTO attr_vector_tmp FROM metabib.uncontrolled_record_attr_value WHERE attr = attr_def.name AND value = ANY( norm_attr_value );
 151             ELSE
 152                 SELECT ARRAY_AGG(id) INTO attr_vector_tmp FROM config.coded_value_map WHERE ctype = attr_def.name AND code = ANY( norm_attr_value );
 153             END IF;
 154
 155             -- Add the new value to the vector
 156             attr_vector := attr_vector || attr_vector_tmp;
 157         END IF;
 158
 159         IF attr_def.sorter AND norm_attr_value[1] IS NOT NULL THEN
 160             DELETE FROM metabib.record_sorter WHERE source = rid AND attr = attr_def.name;
 161             INSERT INTO metabib.record_sorter (source, attr, value) VALUES (rid, attr_def.name, norm_attr_value[1]);
 162         END IF;
 163
 164     END LOOP;
 165
 166 /* We may need to rewrite the vlist to contain
 167    the intersection of new values for requested
 168    attrs and old values for ignored attrs. To
 169    do this, we take the old attr vlist and
 170    subtract any values that are valid for the
 171    requested attrs, and then add back the new
 172    set of attr values. */
 173
 174     IF ARRAY_LENGTH(pattr_list, 1) > 0 THEN
 175         SELECT vlist INTO attr_vector_tmp FROM metabib.record_attr_vector_list WHERE source = rid;
 176         SELECT attr_vector_tmp - ARRAY_AGG(id::INT) INTO attr_vector_tmp FROM metabib.full_attr_id_map WHERE attr = ANY (pattr_list);
 177         attr_vector := attr_vector || attr_vector_tmp;
 178     END IF;
 179
 180     -- On to composite attributes, now that the record attrs have been pulled.  Processed in name order, so later composite
 181     -- attributes can depend on earlier ones.
 182     PERFORM metabib.compile_composite_attr_cache_init();
 183     FOR attr_def IN SELECT * FROM config.record_attr_definition WHERE composite AND name = ANY( attr_list ) ORDER BY name LOOP
 184
 185         FOR ccvm_row IN SELECT * FROM config.coded_value_map c WHERE c.ctype = attr_def.name ORDER BY value LOOP
 186
 187             tmp_val := metabib.compile_composite_attr( ccvm_row.id );
 188             CONTINUE WHEN tmp_val IS NULL OR tmp_val = ''; -- nothing to do
 189
 190             IF attr_def.filter THEN
 191                 IF attr_vector @@ tmp_val::query_int THEN
 192                     attr_vector = attr_vector + intset(ccvm_row.id);
 193                     EXIT WHEN NOT attr_def.multi;
 194                 END IF;
 195             END IF;
 196
 197             IF attr_def.sorter THEN
 198                 IF attr_vector @@ tmp_val THEN
 199                     DELETE FROM metabib.record_sorter WHERE source = rid AND attr = attr_def.name;
 200                     INSERT INTO metabib.record_sorter (source, attr, value) VALUES (rid, attr_def.name, ccvm_row.code);
 201                 END IF;
 202             END IF;
 203
 204         END LOOP;
 205
 206     END LOOP;
 207
 208     IF ARRAY_LENGTH(attr_vector, 1) > 0 THEN
 209         IF rdeleted THEN -- initial insert OR revivication
 210             DELETE FROM metabib.record_attr_vector_list WHERE source = rid;
 211             INSERT INTO metabib.record_attr_vector_list (source, vlist) VALUES (rid, attr_vector);
 212         ELSE
 213             UPDATE metabib.record_attr_vector_list SET vlist = attr_vector WHERE source = rid;
 214         END IF;
 215     END IF;
 216
 217 END;
 218
 219 $func$ LANGUAGE PLPGSQL;
 220
 221 COMMIT;