Open-ILS/src/sql/Pg/030.schema.metabib.sql

   1 /*
   2  * Copyright (C) 2004-2008  Georgia Public Library Service
   3  * Copyright (C) 2007-2008  Equinox Software, Inc.
   4  * Mike Rylander <miker@esilibrary.com>
   5  *
   6  * This program is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU General Public License
   8  * as published by the Free Software Foundation; either version 2
   9  * of the License, or (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14  * GNU General Public License for more details.
  15  *
  16  */
  17
  18 DROP SCHEMA IF EXISTS metabib CASCADE;
  19
  20 BEGIN;
  21 CREATE SCHEMA metabib;
  22
  23 CREATE TABLE metabib.metarecord (
  24         id              BIGSERIAL       PRIMARY KEY,
  25         fingerprint     TEXT            NOT NULL,
  26         master_record   BIGINT,
  27         mods            TEXT
  28 );
  29 CREATE INDEX metabib_metarecord_master_record_idx ON metabib.metarecord (master_record);
  30 CREATE INDEX metabib_metarecord_fingerprint_idx ON metabib.metarecord (fingerprint);
  31
  32 CREATE TABLE metabib.identifier_field_entry (
  33         id              BIGSERIAL       PRIMARY KEY,
  34         source          BIGINT          NOT NULL,
  35         field           INT             NOT NULL,
  36         value           TEXT            NOT NULL,
  37         index_vector    tsvector        NOT NULL
  38 );
  39 CREATE TRIGGER metabib_identifier_field_entry_fti_trigger
  40         BEFORE UPDATE OR INSERT ON metabib.identifier_field_entry
  41         FOR EACH ROW EXECUTE PROCEDURE oils_tsearch2('identifier');
  42
  43 CREATE INDEX metabib_identifier_field_entry_index_vector_idx ON metabib.identifier_field_entry USING GIST (index_vector);
  44 CREATE INDEX metabib_identifier_field_entry_value_idx ON metabib.identifier_field_entry (SUBSTRING(value,1,1024)) WHERE index_vector = ''::TSVECTOR;
  45 CREATE INDEX metabib_identifier_field_entry_source_idx ON metabib.identifier_field_entry (source);
  46
  47 CREATE TABLE metabib.combined_identifier_field_entry (
  48         record          BIGINT          NOT NULL,
  49         metabib_field           INT             NULL,
  50         index_vector    tsvector        NOT NULL
  51 );
  52 CREATE UNIQUE INDEX metabib_combined_identifier_field_entry_fakepk_idx ON metabib.combined_identifier_field_entry (record, COALESCE(metabib_field::TEXT,''));
  53 CREATE INDEX metabib_combined_identifier_field_entry_index_vector_idx ON metabib.combined_identifier_field_entry USING GIST (index_vector);
  54 CREATE INDEX metabib_combined_identifier_field_source_idx ON metabib.combined_identifier_field_entry (metabib_field);
  55
  56 CREATE TABLE metabib.title_field_entry (
  57         id              BIGSERIAL       PRIMARY KEY,
  58         source          BIGINT          NOT NULL,
  59         field           INT             NOT NULL,
  60         value           TEXT            NOT NULL,
  61         index_vector    tsvector        NOT NULL
  62 );
  63 CREATE TRIGGER metabib_title_field_entry_fti_trigger
  64         BEFORE UPDATE OR INSERT ON metabib.title_field_entry
  65         FOR EACH ROW EXECUTE PROCEDURE oils_tsearch2('title');
  66
  67 CREATE INDEX metabib_title_field_entry_index_vector_idx ON metabib.title_field_entry USING GIST (index_vector);
  68 CREATE INDEX metabib_title_field_entry_value_idx ON metabib.title_field_entry (SUBSTRING(value,1,1024)) WHERE index_vector = ''::TSVECTOR;
  69 CREATE INDEX metabib_title_field_entry_source_idx ON metabib.title_field_entry (source);
  70
  71 CREATE TABLE metabib.combined_title_field_entry (
  72         record          BIGINT          NOT NULL,
  73         metabib_field           INT             NULL,
  74         index_vector    tsvector        NOT NULL
  75 );
  76 CREATE UNIQUE INDEX metabib_combined_title_field_entry_fakepk_idx ON metabib.combined_title_field_entry (record, COALESCE(metabib_field::TEXT,''));
  77 CREATE INDEX metabib_combined_title_field_entry_index_vector_idx ON metabib.combined_title_field_entry USING GIST (index_vector);
  78 CREATE INDEX metabib_combined_title_field_source_idx ON metabib.combined_title_field_entry (metabib_field);
  79
  80 CREATE TABLE metabib.author_field_entry (
  81         id              BIGSERIAL       PRIMARY KEY,
  82         source          BIGINT          NOT NULL,
  83         field           INT             NOT NULL,
  84         value           TEXT            NOT NULL,
  85         index_vector    tsvector        NOT NULL
  86 );
  87 CREATE TRIGGER metabib_author_field_entry_fti_trigger
  88         BEFORE UPDATE OR INSERT ON metabib.author_field_entry
  89         FOR EACH ROW EXECUTE PROCEDURE oils_tsearch2('author');
  90
  91 CREATE INDEX metabib_author_field_entry_index_vector_idx ON metabib.author_field_entry USING GIST (index_vector);
  92 CREATE INDEX metabib_author_field_entry_value_idx ON metabib.author_field_entry (SUBSTRING(value,1,1024)) WHERE index_vector = ''::TSVECTOR;
  93 CREATE INDEX metabib_author_field_entry_source_idx ON metabib.author_field_entry (source);
  94
  95 CREATE TABLE metabib.combined_author_field_entry (
  96         record          BIGINT          NOT NULL,
  97         metabib_field           INT             NULL,
  98         index_vector    tsvector        NOT NULL
  99 );
 100 CREATE UNIQUE INDEX metabib_combined_author_field_entry_fakepk_idx ON metabib.combined_author_field_entry (record, COALESCE(metabib_field::TEXT,''));
 101 CREATE INDEX metabib_combined_author_field_entry_index_vector_idx ON metabib.combined_author_field_entry USING GIST (index_vector);
 102 CREATE INDEX metabib_combined_author_field_source_idx ON metabib.combined_author_field_entry (metabib_field);
 103
 104 CREATE TABLE metabib.subject_field_entry (
 105         id              BIGSERIAL       PRIMARY KEY,
 106         source          BIGINT          NOT NULL,
 107         field           INT             NOT NULL,
 108         value           TEXT            NOT NULL,
 109         index_vector    tsvector        NOT NULL
 110 );
 111 CREATE TRIGGER metabib_subject_field_entry_fti_trigger
 112         BEFORE UPDATE OR INSERT ON metabib.subject_field_entry
 113         FOR EACH ROW EXECUTE PROCEDURE oils_tsearch2('subject');
 114
 115 CREATE INDEX metabib_subject_field_entry_index_vector_idx ON metabib.subject_field_entry USING GIST (index_vector);
 116 CREATE INDEX metabib_subject_field_entry_value_idx ON metabib.subject_field_entry (SUBSTRING(value,1,1024)) WHERE index_vector = ''::TSVECTOR;
 117 CREATE INDEX metabib_subject_field_entry_source_idx ON metabib.subject_field_entry (source);
 118
 119 CREATE TABLE metabib.combined_subject_field_entry (
 120         record          BIGINT          NOT NULL,
 121         metabib_field           INT             NULL,
 122         index_vector    tsvector        NOT NULL
 123 );
 124 CREATE UNIQUE INDEX metabib_combined_subject_field_entry_fakepk_idx ON metabib.combined_subject_field_entry (record, COALESCE(metabib_field::TEXT,''));
 125 CREATE INDEX metabib_combined_subject_field_entry_index_vector_idx ON metabib.combined_subject_field_entry USING GIST (index_vector);
 126 CREATE INDEX metabib_combined_subject_field_source_idx ON metabib.combined_subject_field_entry (metabib_field);
 127
 128 CREATE TABLE metabib.keyword_field_entry (
 129         id              BIGSERIAL       PRIMARY KEY,
 130         source          BIGINT          NOT NULL,
 131         field           INT             NOT NULL,
 132         value           TEXT            NOT NULL,
 133         index_vector    tsvector        NOT NULL
 134 );
 135 CREATE TRIGGER metabib_keyword_field_entry_fti_trigger
 136         BEFORE UPDATE OR INSERT ON metabib.keyword_field_entry
 137         FOR EACH ROW EXECUTE PROCEDURE oils_tsearch2('keyword');
 138
 139 CREATE INDEX metabib_keyword_field_entry_index_vector_idx ON metabib.keyword_field_entry USING GIST (index_vector);
 140 CREATE INDEX metabib_keyword_field_entry_value_idx ON metabib.keyword_field_entry (SUBSTRING(value,1,1024)) WHERE index_vector = ''::TSVECTOR;
 141 CREATE INDEX metabib_keyword_field_entry_source_idx ON metabib.keyword_field_entry (source);
 142
 143 CREATE TABLE metabib.combined_keyword_field_entry (
 144         record          BIGINT          NOT NULL,
 145         metabib_field           INT             NULL,
 146         index_vector    tsvector        NOT NULL
 147 );
 148 CREATE UNIQUE INDEX metabib_combined_keyword_field_entry_fakepk_idx ON metabib.combined_keyword_field_entry (record, COALESCE(metabib_field::TEXT,''));
 149 CREATE INDEX metabib_combined_keyword_field_entry_index_vector_idx ON metabib.combined_keyword_field_entry USING GIST (index_vector);
 150 CREATE INDEX metabib_combined_keyword_field_source_idx ON metabib.combined_keyword_field_entry (metabib_field);
 151
 152 CREATE TABLE metabib.series_field_entry (
 153         id              BIGSERIAL       PRIMARY KEY,
 154         source          BIGINT          NOT NULL,
 155         field           INT             NOT NULL,
 156         value           TEXT            NOT NULL,
 157         index_vector    tsvector        NOT NULL
 158 );
 159 CREATE TRIGGER metabib_series_field_entry_fti_trigger
 160         BEFORE UPDATE OR INSERT ON metabib.series_field_entry
 161         FOR EACH ROW EXECUTE PROCEDURE oils_tsearch2('series');
 162
 163 CREATE INDEX metabib_series_field_entry_index_vector_idx ON metabib.series_field_entry USING GIST (index_vector);
 164 CREATE INDEX metabib_series_field_entry_value_idx ON metabib.series_field_entry (SUBSTRING(value,1,1024)) WHERE index_vector = ''::TSVECTOR;
 165 CREATE INDEX metabib_series_field_entry_source_idx ON metabib.series_field_entry (source);
 166
 167 CREATE TABLE metabib.combined_series_field_entry (
 168         record          BIGINT          NOT NULL,
 169         metabib_field           INT             NULL,
 170         index_vector    tsvector        NOT NULL
 171 );
 172 CREATE UNIQUE INDEX metabib_combined_series_field_entry_fakepk_idx ON metabib.combined_series_field_entry (record, COALESCE(metabib_field::TEXT,''));
 173 CREATE INDEX metabib_combined_series_field_entry_index_vector_idx ON metabib.combined_series_field_entry USING GIST (index_vector);
 174 CREATE INDEX metabib_combined_series_field_source_idx ON metabib.combined_series_field_entry (metabib_field);
 175
 176 CREATE TABLE metabib.facet_entry (
 177         id              BIGSERIAL       PRIMARY KEY,
 178         source          BIGINT          NOT NULL,
 179         field           INT             NOT NULL,
 180         value           TEXT            NOT NULL
 181 );
 182 CREATE INDEX metabib_facet_entry_field_idx ON metabib.facet_entry (field);
 183 CREATE INDEX metabib_facet_entry_value_idx ON metabib.facet_entry (SUBSTRING(value,1,1024));
 184 CREATE INDEX metabib_facet_entry_source_idx ON metabib.facet_entry (source);
 185
 186 CREATE TABLE metabib.browse_entry (
 187     id BIGSERIAL PRIMARY KEY,
 188     value TEXT,
 189     index_vector tsvector,
 190     sort_value  TEXT NOT NULL,
 191     UNIQUE(sort_value, value)
 192 );
 193
 194
 195 CREATE INDEX browse_entry_sort_value_idx
 196     ON metabib.browse_entry USING BTREE (sort_value);
 197
 198 CREATE INDEX metabib_browse_entry_index_vector_idx ON metabib.browse_entry USING GIN (index_vector);
 199 CREATE TRIGGER metabib_browse_entry_fti_trigger
 200     BEFORE INSERT OR UPDATE ON metabib.browse_entry
 201     FOR EACH ROW EXECUTE PROCEDURE oils_tsearch2('keyword');
 202
 203
 204 CREATE TABLE metabib.browse_entry_def_map (
 205     id BIGSERIAL PRIMARY KEY,
 206     entry BIGINT REFERENCES metabib.browse_entry (id),
 207     def INT REFERENCES config.metabib_field (id) ON UPDATE CASCADE DEFERRABLE INITIALLY DEFERRED,
 208     source BIGINT REFERENCES biblio.record_entry (id),
 209     authority BIGINT REFERENCES authority.record_entry (id) ON DELETE SET NULL
 210 );
 211 CREATE INDEX browse_entry_def_map_def_idx ON metabib.browse_entry_def_map (def);
 212 CREATE INDEX browse_entry_def_map_entry_idx ON metabib.browse_entry_def_map (entry);
 213 CREATE INDEX browse_entry_def_map_source_idx ON metabib.browse_entry_def_map (source);
 214
 215 CREATE TABLE metabib.browse_entry_simple_heading_map (
 216     id BIGSERIAL PRIMARY KEY,
 217     entry BIGINT REFERENCES metabib.browse_entry (id),
 218     simple_heading BIGINT REFERENCES authority.simple_heading (id) ON DELETE CASCADE
 219 );
 220 CREATE INDEX browse_entry_sh_map_entry_idx ON metabib.browse_entry_simple_heading_map (entry);
 221 CREATE INDEX browse_entry_sh_map_sh_idx ON metabib.browse_entry_simple_heading_map (simple_heading);
 222
 223 CREATE OR REPLACE FUNCTION metabib.facet_normalize_trigger () RETURNS TRIGGER AS $$
 224 DECLARE
 225     normalizer  RECORD;
 226     facet_text  TEXT;
 227 BEGIN
 228     facet_text := NEW.value;
 229
 230     FOR normalizer IN
 231         SELECT  n.func AS func,
 232                 n.param_count AS param_count,
 233                 m.params AS params
 234           FROM  config.index_normalizer n
 235                 JOIN config.metabib_field_index_norm_map m ON (m.norm = n.id)
 236           WHERE m.field = NEW.field AND m.pos < 0
 237           ORDER BY m.pos LOOP
 238
 239             EXECUTE 'SELECT ' || normalizer.func || '(' ||
 240                 quote_literal( facet_text ) ||
 241                 CASE
 242                     WHEN normalizer.param_count > 0
 243                         THEN ',' || REPLACE(REPLACE(BTRIM(normalizer.params,'[]'),E'\'',E'\\\''),E'"',E'\'')
 244                         ELSE ''
 245                     END ||
 246                 ')' INTO facet_text;
 247
 248     END LOOP;
 249
 250     NEW.value = facet_text;
 251
 252     RETURN NEW;
 253 END;
 254 $$ LANGUAGE PLPGSQL;
 255
 256 CREATE TRIGGER facet_normalize_tgr
 257         BEFORE UPDATE OR INSERT ON metabib.facet_entry
 258         FOR EACH ROW EXECUTE PROCEDURE metabib.facet_normalize_trigger();
 259
 260 CREATE OR REPLACE FUNCTION evergreen.facet_force_nfc() RETURNS TRIGGER AS $$
 261 BEGIN
 262     NEW.value := force_unicode_normal_form(NEW.value,'NFC');
 263     RETURN NEW;
 264 END;
 265 $$ LANGUAGE PLPGSQL;
 266
 267 CREATE TRIGGER facet_force_nfc_tgr
 268         BEFORE UPDATE OR INSERT ON metabib.facet_entry
 269         FOR EACH ROW EXECUTE PROCEDURE evergreen.facet_force_nfc();
 270
 271 -- DECREMENTING serial starts at -1
 272 CREATE SEQUENCE metabib.uncontrolled_record_attr_value_id_seq INCREMENT BY -1;
 273
 274 CREATE TABLE metabib.uncontrolled_record_attr_value (
 275     id      BIGINT  PRIMARY KEY DEFAULT nextval('metabib.uncontrolled_record_attr_value_id_seq'),
 276     attr    TEXT    NOT NULL REFERENCES config.record_attr_definition (name),
 277     value   TEXT    NOT NULL
 278 );
 279 CREATE UNIQUE INDEX muv_once_idx ON metabib.uncontrolled_record_attr_value (attr,value);
 280
 281 CREATE VIEW metabib.record_attr_id_map AS
 282     SELECT id, attr, value FROM metabib.uncontrolled_record_attr_value
 283         UNION
 284     SELECT  c.id, c.ctype AS attr, c.code AS value
 285       FROM  config.coded_value_map c
 286             JOIN config.record_attr_definition d ON (d.name = c.ctype AND NOT d.composite);
 287
 288 CREATE VIEW metabib.composite_attr_id_map AS
 289     SELECT  c.id, c.ctype AS attr, c.code AS value
 290       FROM  config.coded_value_map c
 291             JOIN config.record_attr_definition d ON (d.name = c.ctype AND d.composite);
 292
 293 CREATE VIEW metabib.full_attr_id_map AS
 294     SELECT id, attr, value FROM metabib.record_attr_id_map
 295         UNION
 296     SELECT id, attr, value FROM metabib.composite_attr_id_map;
 297
 298
 299 CREATE OR REPLACE FUNCTION metabib.compile_composite_attr_cache_init () RETURNS BOOL AS $f$
 300     $_SHARED{metabib_compile_composite_attr_cache} = {}
 301         if ! exists $_SHARED{metabib_compile_composite_attr_cache};
 302     return exists $_SHARED{metabib_compile_composite_attr_cache};
 303 $f$ LANGUAGE PLPERLU;
 304
 305 CREATE OR REPLACE FUNCTION metabib.compile_composite_attr_cache_disable () RETURNS BOOL AS $f$
 306     delete $_SHARED{metabib_compile_composite_attr_cache};
 307     return ! exists $_SHARED{metabib_compile_composite_attr_cache};
 308 $f$ LANGUAGE PLPERLU;
 309
 310 CREATE OR REPLACE FUNCTION metabib.compile_composite_attr_cache_invalidate () RETURNS BOOL AS $f$
 311     SELECT metabib.compile_composite_attr_cache_disable() AND metabib.compile_composite_attr_cache_init();
 312 $f$ LANGUAGE SQL;
 313
 314 CREATE OR REPLACE FUNCTION metabib.composite_attr_def_cache_inval_tgr () RETURNS TRIGGER AS $f$
 315 BEGIN
 316     PERFORM metabib.compile_composite_attr_cache_invalidate();
 317     RETURN NULL;
 318 END;
 319 $f$ LANGUAGE PLPGSQL;
 320
 321 CREATE TRIGGER ccraed_cache_inval_tgr AFTER INSERT OR UPDATE OR DELETE ON config.composite_attr_entry_definition FOR EACH STATEMENT EXECUTE PROCEDURE metabib.composite_attr_def_cache_inval_tgr();
 322
 323 CREATE OR REPLACE FUNCTION metabib.compile_composite_attr ( cattr_def TEXT ) RETURNS query_int AS $func$
 324
 325     use JSON::XS;
 326
 327     my $json = shift;
 328     my $def = decode_json($json);
 329
 330     die("Composite attribute definition not supplied") unless $def;
 331
 332     my $_cache = (exists $_SHARED{metabib_compile_composite_attr_cache}) ? 1 : 0;
 333
 334     return $_SHARED{metabib_compile_composite_attr_cache}{$json}
 335         if ($_cache && $_SHARED{metabib_compile_composite_attr_cache}{$json});
 336
 337     sub recurse {
 338         my $d = shift;
 339         my $j = '&';
 340         my @list;
 341
 342         if (ref $d eq 'HASH') { # node or AND
 343             if (exists $d->{_attr}) { # it is a node
 344                 my $plan = spi_prepare('SELECT * FROM metabib.full_attr_id_map WHERE attr = $1 AND value = $2', qw/TEXT TEXT/);
 345                 my $id = spi_exec_prepared(
 346                     $plan, {limit => 1}, $d->{_attr}, $d->{_val}
 347                 )->{rows}[0]{id};
 348                 spi_freeplan($plan);
 349                 return $id;
 350             } elsif (exists $d->{_not} && scalar(keys(%$d)) == 1) { # it is a NOT
 351                 return '!' . recurse($$d{_not});
 352             } else { # an AND list
 353                 @list = map { recurse($$d{$_}) } sort keys %$d;
 354             }
 355         } elsif (ref $d eq 'ARRAY') {
 356             $j = '|';
 357             @list = map { recurse($_) } @$d;
 358         }
 359
 360         @list = grep { defined && $_ ne '' } @list;
 361
 362         return '(' . join($j,@list) . ')' if @list;
 363         return '';
 364     }
 365
 366     my $val = recurse($def) || undef;
 367     $_SHARED{metabib_compile_composite_attr_cache}{$json} = $val if $_cache;
 368     return $val;
 369
 370 $func$ IMMUTABLE LANGUAGE plperlu;
 371
 372 CREATE OR REPLACE FUNCTION metabib.compile_composite_attr ( cattr_id INT ) RETURNS query_int AS $func$
 373     SELECT metabib.compile_composite_attr(definition) FROM config.composite_attr_entry_definition WHERE coded_value = $1;
 374 $func$ STRICT IMMUTABLE LANGUAGE SQL;
 375
 376 CREATE TABLE metabib.record_attr_vector_list (
 377     source  BIGINT  PRIMARY KEY REFERENCES biblio.record_entry (id),
 378     vlist   INT[]   NOT NULL -- stores id from ccvm AND murav
 379 );
 380 CREATE INDEX mrca_vlist_idx ON metabib.record_attr_vector_list USING gin ( vlist gin__int_ops );
 381
 382 /* This becomes a view, and we do sorters differently ...
 383 CREATE TABLE metabib.record_attr (
 384         id              BIGINT  PRIMARY KEY REFERENCES biblio.record_entry (id) ON DELETE CASCADE,
 385         attrs   HSTORE  NOT NULL DEFAULT ''::HSTORE
 386 );
 387 CREATE INDEX metabib_svf_attrs_idx ON metabib.record_attr USING GIST (attrs);
 388 CREATE INDEX metabib_svf_date1_idx ON metabib.record_attr ((attrs->'date1'));
 389 CREATE INDEX metabib_svf_dates_idx ON metabib.record_attr ((attrs->'date1'),(attrs->'date2'));
 390 */
 391
 392 /* ... like this */
 393 CREATE TABLE metabib.record_sorter (
 394     id      BIGSERIAL   PRIMARY KEY,
 395     source  BIGINT      NOT NULL REFERENCES biblio.record_entry (id) ON DELETE CASCADE,
 396     attr    TEXT        NOT NULL REFERENCES config.record_attr_definition (name) ON DELETE CASCADE,
 397     value   TEXT        NOT NULL
 398 );
 399 CREATE INDEX metabib_sorter_source_idx ON metabib.record_sorter (source); -- we may not need one of this or the next ... stats will tell
 400 CREATE INDEX metabib_sorter_s_a_idx ON metabib.record_sorter (source, attr);
 401 CREATE INDEX metabib_sorter_a_v_idx ON metabib.record_sorter (attr, value);
 402
 403
 404 CREATE TYPE metabib.record_attr_type AS (
 405     id      BIGINT,
 406     attrs   HSTORE
 407 );
 408
 409 -- Back-compat view ... we're moving to an INTARRAY world
 410 CREATE VIEW metabib.record_attr_flat AS
 411     SELECT  v.source AS id,
 412             m.attr,
 413             m.value
 414       FROM  metabib.full_attr_id_map m
 415             JOIN  metabib.record_attr_vector_list v ON ( m.id = ANY( v.vlist ) );
 416
 417 CREATE VIEW metabib.record_attr AS
 418     SELECT id, HSTORE( ARRAY_AGG( attr ), ARRAY_AGG( value ) ) AS attrs FROM metabib.record_attr_flat GROUP BY 1;
 419
 420 -- Back-back-compat view ... we use to live in an HSTORE world
 421 CREATE TYPE metabib.rec_desc_type AS (
 422     item_type       TEXT,
 423     item_form       TEXT,
 424     bib_level       TEXT,
 425     control_type    TEXT,
 426     char_encoding   TEXT,
 427     enc_level       TEXT,
 428     audience        TEXT,
 429     lit_form        TEXT,
 430     type_mat        TEXT,
 431     cat_form        TEXT,
 432     pub_status      TEXT,
 433     item_lang       TEXT,
 434     vr_format       TEXT,
 435     date1           TEXT,
 436     date2           TEXT
 437 );
 438
 439 CREATE VIEW metabib.rec_descriptor AS
 440     SELECT  id,
 441             id AS record,
 442             (populate_record(NULL::metabib.rec_desc_type, attrs)).*
 443       FROM  metabib.record_attr;
 444
 445 -- Use a sequence that matches previous version, for easier upgrading.
 446 CREATE SEQUENCE metabib.full_rec_id_seq;
 447
 448 CREATE TABLE metabib.real_full_rec (
 449         id                  BIGINT      NOT NULL DEFAULT NEXTVAL('metabib.full_rec_id_seq'::REGCLASS),
 450         record          BIGINT          NOT NULL,
 451         tag             CHAR(3)         NOT NULL,
 452         ind1            TEXT,
 453         ind2            TEXT,
 454         subfield        TEXT,
 455         value           TEXT            NOT NULL,
 456         index_vector    tsvector        NOT NULL
 457 );
 458 ALTER TABLE metabib.real_full_rec ADD PRIMARY KEY (id);
 459
 460 CREATE INDEX metabib_full_rec_tag_subfield_idx ON metabib.real_full_rec (tag,subfield);
 461 CREATE INDEX metabib_full_rec_value_idx ON metabib.real_full_rec (substring(value,1,1024));
 462 /* Enable LIKE to use an index for database clusters with locales other than C or POSIX */
 463 CREATE INDEX metabib_full_rec_value_tpo_index ON metabib.real_full_rec (substring(value,1,1024) text_pattern_ops);
 464 CREATE INDEX metabib_full_rec_record_idx ON metabib.real_full_rec (record);
 465 CREATE INDEX metabib_full_rec_index_vector_idx ON metabib.real_full_rec USING GIST (index_vector);
 466 CREATE INDEX metabib_full_rec_isxn_caseless_idx
 467     ON metabib.real_full_rec (LOWER(value))
 468     WHERE tag IN ('020', '022', '024');
 469 -- This next index might fully supplant the one above, but leaving both for now.
 470 -- (they are not too large)
 471 -- The reason we need this index is to ensure that the query parser always
 472 -- prefers this index over the simpler tag/subfield index, as this greatly
 473 -- increases Vandelay overlay speed for these identifiers, especially when
 474 -- a record has many of these fields (around > 4-6 seems like the cutoff
 475 -- on at least one PG9.1 system)
 476 -- A similar index could be added for other fields (e.g. 010), but one should
 477 -- leave out the LOWER() in all other cases.
 478 -- TODO: verify whether we can discard the non tag/subfield/substring version
 479 -- above (metabib_full_rec_isxn_caseless_idx)
 480 CREATE INDEX metabib_full_rec_02x_tag_subfield_lower_substring
 481     ON metabib.real_full_rec (tag, subfield, LOWER(substring(value, 1, 1024)))
 482     WHERE tag IN ('020', '022', '024');
 483
 484
 485 CREATE TRIGGER metabib_full_rec_fti_trigger
 486         BEFORE UPDATE OR INSERT ON metabib.real_full_rec
 487         FOR EACH ROW EXECUTE PROCEDURE oils_tsearch2('default');
 488
 489 CREATE OR REPLACE VIEW metabib.full_rec AS
 490     SELECT  id,
 491             record,
 492             tag,
 493             ind1,
 494             ind2,
 495             subfield,
 496             SUBSTRING(value,1,1024) AS value,
 497             index_vector
 498       FROM  metabib.real_full_rec;
 499
 500 CREATE OR REPLACE RULE metabib_full_rec_insert_rule
 501     AS ON INSERT TO metabib.full_rec
 502     DO INSTEAD
 503     INSERT INTO metabib.real_full_rec VALUES (
 504         COALESCE(NEW.id, NEXTVAL('metabib.full_rec_id_seq'::REGCLASS)),
 505         NEW.record,
 506         NEW.tag,
 507         NEW.ind1,
 508         NEW.ind2,
 509         NEW.subfield,
 510         NEW.value,
 511         NEW.index_vector
 512     );
 513
 514 CREATE OR REPLACE RULE metabib_full_rec_update_rule
 515     AS ON UPDATE TO metabib.full_rec
 516     DO INSTEAD
 517     UPDATE  metabib.real_full_rec SET
 518         id = NEW.id,
 519         record = NEW.record,
 520         tag = NEW.tag,
 521         ind1 = NEW.ind1,
 522         ind2 = NEW.ind2,
 523         subfield = NEW.subfield,
 524         value = NEW.value,
 525         index_vector = NEW.index_vector
 526       WHERE id = OLD.id;
 527
 528 CREATE OR REPLACE RULE metabib_full_rec_delete_rule
 529     AS ON DELETE TO metabib.full_rec
 530     DO INSTEAD
 531     DELETE FROM metabib.real_full_rec WHERE id = OLD.id;
 532
 533 CREATE TABLE metabib.metarecord_source_map (
 534         id              BIGSERIAL       PRIMARY KEY,
 535         metarecord      BIGINT          NOT NULL,
 536         source          BIGINT          NOT NULL
 537 );
 538 CREATE INDEX metabib_metarecord_source_map_metarecord_idx ON metabib.metarecord_source_map (metarecord);
 539 CREATE INDEX metabib_metarecord_source_map_source_record_idx ON metabib.metarecord_source_map (source);
 540
 541 CREATE TYPE metabib.field_entry_template AS (
 542     field_class         TEXT,
 543     field               INT,
 544     facet_field         BOOL,
 545     search_field        BOOL,
 546     browse_field        BOOL,
 547     source              BIGINT,
 548     value               TEXT,
 549     authority           BIGINT,
 550     sort_value          TEXT
 551 );
 552
 553
 554 CREATE OR REPLACE FUNCTION biblio.extract_metabib_field_entry ( rid BIGINT, default_joiner TEXT ) RETURNS SETOF metabib.field_entry_template AS $func$
 555 DECLARE
 556     bib     biblio.record_entry%ROWTYPE;
 557     idx     config.metabib_field%ROWTYPE;
 558     xfrm        config.xml_transform%ROWTYPE;
 559     prev_xfrm   TEXT;
 560     transformed_xml TEXT;
 561     xml_node    TEXT;
 562     xml_node_list   TEXT[];
 563     facet_text  TEXT;
 564     browse_text TEXT;
 565     sort_value  TEXT;
 566     raw_text    TEXT;
 567     curr_text   TEXT;
 568     joiner      TEXT := default_joiner; -- XXX will index defs supply a joiner?
 569     authority_text TEXT;
 570     authority_link BIGINT;
 571     output_row  metabib.field_entry_template%ROWTYPE;
 572 BEGIN
 573
 574     -- Start out with no field-use bools set
 575     output_row.browse_field = FALSE;
 576     output_row.facet_field = FALSE;
 577     output_row.search_field = FALSE;
 578
 579     -- Get the record
 580     SELECT INTO bib * FROM biblio.record_entry WHERE id = rid;
 581
 582     -- Loop over the indexing entries
 583     FOR idx IN SELECT * FROM config.metabib_field ORDER BY format LOOP
 584
 585         joiner := COALESCE(idx.joiner, default_joiner);
 586
 587         SELECT INTO xfrm * from config.xml_transform WHERE name = idx.format;
 588
 589         -- See if we can skip the XSLT ... it's expensive
 590         IF prev_xfrm IS NULL OR prev_xfrm <> xfrm.name THEN
 591             -- Can't skip the transform
 592             IF xfrm.xslt <> '---' THEN
 593                 transformed_xml := oils_xslt_process(bib.marc,xfrm.xslt);
 594             ELSE
 595                 transformed_xml := bib.marc;
 596             END IF;
 597
 598             prev_xfrm := xfrm.name;
 599         END IF;
 600
 601         xml_node_list := oils_xpath( idx.xpath, transformed_xml, ARRAY[ARRAY[xfrm.prefix, xfrm.namespace_uri]] );
 602
 603         raw_text := NULL;
 604         FOR xml_node IN SELECT x FROM unnest(xml_node_list) AS x LOOP
 605             CONTINUE WHEN xml_node !~ E'^\\s*<';
 606
 607             -- XXX much of this should be moved into oils_xpath_string...
 608             curr_text := ARRAY_TO_STRING(evergreen.array_remove_item_by_value(evergreen.array_remove_item_by_value(
 609                 oils_xpath( '//text()',
 610                     REGEXP_REPLACE(
 611                         REGEXP_REPLACE( -- This escapes all &s not followed by "amp;".  Data ise returned from oils_xpath (above) in UTF-8, not entity encoded
 612                             REGEXP_REPLACE( -- This escapes embeded <s
 613                                 xml_node,
 614                                 $re$(>[^<]+)(<)([^>]+<)$re$,
 615                                 E'\\1&lt;\\3',
 616                                 'g'
 617                             ),
 618                             '&(?!amp;)',
 619                             '&amp;',
 620                             'g'
 621                         ),
 622                         E'\\s+',
 623                         ' ',
 624                         'g'
 625                     )
 626                 ), ' '), ''),
 627                 joiner
 628             );
 629
 630             CONTINUE WHEN curr_text IS NULL OR curr_text = '';
 631
 632             IF raw_text IS NOT NULL THEN
 633                 raw_text := raw_text || joiner;
 634             END IF;
 635
 636             raw_text := COALESCE(raw_text,'') || curr_text;
 637
 638             -- autosuggest/metabib.browse_entry
 639             IF idx.browse_field THEN
 640
 641                 IF idx.browse_xpath IS NOT NULL AND idx.browse_xpath <> '' THEN
 642                     browse_text := oils_xpath_string( idx.browse_xpath, xml_node, joiner, ARRAY[ARRAY[xfrm.prefix, xfrm.namespace_uri]] );
 643                 ELSE
 644                     browse_text := curr_text;
 645                 END IF;
 646
 647                 IF idx.browse_sort_xpath IS NOT NULL AND
 648                     idx.browse_sort_xpath <> '' THEN
 649
 650                     sort_value := oils_xpath_string(
 651                         idx.browse_sort_xpath, xml_node, joiner,
 652                         ARRAY[ARRAY[xfrm.prefix, xfrm.namespace_uri]]
 653                     );
 654                 ELSE
 655                     sort_value := browse_text;
 656                 END IF;
 657
 658                 output_row.field_class = idx.field_class;
 659                 output_row.field = idx.id;
 660                 output_row.source = rid;
 661                 output_row.value = BTRIM(REGEXP_REPLACE(browse_text, E'\\s+', ' ', 'g'));
 662                 output_row.sort_value :=
 663                     public.naco_normalize(sort_value);
 664
 665                 output_row.authority := NULL;
 666
 667                 IF idx.authority_xpath IS NOT NULL AND idx.authority_xpath <> '' THEN
 668                     authority_text := oils_xpath_string(
 669                         idx.authority_xpath, xml_node, joiner,
 670                         ARRAY[
 671                             ARRAY[xfrm.prefix, xfrm.namespace_uri],
 672                             ARRAY['xlink','http://www.w3.org/1999/xlink']
 673                         ]
 674                     );
 675
 676                     IF authority_text ~ '^\d+$' THEN
 677                         authority_link := authority_text::BIGINT;
 678                         PERFORM * FROM authority.record_entry WHERE id = authority_link;
 679                         IF FOUND THEN
 680                             output_row.authority := authority_link;
 681                         END IF;
 682                     END IF;
 683
 684                 END IF;
 685
 686                 output_row.browse_field = TRUE;
 687                 -- Returning browse rows with search_field = true for search+browse
 688                 -- configs allows us to retain granularity of being able to search
 689                 -- browse fields with "starts with" type operators (for example, for
 690                 -- titles of songs in music albums)
 691                 IF idx.search_field THEN
 692                     output_row.search_field = TRUE;
 693                 END IF;
 694                 RETURN NEXT output_row;
 695                 output_row.browse_field = FALSE;
 696                 output_row.search_field = FALSE;
 697                 output_row.sort_value := NULL;
 698             END IF;
 699
 700             -- insert raw node text for faceting
 701             IF idx.facet_field THEN
 702
 703                 IF idx.facet_xpath IS NOT NULL AND idx.facet_xpath <> '' THEN
 704                     facet_text := oils_xpath_string( idx.facet_xpath, xml_node, joiner, ARRAY[ARRAY[xfrm.prefix, xfrm.namespace_uri]] );
 705                 ELSE
 706                     facet_text := curr_text;
 707                 END IF;
 708
 709                 output_row.field_class = idx.field_class;
 710                 output_row.field = -1 * idx.id;
 711                 output_row.source = rid;
 712                 output_row.value = BTRIM(REGEXP_REPLACE(facet_text, E'\\s+', ' ', 'g'));
 713
 714                 output_row.facet_field = TRUE;
 715                 RETURN NEXT output_row;
 716                 output_row.facet_field = FALSE;
 717             END IF;
 718
 719         END LOOP;
 720
 721         CONTINUE WHEN raw_text IS NULL OR raw_text = '';
 722
 723         -- insert combined node text for searching
 724         IF idx.search_field THEN
 725             output_row.field_class = idx.field_class;
 726             output_row.field = idx.id;
 727             output_row.source = rid;
 728             output_row.value = BTRIM(REGEXP_REPLACE(raw_text, E'\\s+', ' ', 'g'));
 729
 730             output_row.search_field = TRUE;
 731             RETURN NEXT output_row;
 732             output_row.search_field = FALSE;
 733         END IF;
 734
 735     END LOOP;
 736
 737 END;
 738
 739 $func$ LANGUAGE PLPGSQL;
 740
 741 CREATE OR REPLACE FUNCTION metabib.update_combined_index_vectors(bib_id BIGINT) RETURNS VOID AS $func$
 742 BEGIN
 743     DELETE FROM metabib.combined_keyword_field_entry WHERE record = bib_id;
 744     INSERT INTO metabib.combined_keyword_field_entry(record, metabib_field, index_vector)
 745         SELECT bib_id, field, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 746         FROM metabib.keyword_field_entry WHERE source = bib_id GROUP BY field;
 747     INSERT INTO metabib.combined_keyword_field_entry(record, metabib_field, index_vector)
 748         SELECT bib_id, NULL, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 749         FROM metabib.keyword_field_entry WHERE source = bib_id;
 750
 751     DELETE FROM metabib.combined_title_field_entry WHERE record = bib_id;
 752     INSERT INTO metabib.combined_title_field_entry(record, metabib_field, index_vector)
 753         SELECT bib_id, field, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 754         FROM metabib.title_field_entry WHERE source = bib_id GROUP BY field;
 755     INSERT INTO metabib.combined_title_field_entry(record, metabib_field, index_vector)
 756         SELECT bib_id, NULL, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 757         FROM metabib.title_field_entry WHERE source = bib_id;
 758
 759     DELETE FROM metabib.combined_author_field_entry WHERE record = bib_id;
 760     INSERT INTO metabib.combined_author_field_entry(record, metabib_field, index_vector)
 761         SELECT bib_id, field, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 762         FROM metabib.author_field_entry WHERE source = bib_id GROUP BY field;
 763     INSERT INTO metabib.combined_author_field_entry(record, metabib_field, index_vector)
 764         SELECT bib_id, NULL, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 765         FROM metabib.author_field_entry WHERE source = bib_id;
 766
 767     DELETE FROM metabib.combined_subject_field_entry WHERE record = bib_id;
 768     INSERT INTO metabib.combined_subject_field_entry(record, metabib_field, index_vector)
 769         SELECT bib_id, field, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 770         FROM metabib.subject_field_entry WHERE source = bib_id GROUP BY field;
 771     INSERT INTO metabib.combined_subject_field_entry(record, metabib_field, index_vector)
 772         SELECT bib_id, NULL, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 773         FROM metabib.subject_field_entry WHERE source = bib_id;
 774
 775     DELETE FROM metabib.combined_series_field_entry WHERE record = bib_id;
 776     INSERT INTO metabib.combined_series_field_entry(record, metabib_field, index_vector)
 777         SELECT bib_id, field, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 778         FROM metabib.series_field_entry WHERE source = bib_id GROUP BY field;
 779     INSERT INTO metabib.combined_series_field_entry(record, metabib_field, index_vector)
 780         SELECT bib_id, NULL, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 781         FROM metabib.series_field_entry WHERE source = bib_id;
 782
 783     DELETE FROM metabib.combined_identifier_field_entry WHERE record = bib_id;
 784     INSERT INTO metabib.combined_identifier_field_entry(record, metabib_field, index_vector)
 785         SELECT bib_id, field, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 786         FROM metabib.identifier_field_entry WHERE source = bib_id GROUP BY field;
 787     INSERT INTO metabib.combined_identifier_field_entry(record, metabib_field, index_vector)
 788         SELECT bib_id, NULL, strip(COALESCE(string_agg(index_vector::TEXT,' '),'')::tsvector)
 789         FROM metabib.identifier_field_entry WHERE source = bib_id;
 790
 791 END;
 792 $func$ LANGUAGE PLPGSQL;
 793
 794 CREATE OR REPLACE FUNCTION metabib.reingest_metabib_field_entries( bib_id BIGINT, skip_facet BOOL DEFAULT FALSE, skip_browse BOOL DEFAULT FALSE, skip_search BOOL DEFAULT FALSE ) RETURNS VOID AS $func$
 795 DECLARE
 796     fclass          RECORD;
 797     ind_data        metabib.field_entry_template%ROWTYPE;
 798     mbe_row         metabib.browse_entry%ROWTYPE;
 799     mbe_id          BIGINT;
 800     b_skip_facet    BOOL;
 801     b_skip_browse   BOOL;
 802     b_skip_search   BOOL;
 803     value_prepped   TEXT;
 804 BEGIN
 805
 806     SELECT COALESCE(NULLIF(skip_facet, FALSE), EXISTS (SELECT enabled FROM config.internal_flag WHERE name =  'ingest.skip_facet_indexing' AND enabled)) INTO b_skip_facet;
 807     SELECT COALESCE(NULLIF(skip_browse, FALSE), EXISTS (SELECT enabled FROM config.internal_flag WHERE name =  'ingest.skip_browse_indexing' AND enabled)) INTO b_skip_browse;
 808     SELECT COALESCE(NULLIF(skip_search, FALSE), EXISTS (SELECT enabled FROM config.internal_flag WHERE name =  'ingest.skip_search_indexing' AND enabled)) INTO b_skip_search;
 809
 810     PERFORM * FROM config.internal_flag WHERE name = 'ingest.assume_inserts_only' AND enabled;
 811     IF NOT FOUND THEN
 812         IF NOT b_skip_search THEN
 813             FOR fclass IN SELECT * FROM config.metabib_class LOOP
 814                 -- RAISE NOTICE 'Emptying out %', fclass.name;
 815                 EXECUTE $$DELETE FROM metabib.$$ || fclass.name || $$_field_entry WHERE source = $$ || bib_id;
 816             END LOOP;
 817         END IF;
 818         IF NOT b_skip_facet THEN
 819             DELETE FROM metabib.facet_entry WHERE source = bib_id;
 820         END IF;
 821         IF NOT b_skip_browse THEN
 822             DELETE FROM metabib.browse_entry_def_map WHERE source = bib_id;
 823         END IF;
 824     END IF;
 825
 826     FOR ind_data IN SELECT * FROM biblio.extract_metabib_field_entry( bib_id ) LOOP
 827         IF ind_data.field < 0 THEN
 828             ind_data.field = -1 * ind_data.field;
 829         END IF;
 830
 831         IF ind_data.facet_field AND NOT b_skip_facet THEN
 832             INSERT INTO metabib.facet_entry (field, source, value)
 833                 VALUES (ind_data.field, ind_data.source, ind_data.value);
 834         END IF;
 835
 836         IF ind_data.browse_field AND NOT b_skip_browse THEN
 837             -- A caveat about this SELECT: this should take care of replacing
 838             -- old mbe rows when data changes, but not if normalization (by
 839             -- which I mean specifically the output of
 840             -- evergreen.oils_tsearch2()) changes.  It may or may not be
 841             -- expensive to add a comparison of index_vector to index_vector
 842             -- to the WHERE clause below.
 843
 844             value_prepped := metabib.browse_normalize(ind_data.value, ind_data.field);
 845             SELECT INTO mbe_row * FROM metabib.browse_entry
 846                 WHERE value = value_prepped AND sort_value = ind_data.sort_value;
 847
 848             IF FOUND THEN
 849                 mbe_id := mbe_row.id;
 850             ELSE
 851                 INSERT INTO metabib.browse_entry
 852                     ( value, sort_value ) VALUES
 853                     ( value_prepped, ind_data.sort_value );
 854
 855                 mbe_id := CURRVAL('metabib.browse_entry_id_seq'::REGCLASS);
 856             END IF;
 857
 858             INSERT INTO metabib.browse_entry_def_map (entry, def, source, authority)
 859                 VALUES (mbe_id, ind_data.field, ind_data.source, ind_data.authority);
 860         END IF;
 861
 862         IF ind_data.search_field AND NOT b_skip_search THEN
 863             -- Avoid inserting duplicate rows
 864             EXECUTE 'SELECT 1 FROM metabib.' || ind_data.field_class ||
 865                 '_field_entry WHERE field = $1 AND source = $2 AND value = $3'
 866                 INTO mbe_id USING ind_data.field, ind_data.source, ind_data.value;
 867                 -- RAISE NOTICE 'Search for an already matching row returned %', mbe_id;
 868             IF mbe_id IS NULL THEN
 869                 EXECUTE $$
 870                 INSERT INTO metabib.$$ || ind_data.field_class || $$_field_entry (field, source, value)
 871                     VALUES ($$ ||
 872                         quote_literal(ind_data.field) || $$, $$ ||
 873                         quote_literal(ind_data.source) || $$, $$ ||
 874                         quote_literal(ind_data.value) ||
 875                     $$);$$;
 876             END IF;
 877         END IF;
 878
 879     END LOOP;
 880
 881     IF NOT b_skip_search THEN
 882         PERFORM metabib.update_combined_index_vectors(bib_id);
 883     END IF;
 884
 885     RETURN;
 886 END;
 887 $func$ LANGUAGE PLPGSQL;
 888
 889 -- default to a space joiner
 890 CREATE OR REPLACE FUNCTION biblio.extract_metabib_field_entry ( BIGINT ) RETURNS SETOF metabib.field_entry_template AS $func$
 891         SELECT * FROM biblio.extract_metabib_field_entry($1, ' ');
 892 $func$ LANGUAGE SQL;
 893
 894 CREATE OR REPLACE FUNCTION authority.flatten_marc ( rid BIGINT ) RETURNS SETOF authority.full_rec AS $func$
 895 DECLARE
 896         auth    authority.record_entry%ROWTYPE;
 897         output  authority.full_rec%ROWTYPE;
 898         field   RECORD;
 899 BEGIN
 900         SELECT INTO auth * FROM authority.record_entry WHERE id = rid;
 901
 902         FOR field IN SELECT * FROM vandelay.flatten_marc( auth.marc ) LOOP
 903                 output.record := rid;
 904                 output.ind1 := field.ind1;
 905                 output.ind2 := field.ind2;
 906                 output.tag := field.tag;
 907                 output.subfield := field.subfield;
 908                 output.value := field.value;
 909
 910                 RETURN NEXT output;
 911         END LOOP;
 912 END;
 913 $func$ LANGUAGE PLPGSQL;
 914
 915 CREATE OR REPLACE FUNCTION biblio.flatten_marc ( rid BIGINT ) RETURNS SETOF metabib.full_rec AS $func$
 916 DECLARE
 917         bib     biblio.record_entry%ROWTYPE;
 918         output  metabib.full_rec%ROWTYPE;
 919         field   RECORD;
 920 BEGIN
 921         SELECT INTO bib * FROM biblio.record_entry WHERE id = rid;
 922
 923         FOR field IN SELECT * FROM vandelay.flatten_marc( bib.marc ) LOOP
 924                 output.record := rid;
 925                 output.ind1 := field.ind1;
 926                 output.ind2 := field.ind2;
 927                 output.tag := field.tag;
 928                 output.subfield := field.subfield;
 929                 output.value := field.value;
 930
 931                 RETURN NEXT output;
 932         END LOOP;
 933 END;
 934 $func$ LANGUAGE PLPGSQL;
 935
 936 CREATE OR REPLACE FUNCTION vandelay.marc21_record_type( marc TEXT ) RETURNS config.marc21_rec_type_map AS $func$
 937 DECLARE
 938         ldr         TEXT;
 939         tval        TEXT;
 940         tval_rec    RECORD;
 941         bval        TEXT;
 942         bval_rec    RECORD;
 943     retval      config.marc21_rec_type_map%ROWTYPE;
 944 BEGIN
 945     ldr := oils_xpath_string( '//*[local-name()="leader"]', marc );
 946
 947     IF ldr IS NULL OR ldr = '' THEN
 948         SELECT * INTO retval FROM config.marc21_rec_type_map WHERE code = 'BKS';
 949         RETURN retval;
 950     END IF;
 951
 952     SELECT * INTO tval_rec FROM config.marc21_ff_pos_map WHERE fixed_field = 'Type' LIMIT 1; -- They're all the same
 953     SELECT * INTO bval_rec FROM config.marc21_ff_pos_map WHERE fixed_field = 'BLvl' LIMIT 1; -- They're all the same
 954
 955
 956     tval := SUBSTRING( ldr, tval_rec.start_pos + 1, tval_rec.length );
 957     bval := SUBSTRING( ldr, bval_rec.start_pos + 1, bval_rec.length );
 958
 959     -- RAISE NOTICE 'type %, blvl %, ldr %', tval, bval, ldr;
 960
 961     SELECT * INTO retval FROM config.marc21_rec_type_map WHERE type_val LIKE '%' || tval || '%' AND blvl_val LIKE '%' || bval || '%';
 962
 963
 964     IF retval.code IS NULL THEN
 965         SELECT * INTO retval FROM config.marc21_rec_type_map WHERE code = 'BKS';
 966     END IF;
 967
 968     RETURN retval;
 969 END;
 970 $func$ LANGUAGE PLPGSQL;
 971
 972 CREATE OR REPLACE FUNCTION biblio.marc21_record_type( rid BIGINT ) RETURNS config.marc21_rec_type_map AS $func$
 973     SELECT * FROM vandelay.marc21_record_type( (SELECT marc FROM biblio.record_entry WHERE id = $1) );
 974 $func$ LANGUAGE SQL;
 975
 976 CREATE OR REPLACE FUNCTION vandelay.marc21_extract_fixed_field_list( marc TEXT, ff TEXT ) RETURNS TEXT[] AS $func$
 977 DECLARE
 978     rtype       TEXT;
 979     ff_pos      RECORD;
 980     tag_data    RECORD;
 981     val         TEXT;
 982     collection  TEXT[] := '{}'::TEXT[];
 983 BEGIN
 984     rtype := (vandelay.marc21_record_type( marc )).code;
 985     FOR ff_pos IN SELECT * FROM config.marc21_ff_pos_map WHERE fixed_field = ff AND rec_type = rtype ORDER BY tag DESC LOOP
 986         IF ff_pos.tag = 'ldr' THEN
 987             val := oils_xpath_string('//*[local-name()="leader"]', marc);
 988             IF val IS NOT NULL THEN
 989                 val := SUBSTRING( val, ff_pos.start_pos + 1, ff_pos.length );
 990                 collection := collection || val;
 991             END IF;
 992         ELSE
 993             FOR tag_data IN SELECT value FROM UNNEST( oils_xpath( '//*[@tag="' || UPPER(ff_pos.tag) || '"]/text()', marc ) ) x(value) LOOP
 994                 val := SUBSTRING( tag_data.value, ff_pos.start_pos + 1, ff_pos.length );
 995                 collection := collection || val;
 996             END LOOP;
 997         END IF;
 998         val := REPEAT( ff_pos.default_val, ff_pos.length );
 999         collection := collection || val;
1000     END LOOP;
1001
1002     RETURN collection;
1003 END;
1004 $func$ LANGUAGE PLPGSQL;
1005
1006 CREATE OR REPLACE FUNCTION vandelay.marc21_extract_fixed_field( marc TEXT, ff TEXT ) RETURNS TEXT AS $func$
1007 DECLARE
1008     rtype       TEXT;
1009     ff_pos      RECORD;
1010     tag_data    RECORD;
1011     val         TEXT;
1012 BEGIN
1013     rtype := (vandelay.marc21_record_type( marc )).code;
1014     FOR ff_pos IN SELECT * FROM config.marc21_ff_pos_map WHERE fixed_field = ff AND rec_type = rtype ORDER BY tag DESC LOOP
1015         IF ff_pos.tag = 'ldr' THEN
1016             val := oils_xpath_string('//*[local-name()="leader"]', marc);
1017             IF val IS NOT NULL THEN
1018                 val := SUBSTRING( val, ff_pos.start_pos + 1, ff_pos.length );
1019                 RETURN val;
1020             END IF;
1021         ELSE
1022             FOR tag_data IN SELECT value FROM UNNEST( oils_xpath( '//*[@tag="' || UPPER(ff_pos.tag) || '"]/text()', marc ) ) x(value) LOOP
1023                 val := SUBSTRING( tag_data.value, ff_pos.start_pos + 1, ff_pos.length );
1024                 RETURN val;
1025             END LOOP;
1026         END IF;
1027         val := REPEAT( ff_pos.default_val, ff_pos.length );
1028         RETURN val;
1029     END LOOP;
1030
1031     RETURN NULL;
1032 END;
1033 $func$ LANGUAGE PLPGSQL;
1034
1035 CREATE OR REPLACE FUNCTION biblio.marc21_extract_fixed_field_list( rid BIGINT, ff TEXT ) RETURNS TEXT[] AS $func$
1036     SELECT * FROM vandelay.marc21_extract_fixed_field_list( (SELECT marc FROM biblio.record_entry WHERE id = $1), $2 );
1037 $func$ LANGUAGE SQL;
1038
1039 CREATE OR REPLACE FUNCTION biblio.marc21_extract_fixed_field( rid BIGINT, ff TEXT ) RETURNS TEXT AS $func$
1040     SELECT * FROM vandelay.marc21_extract_fixed_field( (SELECT marc FROM biblio.record_entry WHERE id = $1), $2 );
1041 $func$ LANGUAGE SQL;
1042
1043 -- CREATE TYPE biblio.record_ff_map AS (record BIGINT, ff_name TEXT, ff_value TEXT);
1044 CREATE OR REPLACE FUNCTION vandelay.marc21_extract_all_fixed_fields( marc TEXT ) RETURNS SETOF biblio.record_ff_map AS $func$
1045 DECLARE
1046     tag_data    TEXT;
1047     rtype       TEXT;
1048     ff_pos      RECORD;
1049     output      biblio.record_ff_map%ROWTYPE;
1050 BEGIN
1051     rtype := (vandelay.marc21_record_type( marc )).code;
1052
1053     FOR ff_pos IN SELECT * FROM config.marc21_ff_pos_map WHERE rec_type = rtype ORDER BY tag DESC LOOP
1054         output.ff_name  := ff_pos.fixed_field;
1055         output.ff_value := NULL;
1056
1057         IF ff_pos.tag = 'ldr' THEN
1058             output.ff_value := oils_xpath_string('//*[local-name()="leader"]', marc);
1059             IF output.ff_value IS NOT NULL THEN
1060                 output.ff_value := SUBSTRING( output.ff_value, ff_pos.start_pos + 1, ff_pos.length );
1061                 RETURN NEXT output;
1062                 output.ff_value := NULL;
1063             END IF;
1064         ELSE
1065             FOR tag_data IN SELECT value FROM UNNEST( oils_xpath( '//*[@tag="' || UPPER(ff_pos.tag) || '"]/text()', marc ) ) x(value) LOOP
1066                 output.ff_value := SUBSTRING( tag_data, ff_pos.start_pos + 1, ff_pos.length );
1067                 IF output.ff_value IS NULL THEN output.ff_value := REPEAT( ff_pos.default_val, ff_pos.length ); END IF;
1068                 RETURN NEXT output;
1069                 output.ff_value := NULL;
1070             END LOOP;
1071         END IF;
1072
1073     END LOOP;
1074
1075     RETURN;
1076 END;
1077 $func$ LANGUAGE PLPGSQL;
1078
1079 CREATE OR REPLACE FUNCTION biblio.marc21_extract_all_fixed_fields( rid BIGINT ) RETURNS SETOF biblio.record_ff_map AS $func$
1080     SELECT $1 AS record, ff_name, ff_value FROM vandelay.marc21_extract_all_fixed_fields( (SELECT marc FROM biblio.record_entry WHERE id = $1) );
1081 $func$ LANGUAGE SQL;
1082
1083 CREATE OR REPLACE FUNCTION biblio.marc21_physical_characteristics( rid BIGINT ) RETURNS SETOF biblio.marc21_physical_characteristics AS $func$
1084     SELECT id, $1 AS record, ptype, subfield, value FROM vandelay.marc21_physical_characteristics( (SELECT marc FROM biblio.record_entry WHERE id = $1) );
1085 $func$ LANGUAGE SQL;
1086
1087 CREATE OR REPLACE FUNCTION biblio.extract_quality ( marc TEXT, best_lang TEXT, best_type TEXT ) RETURNS INT AS $func$
1088 DECLARE
1089     qual        INT;
1090     ldr         TEXT;
1091     tval        TEXT;
1092     tval_rec    RECORD;
1093     bval        TEXT;
1094     bval_rec    RECORD;
1095     type_map    RECORD;
1096     ff_pos      RECORD;
1097     ff_tag_data TEXT;
1098 BEGIN
1099
1100     IF marc IS NULL OR marc = '' THEN
1101         RETURN NULL;
1102     END IF;
1103
1104     -- First, the count of tags
1105     qual := ARRAY_UPPER(oils_xpath('*[local-name()="datafield"]', marc), 1);
1106
1107     -- now go through a bunch of pain to get the record type
1108     IF best_type IS NOT NULL THEN
1109         ldr := (oils_xpath('//*[local-name()="leader"]/text()', marc))[1];
1110
1111         IF ldr IS NOT NULL THEN
1112             SELECT * INTO tval_rec FROM config.marc21_ff_pos_map WHERE fixed_field = 'Type' LIMIT 1; -- They're all the same
1113             SELECT * INTO bval_rec FROM config.marc21_ff_pos_map WHERE fixed_field = 'BLvl' LIMIT 1; -- They're all the same
1114
1115
1116             tval := SUBSTRING( ldr, tval_rec.start_pos + 1, tval_rec.length );
1117             bval := SUBSTRING( ldr, bval_rec.start_pos + 1, bval_rec.length );
1118
1119             -- RAISE NOTICE 'type %, blvl %, ldr %', tval, bval, ldr;
1120
1121             SELECT * INTO type_map FROM config.marc21_rec_type_map WHERE type_val LIKE '%' || tval || '%' AND blvl_val LIKE '%' || bval || '%';
1122
1123             IF type_map.code IS NOT NULL THEN
1124                 IF best_type = type_map.code THEN
1125                     qual := qual + qual / 2;
1126                 END IF;
1127
1128                 FOR ff_pos IN SELECT * FROM config.marc21_ff_pos_map WHERE fixed_field = 'Lang' AND rec_type = type_map.code ORDER BY tag DESC LOOP
1129                     ff_tag_data := SUBSTRING((oils_xpath('//*[@tag="' || ff_pos.tag || '"]/text()',marc))[1], ff_pos.start_pos + 1, ff_pos.length);
1130                     IF ff_tag_data = best_lang THEN
1131                             qual := qual + 100;
1132                     END IF;
1133                 END LOOP;
1134             END IF;
1135         END IF;
1136     END IF;
1137
1138     -- Now look for some quality metrics
1139     -- DCL record?
1140     IF ARRAY_UPPER(oils_xpath('//*[@tag="040"]/*[@code="a" and contains(.,"DLC")]', marc), 1) = 1 THEN
1141         qual := qual + 10;
1142     END IF;
1143
1144     -- From OCLC?
1145     IF (oils_xpath('//*[@tag="003"]/text()', marc))[1] ~* E'oclo?c' THEN
1146         qual := qual + 10;
1147     END IF;
1148
1149     RETURN qual;
1150
1151 END;
1152 $func$ LANGUAGE PLPGSQL;
1153
1154 CREATE OR REPLACE FUNCTION biblio.extract_fingerprint ( marc text ) RETURNS TEXT AS $func$
1155 DECLARE
1156         idx             config.biblio_fingerprint%ROWTYPE;
1157         xfrm            config.xml_transform%ROWTYPE;
1158         prev_xfrm       TEXT;
1159         transformed_xml TEXT;
1160         xml_node        TEXT;
1161         xml_node_list   TEXT[];
1162         raw_text        TEXT;
1163     output_text TEXT := '';
1164 BEGIN
1165
1166     IF marc IS NULL OR marc = '' THEN
1167         RETURN NULL;
1168     END IF;
1169
1170         -- Loop over the indexing entries
1171         FOR idx IN SELECT * FROM config.biblio_fingerprint ORDER BY format, id LOOP
1172
1173                 SELECT INTO xfrm * from config.xml_transform WHERE name = idx.format;
1174
1175                 -- See if we can skip the XSLT ... it's expensive
1176                 IF prev_xfrm IS NULL OR prev_xfrm <> xfrm.name THEN
1177                         -- Can't skip the transform
1178                         IF xfrm.xslt <> '---' THEN
1179                                 transformed_xml := oils_xslt_process(marc,xfrm.xslt);
1180                         ELSE
1181                                 transformed_xml := marc;
1182                         END IF;
1183
1184                         prev_xfrm := xfrm.name;
1185                 END IF;
1186
1187                 raw_text := COALESCE(
1188             naco_normalize(
1189                 ARRAY_TO_STRING(
1190                     oils_xpath(
1191                         '//text()',
1192                         (oils_xpath(
1193                             idx.xpath,
1194                             transformed_xml,
1195                             ARRAY[ARRAY[xfrm.prefix, xfrm.namespace_uri]]
1196                         ))[1]
1197                     ),
1198                     ''
1199                 )
1200             ),
1201             ''
1202         );
1203
1204         raw_text := REGEXP_REPLACE(raw_text, E'\\[.+?\\]', E'');
1205         raw_text := REGEXP_REPLACE(raw_text, E'\\mthe\\M|\\man?d?d\\M', E'', 'g'); -- arg! the pain!
1206
1207         IF idx.first_word IS TRUE THEN
1208             raw_text := REGEXP_REPLACE(raw_text, E'^(\\w+).*?$', E'\\1');
1209         END IF;
1210
1211                 output_text := output_text || REGEXP_REPLACE(raw_text, E'\\s+', '', 'g');
1212
1213         END LOOP;
1214
1215     RETURN output_text;
1216
1217 END;
1218 $func$ LANGUAGE PLPGSQL;
1219
1220 -- BEFORE UPDATE OR INSERT trigger for biblio.record_entry
1221 CREATE OR REPLACE FUNCTION biblio.fingerprint_trigger () RETURNS TRIGGER AS $func$
1222 BEGIN
1223
1224     -- For TG_ARGV, first param is language (like 'eng'), second is record type (like 'BKS')
1225
1226     IF NEW.deleted IS TRUE THEN -- we don't much care, then, do we?
1227         RETURN NEW;
1228     END IF;
1229
1230     NEW.fingerprint := biblio.extract_fingerprint(NEW.marc);
1231     NEW.quality := biblio.extract_quality(NEW.marc, TG_ARGV[0], TG_ARGV[1]);
1232
1233     RETURN NEW;
1234
1235 END;
1236 $func$ LANGUAGE PLPGSQL;
1237
1238 CREATE OR REPLACE FUNCTION metabib.reingest_metabib_full_rec( bib_id BIGINT ) RETURNS VOID AS $func$
1239 BEGIN
1240     PERFORM * FROM config.internal_flag WHERE name = 'ingest.assume_inserts_only' AND enabled;
1241     IF NOT FOUND THEN
1242         DELETE FROM metabib.real_full_rec WHERE record = bib_id;
1243     END IF;
1244     INSERT INTO metabib.real_full_rec (record, tag, ind1, ind2, subfield, value)
1245         SELECT record, tag, ind1, ind2, subfield, value FROM biblio.flatten_marc( bib_id );
1246
1247     RETURN;
1248 END;
1249 $func$ LANGUAGE PLPGSQL;
1250
1251 CREATE OR REPLACE FUNCTION biblio.extract_located_uris( bib_id BIGINT, marcxml TEXT, editor_id INT ) RETURNS VOID AS $func$
1252 DECLARE
1253     uris            TEXT[];
1254     uri_xml         TEXT;
1255     uri_label       TEXT;
1256     uri_href        TEXT;
1257     uri_use         TEXT;
1258     uri_owner_list  TEXT[];
1259     uri_owner       TEXT;
1260     uri_owner_id    INT;
1261     uri_id          INT;
1262     uri_cn_id       INT;
1263     uri_map_id      INT;
1264 BEGIN
1265
1266     -- Clear any URI mappings and call numbers for this bib.
1267     -- This leads to acn / auricnm inflation, but also enables
1268     -- old acn/auricnm's to go away and for bibs to be deleted.
1269     FOR uri_cn_id IN SELECT id FROM asset.call_number WHERE record = bib_id AND label = '##URI##' AND NOT deleted LOOP
1270         DELETE FROM asset.uri_call_number_map WHERE call_number = uri_cn_id;
1271         DELETE FROM asset.call_number WHERE id = uri_cn_id;
1272     END LOOP;
1273
1274     uris := oils_xpath('//*[@tag="856" and (@ind1="4" or @ind1="1") and (@ind2="0" or @ind2="1")]',marcxml);
1275     IF ARRAY_UPPER(uris,1) > 0 THEN
1276         FOR i IN 1 .. ARRAY_UPPER(uris, 1) LOOP
1277             -- First we pull info out of the 856
1278             uri_xml     := uris[i];
1279
1280             uri_href    := (oils_xpath('//*[@code="u"]/text()',uri_xml))[1];
1281             uri_label   := (oils_xpath('//*[@code="y"]/text()|//*[@code="3"]/text()',uri_xml))[1];
1282             uri_use     := (oils_xpath('//*[@code="z"]/text()|//*[@code="2"]/text()|//*[@code="n"]/text()',uri_xml))[1];
1283
1284             IF uri_label IS NULL THEN
1285                 uri_label := uri_href;
1286             END IF;
1287             CONTINUE WHEN uri_href IS NULL;
1288
1289             -- Get the distinct list of libraries wanting to use
1290             SELECT  ARRAY_AGG(
1291                         DISTINCT REGEXP_REPLACE(
1292                             x,
1293                             $re$^.*?\((\w+)\).*$$re$,
1294                             E'\\1'
1295                         )
1296                     ) INTO uri_owner_list
1297               FROM  UNNEST(
1298                         oils_xpath(
1299                             '//*[@code="9"]/text()|//*[@code="w"]/text()|//*[@code="n"]/text()',
1300                             uri_xml
1301                         )
1302                     )x;
1303
1304             IF ARRAY_UPPER(uri_owner_list,1) > 0 THEN
1305
1306                 -- look for a matching uri
1307                 IF uri_use IS NULL THEN
1308                     SELECT id INTO uri_id
1309                         FROM asset.uri
1310                         WHERE label = uri_label AND href = uri_href AND use_restriction IS NULL AND active
1311                         ORDER BY id LIMIT 1;
1312                     IF NOT FOUND THEN -- create one
1313                         INSERT INTO asset.uri (label, href, use_restriction) VALUES (uri_label, uri_href, uri_use);
1314                         SELECT id INTO uri_id
1315                             FROM asset.uri
1316                             WHERE label = uri_label AND href = uri_href AND use_restriction IS NULL AND active;
1317                     END IF;
1318                 ELSE
1319                     SELECT id INTO uri_id
1320                         FROM asset.uri
1321                         WHERE label = uri_label AND href = uri_href AND use_restriction = uri_use AND active
1322                         ORDER BY id LIMIT 1;
1323                     IF NOT FOUND THEN -- create one
1324                         INSERT INTO asset.uri (label, href, use_restriction) VALUES (uri_label, uri_href, uri_use);
1325                         SELECT id INTO uri_id
1326                             FROM asset.uri
1327                             WHERE label = uri_label AND href = uri_href AND use_restriction = uri_use AND active;
1328                     END IF;
1329                 END IF;
1330
1331                 FOR j IN 1 .. ARRAY_UPPER(uri_owner_list, 1) LOOP
1332                     uri_owner := uri_owner_list[j];
1333
1334                     SELECT id INTO uri_owner_id FROM actor.org_unit WHERE shortname = uri_owner;
1335                     CONTINUE WHEN NOT FOUND;
1336
1337                     -- we need a call number to link through
1338                     SELECT id INTO uri_cn_id FROM asset.call_number WHERE owning_lib = uri_owner_id AND record = bib_id AND label = '##URI##' AND NOT deleted;
1339                     IF NOT FOUND THEN
1340                         INSERT INTO asset.call_number (owning_lib, record, create_date, edit_date, creator, editor, label)
1341                             VALUES (uri_owner_id, bib_id, 'now', 'now', editor_id, editor_id, '##URI##');
1342                         SELECT id INTO uri_cn_id FROM asset.call_number WHERE owning_lib = uri_owner_id AND record = bib_id AND label = '##URI##' AND NOT deleted;
1343                     END IF;
1344
1345                     -- now, link them if they're not already
1346                     SELECT id INTO uri_map_id FROM asset.uri_call_number_map WHERE call_number = uri_cn_id AND uri = uri_id;
1347                     IF NOT FOUND THEN
1348                         INSERT INTO asset.uri_call_number_map (call_number, uri) VALUES (uri_cn_id, uri_id);
1349                     END IF;
1350
1351                 END LOOP;
1352
1353             END IF;
1354
1355         END LOOP;
1356     END IF;
1357
1358     RETURN;
1359 END;
1360 $func$ LANGUAGE PLPGSQL;
1361
1362 CREATE OR REPLACE FUNCTION metabib.remap_metarecord_for_bib( bib_id BIGINT, fp TEXT, bib_is_deleted BOOL DEFAULT FALSE, retain_deleted BOOL DEFAULT FALSE ) RETURNS BIGINT AS $func$
1363 DECLARE
1364     new_mapping     BOOL := TRUE;
1365     source_count    INT;
1366     old_mr          BIGINT;
1367     tmp_mr          metabib.metarecord%ROWTYPE;
1368     deleted_mrs     BIGINT[];
1369 BEGIN
1370
1371     -- We need to make sure we're not a deleted master record of an MR
1372     IF bib_is_deleted THEN
1373         FOR old_mr IN SELECT id FROM metabib.metarecord WHERE master_record = bib_id LOOP
1374
1375             IF NOT retain_deleted THEN -- Go away for any MR that we're master of, unless retained
1376                 DELETE FROM metabib.metarecord_source_map WHERE source = bib_id;
1377             END IF;
1378
1379             -- Now, are there any more sources on this MR?
1380             SELECT COUNT(*) INTO source_count FROM metabib.metarecord_source_map WHERE metarecord = old_mr;
1381
1382             IF source_count = 0 AND NOT retain_deleted THEN -- No other records
1383                 deleted_mrs := ARRAY_APPEND(deleted_mrs, old_mr); -- Just in case...
1384                 DELETE FROM metabib.metarecord WHERE id = old_mr;
1385
1386             ELSE -- indeed there are. Update it with a null cache and recalcualated master record
1387                 UPDATE  metabib.metarecord
1388                   SET   mods = NULL,
1389                         master_record = ( SELECT id FROM biblio.record_entry WHERE fingerprint = fp AND NOT deleted ORDER BY quality DESC LIMIT 1)
1390                   WHERE id = old_mr;
1391             END IF;
1392         END LOOP;
1393
1394     ELSE -- insert or update
1395
1396         FOR tmp_mr IN SELECT m.* FROM metabib.metarecord m JOIN metabib.metarecord_source_map s ON (s.metarecord = m.id) WHERE s.source = bib_id LOOP
1397
1398             -- Find the first fingerprint-matching
1399             IF old_mr IS NULL AND fp = tmp_mr.fingerprint THEN
1400                 old_mr := tmp_mr.id;
1401                 new_mapping := FALSE;
1402
1403             ELSE -- Our fingerprint changed ... maybe remove the old MR
1404                 DELETE FROM metabib.metarecord_source_map WHERE metarecord = old_mr AND source = bib_id; -- remove the old source mapping
1405                 SELECT COUNT(*) INTO source_count FROM metabib.metarecord_source_map WHERE metarecord = tmp_mr.id;
1406                 IF source_count = 0 THEN -- No other records
1407                     deleted_mrs := ARRAY_APPEND(deleted_mrs, tmp_mr.id);
1408                     DELETE FROM metabib.metarecord WHERE id = tmp_mr.id;
1409                 END IF;
1410             END IF;
1411
1412         END LOOP;
1413
1414         -- we found no suitable, preexisting MR based on old source maps
1415         IF old_mr IS NULL THEN
1416             SELECT id INTO old_mr FROM metabib.metarecord WHERE fingerprint = fp; -- is there one for our current fingerprint?
1417
1418             IF old_mr IS NULL THEN -- nope, create one and grab its id
1419                 INSERT INTO metabib.metarecord ( fingerprint, master_record ) VALUES ( fp, bib_id );
1420                 SELECT id INTO old_mr FROM metabib.metarecord WHERE fingerprint = fp;
1421
1422             ELSE -- indeed there is. update it with a null cache and recalcualated master record
1423                 UPDATE  metabib.metarecord
1424                   SET   mods = NULL,
1425                         master_record = ( SELECT id FROM biblio.record_entry WHERE fingerprint = fp AND NOT deleted ORDER BY quality DESC LIMIT 1)
1426                   WHERE id = old_mr;
1427             END IF;
1428
1429         ELSE -- there was one we already attached to, update its mods cache and master_record
1430             UPDATE  metabib.metarecord
1431               SET   mods = NULL,
1432                     master_record = ( SELECT id FROM biblio.record_entry WHERE fingerprint = fp AND NOT deleted ORDER BY quality DESC LIMIT 1)
1433               WHERE id = old_mr;
1434         END IF;
1435
1436         IF new_mapping THEN
1437             INSERT INTO metabib.metarecord_source_map (metarecord, source) VALUES (old_mr, bib_id); -- new source mapping
1438         END IF;
1439
1440     END IF;
1441
1442     IF ARRAY_UPPER(deleted_mrs,1) > 0 THEN
1443         UPDATE action.hold_request SET target = old_mr WHERE target IN ( SELECT unnest(deleted_mrs) ) AND hold_type = 'M'; -- if we had to delete any MRs above, make sure their holds are moved
1444     END IF;
1445
1446     RETURN old_mr;
1447
1448 END;
1449 $func$ LANGUAGE PLPGSQL;
1450
1451
1452 CREATE OR REPLACE FUNCTION biblio.map_authority_linking (bibid BIGINT, marc TEXT) RETURNS BIGINT AS $func$
1453     DELETE FROM authority.bib_linking WHERE bib = $1;
1454     INSERT INTO authority.bib_linking (bib, authority)
1455         SELECT  y.bib,
1456                 y.authority
1457           FROM (    SELECT  DISTINCT $1 AS bib,
1458                             BTRIM(remove_paren_substring(txt))::BIGINT AS authority
1459                       FROM  unnest(oils_xpath('//*[@code="0"]/text()',$2)) x(txt)
1460                       WHERE BTRIM(remove_paren_substring(txt)) ~ $re$^\d+$$re$
1461                 ) y JOIN authority.record_entry r ON r.id = y.authority;
1462     SELECT $1;
1463 $func$ LANGUAGE SQL;
1464
1465 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$
1466 DECLARE
1467     transformed_xml TEXT;
1468     rmarc           TEXT := prmarc;
1469     tmp_val         TEXT;
1470     prev_xfrm       TEXT;
1471     normalizer      RECORD;
1472     xfrm            config.xml_transform%ROWTYPE;
1473     attr_vector     INT[] := '{}'::INT[];
1474     attr_vector_tmp INT[];
1475     attr_list       TEXT[] := pattr_list;
1476     attr_value      TEXT[];
1477     norm_attr_value TEXT[];
1478     tmp_xml         XML;
1479     attr_def        config.record_attr_definition%ROWTYPE;
1480     ccvm_row        config.coded_value_map%ROWTYPE;
1481 BEGIN
1482
1483     IF attr_list IS NULL OR rdeleted THEN -- need to do the full dance on INSERT or undelete
1484         SELECT ARRAY_AGG(name) INTO attr_list FROM config.record_attr_definition;
1485     END IF;
1486
1487     IF rmarc IS NULL THEN
1488         SELECT marc INTO rmarc FROM biblio.record_entry WHERE id = rid;
1489     END IF;
1490
1491     FOR attr_def IN SELECT * FROM config.record_attr_definition WHERE NOT composite AND name = ANY( attr_list ) ORDER BY format LOOP
1492
1493         attr_value := '{}'::TEXT[];
1494         norm_attr_value := '{}'::TEXT[];
1495         attr_vector_tmp := '{}'::INT[];
1496
1497         SELECT * INTO ccvm_row FROM config.coded_value_map c WHERE c.ctype = attr_def.name LIMIT 1;
1498
1499         -- tag+sf attrs only support SVF
1500         IF attr_def.tag IS NOT NULL THEN -- tag (and optional subfield list) selection
1501             SELECT  ARRAY[ARRAY_TO_STRING(ARRAY_AGG(value), COALESCE(attr_def.joiner,' '))] INTO attr_value
1502               FROM  (SELECT * FROM metabib.full_rec ORDER BY tag, subfield) AS x
1503               WHERE record = rid
1504                     AND tag LIKE attr_def.tag
1505                     AND CASE
1506                         WHEN attr_def.sf_list IS NOT NULL
1507                             THEN POSITION(subfield IN attr_def.sf_list) > 0
1508                         ELSE TRUE
1509                     END
1510               GROUP BY tag
1511               ORDER BY tag
1512               LIMIT 1;
1513
1514         ELSIF attr_def.fixed_field IS NOT NULL THEN -- a named fixed field, see config.marc21_ff_pos_map.fixed_field
1515             attr_value := vandelay.marc21_extract_fixed_field_list(rmarc, attr_def.fixed_field);
1516
1517             IF NOT attr_def.multi THEN
1518                 attr_value := ARRAY[attr_value[1]];
1519             END IF;
1520
1521         ELSIF attr_def.xpath IS NOT NULL THEN -- and xpath expression
1522
1523             SELECT INTO xfrm * FROM config.xml_transform WHERE name = attr_def.format;
1524
1525             -- See if we can skip the XSLT ... it's expensive
1526             IF prev_xfrm IS NULL OR prev_xfrm <> xfrm.name THEN
1527                 -- Can't skip the transform
1528                 IF xfrm.xslt <> '---' THEN
1529                     transformed_xml := oils_xslt_process(rmarc,xfrm.xslt);
1530                 ELSE
1531                     transformed_xml := rmarc;
1532                 END IF;
1533
1534                 prev_xfrm := xfrm.name;
1535             END IF;
1536
1537             IF xfrm.name IS NULL THEN
1538                 -- just grab the marcxml (empty) transform
1539                 SELECT INTO xfrm * FROM config.xml_transform WHERE xslt = '---' LIMIT 1;
1540                 prev_xfrm := xfrm.name;
1541             END IF;
1542
1543             FOR tmp_xml IN SELECT XPATH(attr_def.xpath, transformed_xml, ARRAY[ARRAY[xfrm.prefix, xfrm.namespace_uri]]) LOOP
1544                 tmp_val := oils_xpath_string(
1545                                 '//*',
1546                                 tmp_xml::TEXT,
1547                                 COALESCE(attr_def.joiner,' '),
1548                                 ARRAY[ARRAY[xfrm.prefix, xfrm.namespace_uri]]
1549                             );
1550                 IF tmp_val IS NOT NULL AND BTRIM(tmp_val) <> '' THEN
1551                     attr_value := attr_value || tmp_val;
1552                     EXIT WHEN NOT attr_def.multi;
1553                 END IF;
1554             END LOOP;
1555
1556         ELSIF attr_def.phys_char_sf IS NOT NULL THEN -- a named Physical Characteristic, see config.marc21_physical_characteristic_*_map
1557             SELECT  ARRAY_AGG(m.value) INTO attr_value
1558               FROM  vandelay.marc21_physical_characteristics(rmarc) v
1559                     LEFT JOIN config.marc21_physical_characteristic_value_map m ON (m.id = v.value)
1560               WHERE v.subfield = attr_def.phys_char_sf AND (m.value IS NOT NULL AND BTRIM(m.value) <> '')
1561                     AND ( ccvm_row.id IS NULL OR ( ccvm_row.id IS NOT NULL AND v.id IS NOT NULL) );
1562
1563             IF NOT attr_def.multi THEN
1564                 attr_value := ARRAY[attr_value[1]];
1565             END IF;
1566
1567         END IF;
1568
1569                 -- apply index normalizers to attr_value
1570         FOR tmp_val IN SELECT value FROM UNNEST(attr_value) x(value) LOOP
1571             FOR normalizer IN
1572                 SELECT  n.func AS func,
1573                         n.param_count AS param_count,
1574                         m.params AS params
1575                   FROM  config.index_normalizer n
1576                         JOIN config.record_attr_index_norm_map m ON (m.norm = n.id)
1577                   WHERE attr = attr_def.name
1578                   ORDER BY m.pos LOOP
1579                     EXECUTE 'SELECT ' || normalizer.func || '(' ||
1580                     COALESCE( quote_literal( tmp_val ), 'NULL' ) ||
1581                         CASE
1582                             WHEN normalizer.param_count > 0
1583                                 THEN ',' || REPLACE(REPLACE(BTRIM(normalizer.params,'[]'),E'\'',E'\\\''),E'"',E'\'')
1584                                 ELSE ''
1585                             END ||
1586                     ')' INTO tmp_val;
1587
1588             END LOOP;
1589             IF tmp_val IS NOT NULL AND BTRIM(tmp_val) <> '' THEN
1590                 norm_attr_value := norm_attr_value || tmp_val;
1591             END IF;
1592         END LOOP;
1593
1594         IF attr_def.filter THEN
1595             -- Create unknown uncontrolled values and find the IDs of the values
1596             IF ccvm_row.id IS NULL THEN
1597                 FOR tmp_val IN SELECT value FROM UNNEST(norm_attr_value) x(value) LOOP
1598                     IF tmp_val IS NOT NULL AND BTRIM(tmp_val) <> '' THEN
1599                         BEGIN -- use subtransaction to isolate unique constraint violations
1600                             INSERT INTO metabib.uncontrolled_record_attr_value ( attr, value ) VALUES ( attr_def.name, tmp_val );
1601                         EXCEPTION WHEN unique_violation THEN END;
1602                     END IF;
1603                 END LOOP;
1604
1605                 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 );
1606             ELSE
1607                 SELECT ARRAY_AGG(id) INTO attr_vector_tmp FROM config.coded_value_map WHERE ctype = attr_def.name AND code = ANY( norm_attr_value );
1608             END IF;
1609
1610             -- Add the new value to the vector
1611             attr_vector := attr_vector || attr_vector_tmp;
1612         END IF;
1613
1614         IF attr_def.sorter AND norm_attr_value[1] IS NOT NULL THEN
1615             DELETE FROM metabib.record_sorter WHERE source = rid AND attr = attr_def.name;
1616             INSERT INTO metabib.record_sorter (source, attr, value) VALUES (rid, attr_def.name, norm_attr_value[1]);
1617         END IF;
1618
1619     END LOOP;
1620
1621 /* We may need to rewrite the vlist to contain
1622    the intersection of new values for requested
1623    attrs and old values for ignored attrs. To
1624    do this, we take the old attr vlist and
1625    subtract any values that are valid for the
1626    requested attrs, and then add back the new
1627    set of attr values. */
1628
1629     IF ARRAY_LENGTH(pattr_list, 1) > 0 THEN
1630         SELECT vlist INTO attr_vector_tmp FROM metabib.record_attr_vector_list WHERE source = rid;
1631         SELECT attr_vector_tmp - ARRAY_AGG(id::INT) INTO attr_vector_tmp FROM metabib.full_attr_id_map WHERE attr = ANY (pattr_list);
1632         attr_vector := attr_vector || attr_vector_tmp;
1633     END IF;
1634
1635     -- On to composite attributes, now that the record attrs have been pulled.  Processed in name order, so later composite
1636     -- attributes can depend on earlier ones.
1637     PERFORM metabib.compile_composite_attr_cache_init();
1638     FOR attr_def IN SELECT * FROM config.record_attr_definition WHERE composite AND name = ANY( attr_list ) ORDER BY name LOOP
1639
1640         FOR ccvm_row IN SELECT * FROM config.coded_value_map c WHERE c.ctype = attr_def.name ORDER BY value LOOP
1641
1642             tmp_val := metabib.compile_composite_attr( ccvm_row.id );
1643             CONTINUE WHEN tmp_val IS NULL OR tmp_val = ''; -- nothing to do
1644
1645             IF attr_def.filter THEN
1646                 IF attr_vector @@ tmp_val::query_int THEN
1647                     attr_vector = attr_vector + intset(ccvm_row.id);
1648                     EXIT WHEN NOT attr_def.multi;
1649                 END IF;
1650             END IF;
1651
1652             IF attr_def.sorter THEN
1653                 IF attr_vector @@ tmp_val THEN
1654                     DELETE FROM metabib.record_sorter WHERE source = rid AND attr = attr_def.name;
1655                     INSERT INTO metabib.record_sorter (source, attr, value) VALUES (rid, attr_def.name, ccvm_row.code);
1656                 END IF;
1657             END IF;
1658
1659         END LOOP;
1660
1661     END LOOP;
1662
1663     IF ARRAY_LENGTH(attr_vector, 1) > 0 THEN
1664         IF rdeleted THEN -- initial insert OR revivication
1665             DELETE FROM metabib.record_attr_vector_list WHERE source = rid;
1666             INSERT INTO metabib.record_attr_vector_list (source, vlist) VALUES (rid, attr_vector);
1667         ELSE
1668             UPDATE metabib.record_attr_vector_list SET vlist = attr_vector WHERE source = rid;
1669         END IF;
1670     END IF;
1671
1672 END;
1673
1674 $func$ LANGUAGE PLPGSQL;
1675
1676
1677 -- AFTER UPDATE OR INSERT trigger for biblio.record_entry
1678 CREATE OR REPLACE FUNCTION biblio.indexing_ingest_or_delete () RETURNS TRIGGER AS $func$
1679 DECLARE
1680     tmp_bool BOOL;
1681 BEGIN
1682
1683     IF NEW.deleted THEN -- If this bib is deleted
1684
1685         PERFORM * FROM config.internal_flag WHERE
1686             name = 'ingest.metarecord_mapping.preserve_on_delete' AND enabled;
1687
1688         tmp_bool := FOUND; -- Just in case this is changed by some other statement
1689
1690         PERFORM metabib.remap_metarecord_for_bib( NEW.id, NEW.fingerprint, TRUE, tmp_bool );
1691
1692         IF NOT tmp_bool THEN
1693             -- One needs to keep these around to support searches
1694             -- with the #deleted modifier, so one should turn on the named
1695             -- internal flag for that functionality.
1696             DELETE FROM metabib.record_attr_vector_list WHERE source = NEW.id;
1697         END IF;
1698
1699         DELETE FROM authority.bib_linking WHERE bib = NEW.id; -- Avoid updating fields in bibs that are no longer visible
1700         DELETE FROM biblio.peer_bib_copy_map WHERE peer_record = NEW.id; -- Separate any multi-homed items
1701         DELETE FROM metabib.browse_entry_def_map WHERE source = NEW.id; -- Don't auto-suggest deleted bibs
1702         RETURN NEW; -- and we're done
1703     END IF;
1704
1705     IF TG_OP = 'UPDATE' THEN -- re-ingest?
1706         PERFORM * FROM config.internal_flag WHERE name = 'ingest.reingest.force_on_same_marc' AND enabled;
1707
1708         IF NOT FOUND AND OLD.marc = NEW.marc THEN -- don't do anything if the MARC didn't change
1709             RETURN NEW;
1710         END IF;
1711     END IF;
1712
1713     -- Record authority linking
1714     PERFORM * FROM config.internal_flag WHERE name = 'ingest.disable_authority_linking' AND enabled;
1715     IF NOT FOUND THEN
1716         PERFORM biblio.map_authority_linking( NEW.id, NEW.marc );
1717     END IF;
1718
1719     -- Flatten and insert the mfr data
1720     PERFORM * FROM config.internal_flag WHERE name = 'ingest.disable_metabib_full_rec' AND enabled;
1721     IF NOT FOUND THEN
1722         PERFORM metabib.reingest_metabib_full_rec(NEW.id);
1723
1724         -- Now we pull out attribute data, which is dependent on the mfr for all but XPath-based fields
1725         PERFORM * FROM config.internal_flag WHERE name = 'ingest.disable_metabib_rec_descriptor' AND enabled;
1726         IF NOT FOUND THEN
1727             PERFORM metabib.reingest_record_attributes(NEW.id, NULL, NEW.marc, TG_OP = 'INSERT' OR OLD.deleted);
1728         END IF;
1729     END IF;
1730
1731     -- Gather and insert the field entry data
1732     PERFORM metabib.reingest_metabib_field_entries(NEW.id);
1733
1734     -- Located URI magic
1735     IF TG_OP = 'INSERT' THEN
1736         PERFORM * FROM config.internal_flag WHERE name = 'ingest.disable_located_uri' AND enabled;
1737         IF NOT FOUND THEN
1738             PERFORM biblio.extract_located_uris( NEW.id, NEW.marc, NEW.editor );
1739         END IF;
1740     ELSE
1741         PERFORM * FROM config.internal_flag WHERE name = 'ingest.disable_located_uri' AND enabled;
1742         IF NOT FOUND THEN
1743             PERFORM biblio.extract_located_uris( NEW.id, NEW.marc, NEW.editor );
1744         END IF;
1745     END IF;
1746
1747     -- (re)map metarecord-bib linking
1748     IF TG_OP = 'INSERT' THEN -- if not deleted and performing an insert, check for the flag
1749         PERFORM * FROM config.internal_flag WHERE name = 'ingest.metarecord_mapping.skip_on_insert' AND enabled;
1750         IF NOT FOUND THEN
1751             PERFORM metabib.remap_metarecord_for_bib( NEW.id, NEW.fingerprint );
1752         END IF;
1753     ELSE -- we're doing an update, and we're not deleted, remap
1754         PERFORM * FROM config.internal_flag WHERE name = 'ingest.metarecord_mapping.skip_on_update' AND enabled;
1755         IF NOT FOUND THEN
1756             PERFORM metabib.remap_metarecord_for_bib( NEW.id, NEW.fingerprint );
1757         END IF;
1758     END IF;
1759
1760     RETURN NEW;
1761 END;
1762 $func$ LANGUAGE PLPGSQL;
1763
1764 CREATE OR REPLACE FUNCTION metabib.browse_normalize(facet_text TEXT, mapped_field INT) RETURNS TEXT AS $$
1765 DECLARE
1766     normalizer  RECORD;
1767 BEGIN
1768
1769     FOR normalizer IN
1770         SELECT  n.func AS func,
1771                 n.param_count AS param_count,
1772                 m.params AS params
1773           FROM  config.index_normalizer n
1774                 JOIN config.metabib_field_index_norm_map m ON (m.norm = n.id)
1775           WHERE m.field = mapped_field AND m.pos < 0
1776           ORDER BY m.pos LOOP
1777
1778             EXECUTE 'SELECT ' || normalizer.func || '(' ||
1779                 quote_literal( facet_text ) ||
1780                 CASE
1781                     WHEN normalizer.param_count > 0
1782                         THEN ',' || REPLACE(REPLACE(BTRIM(normalizer.params,'[]'),E'\'',E'\\\''),E'"',E'\'')
1783                         ELSE ''
1784                     END ||
1785                 ')' INTO facet_text;
1786
1787     END LOOP;
1788
1789     RETURN facet_text;
1790 END;
1791
1792 $$ LANGUAGE PLPGSQL;
1793
1794
1795 -- This mimics a specific part of QueryParser, turning the first part of a
1796 -- classed search (search_class) into a set of classes and possibly fields.
1797 -- search_class might look like "author" or "title|proper" or "ti|uniform"
1798 -- or "au" or "au|corporate|personal" or anything like that, where the first
1799 -- element of the list you get by separating on the "|" character is either
1800 -- a registered class (config.metabib_class) or an alias
1801 -- (config.metabib_search_alias), and the rest of any such elements are
1802 -- fields (config.metabib_field).
1803 CREATE OR REPLACE
1804     FUNCTION metabib.search_class_to_registered_components(search_class TEXT)
1805     RETURNS SETOF RECORD AS $func$
1806 DECLARE
1807     search_parts        TEXT[];
1808     field_name          TEXT;
1809     search_part_count   INTEGER;
1810     rec                 RECORD;
1811     registered_class    config.metabib_class%ROWTYPE;
1812     registered_alias    config.metabib_search_alias%ROWTYPE;
1813     registered_field    config.metabib_field%ROWTYPE;
1814 BEGIN
1815     search_parts := REGEXP_SPLIT_TO_ARRAY(search_class, E'\\|');
1816
1817     search_part_count := ARRAY_LENGTH(search_parts, 1);
1818     IF search_part_count = 0 THEN
1819         RETURN;
1820     ELSE
1821         SELECT INTO registered_class
1822             * FROM config.metabib_class WHERE name = search_parts[1];
1823         IF FOUND THEN
1824             IF search_part_count < 2 THEN   -- all fields
1825                 rec := (registered_class.name, NULL::INTEGER);
1826                 RETURN NEXT rec;
1827                 RETURN; -- done
1828             END IF;
1829             FOR field_name IN SELECT *
1830                 FROM UNNEST(search_parts[2:search_part_count]) LOOP
1831                 SELECT INTO registered_field
1832                     * FROM config.metabib_field
1833                     WHERE name = field_name AND
1834                         field_class = registered_class.name;
1835                 IF FOUND THEN
1836                     rec := (registered_class.name, registered_field.id);
1837                     RETURN NEXT rec;
1838                 END IF;
1839             END LOOP;
1840         ELSE
1841             -- maybe we have an alias?
1842             SELECT INTO registered_alias
1843                 * FROM config.metabib_search_alias WHERE alias=search_parts[1];
1844             IF NOT FOUND THEN
1845                 RETURN;
1846             ELSE
1847                 IF search_part_count < 2 THEN   -- return w/e the alias says
1848                     rec := (
1849                         registered_alias.field_class, registered_alias.field
1850                     );
1851                     RETURN NEXT rec;
1852                     RETURN; -- done
1853                 ELSE
1854                     FOR field_name IN SELECT *
1855                         FROM UNNEST(search_parts[2:search_part_count]) LOOP
1856                         SELECT INTO registered_field
1857                             * FROM config.metabib_field
1858                             WHERE name = field_name AND
1859                                 field_class = registered_alias.field_class;
1860                         IF FOUND THEN
1861                             rec := (
1862                                 registered_alias.field_class,
1863                                 registered_field.id
1864                             );
1865                             RETURN NEXT rec;
1866                         END IF;
1867                     END LOOP;
1868                 END IF;
1869             END IF;
1870         END IF;
1871     END IF;
1872 END;
1873 $func$ LANGUAGE PLPGSQL ROWS 1;
1874
1875
1876 -- Given a string such as a user might type into a search box, prepare
1877 -- two changed variants for TO_TSQUERY(). See
1878 -- http://www.postgresql.org/docs/9.0/static/textsearch-controls.html
1879 -- The first variant is normalized to match indexed documents regardless
1880 -- of diacritics.  The second variant keeps its diacritics for proper
1881 -- highlighting via TS_HEADLINE().
1882 CREATE OR REPLACE
1883     FUNCTION metabib.autosuggest_prepare_tsquery(orig TEXT) RETURNS TEXT[] AS
1884 $$
1885 DECLARE
1886     orig_ended_in_space     BOOLEAN;
1887     result                  RECORD;
1888     plain                   TEXT;
1889     normalized              TEXT;
1890 BEGIN
1891     orig_ended_in_space := orig ~ E'\\s$';
1892
1893     orig := ARRAY_TO_STRING(
1894         evergreen.regexp_split_to_array(orig, E'\\W+'), ' '
1895     );
1896
1897     normalized := public.naco_normalize(orig); -- also trim()s
1898     plain := trim(orig);
1899
1900     IF NOT orig_ended_in_space THEN
1901         plain := plain || ':*';
1902         normalized := normalized || ':*';
1903     END IF;
1904
1905     plain := ARRAY_TO_STRING(
1906         evergreen.regexp_split_to_array(plain, E'\\s+'), ' & '
1907     );
1908     normalized := ARRAY_TO_STRING(
1909         evergreen.regexp_split_to_array(normalized, E'\\s+'), ' & '
1910     );
1911
1912     RETURN ARRAY[normalized, plain];
1913 END;
1914 $$ LANGUAGE PLPGSQL;
1915
1916
1917 CREATE OR REPLACE
1918     FUNCTION metabib.suggest_browse_entries(
1919         raw_query_text  TEXT,   -- actually typed by humans at the UI level
1920         search_class    TEXT,   -- 'alias' or 'class' or 'class|field..', etc
1921         headline_opts   TEXT,   -- markup options for ts_headline()
1922         visibility_org  INTEGER,-- null if you don't want opac visibility test
1923         query_limit     INTEGER,-- use in LIMIT clause of interal query
1924         normalization   INTEGER -- argument to TS_RANK_CD()
1925     ) RETURNS TABLE (
1926         value                   TEXT,   -- plain
1927         field                   INTEGER,
1928         buoyant_and_class_match BOOL,
1929         field_match             BOOL,
1930         field_weight            INTEGER,
1931         rank                    REAL,
1932         buoyant                 BOOL,
1933         match                   TEXT    -- marked up
1934     ) AS $func$
1935 DECLARE
1936     prepared_query_texts    TEXT[];
1937     query                   TSQUERY;
1938     plain_query             TSQUERY;
1939     opac_visibility_join    TEXT;
1940     search_class_join       TEXT;
1941     r_fields                RECORD;
1942 BEGIN
1943     prepared_query_texts := metabib.autosuggest_prepare_tsquery(raw_query_text);
1944
1945     query := TO_TSQUERY('keyword', prepared_query_texts[1]);
1946     plain_query := TO_TSQUERY('keyword', prepared_query_texts[2]);
1947
1948     visibility_org := NULLIF(visibility_org,-1);
1949     IF visibility_org IS NOT NULL THEN
1950         opac_visibility_join := '
1951     JOIN asset.opac_visible_copies aovc ON (
1952         aovc.record = x.source AND
1953         aovc.circ_lib IN (SELECT id FROM actor.org_unit_descendants($4))
1954     )';
1955     ELSE
1956         opac_visibility_join := '';
1957     END IF;
1958
1959     -- The following determines whether we only provide suggestsons matching
1960     -- the user's selected search_class, or whether we show other suggestions
1961     -- too. The reason for MIN() is that for search_classes like
1962     -- 'title|proper|uniform' you would otherwise get multiple rows.  The
1963     -- implication is that if title as a class doesn't have restrict,
1964     -- nor does the proper field, but the uniform field does, you're going
1965     -- to get 'false' for your overall evaluation of 'should we restrict?'
1966     -- To invert that, change from MIN() to MAX().
1967
1968     SELECT
1969         INTO r_fields
1970             MIN(cmc.restrict::INT) AS restrict_class,
1971             MIN(cmf.restrict::INT) AS restrict_field
1972         FROM metabib.search_class_to_registered_components(search_class)
1973             AS _registered (field_class TEXT, field INT)
1974         JOIN
1975             config.metabib_class cmc ON (cmc.name = _registered.field_class)
1976         LEFT JOIN
1977             config.metabib_field cmf ON (cmf.id = _registered.field);
1978
1979     -- evaluate 'should we restrict?'
1980     IF r_fields.restrict_field::BOOL OR r_fields.restrict_class::BOOL THEN
1981         search_class_join := '
1982     JOIN
1983         metabib.search_class_to_registered_components($2)
1984         AS _registered (field_class TEXT, field INT) ON (
1985             (_registered.field IS NULL AND
1986                 _registered.field_class = cmf.field_class) OR
1987             (_registered.field = cmf.id)
1988         )
1989     ';
1990     ELSE
1991         search_class_join := '
1992     LEFT JOIN
1993         metabib.search_class_to_registered_components($2)
1994         AS _registered (field_class TEXT, field INT) ON (
1995             _registered.field_class = cmc.name
1996         )
1997     ';
1998     END IF;
1999
2000     RETURN QUERY EXECUTE '
2001 SELECT  DISTINCT
2002         x.value,
2003         x.id,
2004         x.push,
2005         x.restrict,
2006         x.weight,
2007         x.ts_rank_cd,
2008         x.buoyant,
2009         TS_HEADLINE(value, $7, $3)
2010   FROM  (SELECT DISTINCT
2011                 mbe.value,
2012                 cmf.id,
2013                 cmc.buoyant AND _registered.field_class IS NOT NULL AS push,
2014                 _registered.field = cmf.id AS restrict,
2015                 cmf.weight,
2016                 TS_RANK_CD(mbe.index_vector, $1, $6),
2017                 cmc.buoyant,
2018                 mbedm.source
2019           FROM  metabib.browse_entry_def_map mbedm
2020                 JOIN (SELECT * FROM metabib.browse_entry WHERE index_vector @@ $1 LIMIT 10000) mbe ON (mbe.id = mbedm.entry)
2021                 JOIN config.metabib_field cmf ON (cmf.id = mbedm.def)
2022                 JOIN config.metabib_class cmc ON (cmf.field_class = cmc.name)
2023                 '  || search_class_join || '
2024           ORDER BY 3 DESC, 4 DESC NULLS LAST, 5 DESC, 6 DESC, 7 DESC, 1 ASC
2025           LIMIT 1000) AS x
2026         ' || opac_visibility_join || '
2027   ORDER BY 3 DESC, 4 DESC NULLS LAST, 5 DESC, 6 DESC, 7 DESC, 1 ASC
2028   LIMIT $5
2029 '   -- sic, repeat the order by clause in the outer select too
2030     USING
2031         query, search_class, headline_opts,
2032         visibility_org, query_limit, normalization, plain_query
2033         ;
2034
2035     -- sort order:
2036     --  buoyant AND chosen class = match class
2037     --  chosen field = match field
2038     --  field weight
2039     --  rank
2040     --  buoyancy
2041     --  value itself
2042
2043 END;
2044 $func$ LANGUAGE PLPGSQL;
2045
2046 CREATE OR REPLACE FUNCTION public.oils_tsearch2 () RETURNS TRIGGER AS $$
2047 DECLARE
2048     normalizer      RECORD;
2049     value           TEXT := '';
2050     temp_vector     TEXT := '';
2051     ts_rec          RECORD;
2052     cur_weight      "char";
2053 BEGIN
2054
2055     value := NEW.value;
2056     NEW.index_vector = ''::tsvector;
2057
2058     IF TG_TABLE_NAME::TEXT ~ 'field_entry$' THEN
2059         FOR normalizer IN
2060             SELECT  n.func AS func,
2061                     n.param_count AS param_count,
2062                     m.params AS params
2063               FROM  config.index_normalizer n
2064                     JOIN config.metabib_field_index_norm_map m ON (m.norm = n.id)
2065               WHERE field = NEW.field AND m.pos < 0
2066               ORDER BY m.pos LOOP
2067                 EXECUTE 'SELECT ' || normalizer.func || '(' ||
2068                     quote_literal( value ) ||
2069                     CASE
2070                         WHEN normalizer.param_count > 0
2071                             THEN ',' || REPLACE(REPLACE(BTRIM(normalizer.params,'[]'),E'\'',E'\\\''),E'"',E'\'')
2072                             ELSE ''
2073                         END ||
2074                     ')' INTO value;
2075
2076         END LOOP;
2077
2078         NEW.value = value;
2079
2080         FOR normalizer IN
2081             SELECT  n.func AS func,
2082                     n.param_count AS param_count,
2083                     m.params AS params
2084               FROM  config.index_normalizer n
2085                     JOIN config.metabib_field_index_norm_map m ON (m.norm = n.id)
2086               WHERE field = NEW.field AND m.pos >= 0
2087               ORDER BY m.pos LOOP
2088                 EXECUTE 'SELECT ' || normalizer.func || '(' ||
2089                     quote_literal( value ) ||
2090                     CASE
2091                         WHEN normalizer.param_count > 0
2092                             THEN ',' || REPLACE(REPLACE(BTRIM(normalizer.params,'[]'),E'\'',E'\\\''),E'"',E'\'')
2093                             ELSE ''
2094                         END ||
2095                     ')' INTO value;
2096
2097         END LOOP;
2098    END IF;
2099
2100     IF TG_TABLE_NAME::TEXT ~ 'browse_entry$' THEN
2101
2102         value :=  ARRAY_TO_STRING(
2103             evergreen.regexp_split_to_array(value, E'\\W+'), ' '
2104         );
2105         value := public.search_normalize(value);
2106         NEW.index_vector = to_tsvector(TG_ARGV[0]::regconfig, value);
2107
2108     ELSIF TG_TABLE_NAME::TEXT ~ 'field_entry$' THEN
2109         FOR ts_rec IN
2110
2111             SELECT DISTINCT m.ts_config, m.index_weight
2112             FROM config.metabib_class_ts_map m
2113                  LEFT JOIN metabib.record_attr_vector_list r ON (r.source = NEW.source)
2114                  LEFT JOIN config.coded_value_map ccvm ON (
2115                     ccvm.ctype IN ('item_lang', 'language') AND
2116                     ccvm.code = m.index_lang AND
2117                     r.vlist @> intset(ccvm.id)
2118                 )
2119             WHERE m.field_class = TG_ARGV[0]
2120                 AND m.active
2121                 AND (m.always OR NOT EXISTS (SELECT 1 FROM config.metabib_field_ts_map WHERE metabib_field = NEW.field))
2122                 AND (m.index_lang IS NULL OR ccvm.id IS NOT NULL)
2123                         UNION
2124             SELECT DISTINCT m.ts_config, m.index_weight
2125             FROM config.metabib_field_ts_map m
2126                  LEFT JOIN metabib.record_attr_vector_list r ON (r.source = NEW.source)
2127                  LEFT JOIN config.coded_value_map ccvm ON (
2128                     ccvm.ctype IN ('item_lang', 'language') AND
2129                     ccvm.code = m.index_lang AND
2130                     r.vlist @> intset(ccvm.id)
2131                 )
2132             WHERE m.metabib_field = NEW.field
2133                 AND m.active
2134                 AND (m.index_lang IS NULL OR ccvm.id IS NOT NULL)
2135             ORDER BY index_weight ASC
2136
2137         LOOP
2138
2139             IF cur_weight IS NOT NULL AND cur_weight != ts_rec.index_weight THEN
2140                 NEW.index_vector = NEW.index_vector || setweight(temp_vector::tsvector,cur_weight);
2141                 temp_vector = '';
2142             END IF;
2143
2144             cur_weight = ts_rec.index_weight;
2145             SELECT INTO temp_vector temp_vector || ' ' || to_tsvector(ts_rec.ts_config::regconfig, value)::TEXT;
2146
2147         END LOOP;
2148         NEW.index_vector = NEW.index_vector || setweight(temp_vector::tsvector,cur_weight);
2149     ELSE
2150         NEW.index_vector = to_tsvector(TG_ARGV[0]::regconfig, value);
2151     END IF;
2152
2153     RETURN NEW;
2154 END;
2155 $$ LANGUAGE PLPGSQL;
2156
2157
2158 CREATE TYPE metabib.flat_browse_entry_appearance AS (
2159     browse_entry    BIGINT,
2160     value           TEXT,
2161     fields          TEXT,
2162     authorities     TEXT,
2163     sees            TEXT,
2164     sources         INT,        -- visible ones, that is
2165     asources        INT,        -- visible ones, that is
2166     row_number      INT,        -- internal use, sort of
2167     accurate        BOOL,       -- Count in sources field is accurate? Not
2168                                 -- if we had more than a browse superpage
2169                                 -- of records to look at.
2170     aaccurate       BOOL,       -- See previous comment...
2171     pivot_point     BIGINT
2172 );
2173
2174
2175 CREATE OR REPLACE FUNCTION metabib.browse_bib_pivot(
2176     INT[],
2177     TEXT
2178 ) RETURNS BIGINT AS $p$
2179     SELECT  mbe.id
2180       FROM  metabib.browse_entry mbe
2181             JOIN metabib.browse_entry_def_map mbedm ON (
2182                 mbedm.entry = mbe.id
2183                 AND mbedm.def = ANY($1)
2184             )
2185       WHERE mbe.sort_value >= public.naco_normalize($2)
2186       ORDER BY mbe.sort_value, mbe.value LIMIT 1;
2187 $p$ LANGUAGE SQL STABLE;
2188
2189 CREATE OR REPLACE FUNCTION metabib.browse_authority_pivot(
2190     INT[],
2191     TEXT
2192 ) RETURNS BIGINT AS $p$
2193     SELECT  mbe.id
2194       FROM  metabib.browse_entry mbe
2195             JOIN metabib.browse_entry_simple_heading_map mbeshm ON ( mbeshm.entry = mbe.id )
2196             JOIN authority.simple_heading ash ON ( mbeshm.simple_heading = ash.id )
2197             JOIN authority.control_set_auth_field_metabib_field_map_refs map ON (
2198                 ash.atag = map.authority_field
2199                 AND map.metabib_field = ANY($1)
2200             )
2201       WHERE mbe.sort_value >= public.naco_normalize($2)
2202       ORDER BY mbe.sort_value, mbe.value LIMIT 1;
2203 $p$ LANGUAGE SQL STABLE;
2204
2205 CREATE OR REPLACE FUNCTION metabib.browse_authority_refs_pivot(
2206     INT[],
2207     TEXT
2208 ) RETURNS BIGINT AS $p$
2209     SELECT  mbe.id
2210       FROM  metabib.browse_entry mbe
2211             JOIN metabib.browse_entry_simple_heading_map mbeshm ON ( mbeshm.entry = mbe.id )
2212             JOIN authority.simple_heading ash ON ( mbeshm.simple_heading = ash.id )
2213             JOIN authority.control_set_auth_field_metabib_field_map_refs_only map ON (
2214                 ash.atag = map.authority_field
2215                 AND map.metabib_field = ANY($1)
2216             )
2217       WHERE mbe.sort_value >= public.naco_normalize($2)
2218       ORDER BY mbe.sort_value, mbe.value LIMIT 1;
2219 $p$ LANGUAGE SQL STABLE;
2220
2221 CREATE OR REPLACE FUNCTION metabib.browse_pivot(
2222     INT[],
2223     TEXT
2224 ) RETURNS BIGINT AS $p$
2225     SELECT  id FROM metabib.browse_entry
2226       WHERE id IN (
2227                 metabib.browse_bib_pivot($1, $2),
2228                 metabib.browse_authority_refs_pivot($1,$2) -- only look in 4xx, 5xx, 7xx of authority
2229             )
2230       ORDER BY sort_value, value LIMIT 1;
2231 $p$ LANGUAGE SQL STABLE;
2232
2233
2234 CREATE OR REPLACE FUNCTION metabib.staged_browse(
2235     query                   TEXT,
2236     fields                  INT[],
2237     context_org             INT,
2238     context_locations       INT[],
2239     staff                   BOOL,
2240     browse_superpage_size   INT,
2241     count_up_from_zero      BOOL,   -- if false, count down from -1
2242     result_limit            INT,
2243     next_pivot_pos          INT
2244 ) RETURNS SETOF metabib.flat_browse_entry_appearance AS $p$
2245 DECLARE
2246     curs                    REFCURSOR;
2247     rec                     RECORD;
2248     qpfts_query             TEXT;
2249     aqpfts_query            TEXT;
2250     afields                 INT[];
2251     bfields                 INT[];
2252     result_row              metabib.flat_browse_entry_appearance%ROWTYPE;
2253     results_skipped         INT := 0;
2254     row_counter             INT := 0;
2255     row_number              INT;
2256     slice_start             INT;
2257     slice_end               INT;
2258     full_end                INT;
2259     all_records             BIGINT[];
2260     all_brecords             BIGINT[];
2261     all_arecords            BIGINT[];
2262     superpage_of_records    BIGINT[];
2263     superpage_size          INT;
2264 BEGIN
2265     IF count_up_from_zero THEN
2266         row_number := 0;
2267     ELSE
2268         row_number := -1;
2269     END IF;
2270
2271     OPEN curs FOR EXECUTE query;
2272
2273     LOOP
2274         FETCH curs INTO rec;
2275         IF NOT FOUND THEN
2276             IF result_row.pivot_point IS NOT NULL THEN
2277                 RETURN NEXT result_row;
2278             END IF;
2279             RETURN;
2280         END IF;
2281
2282
2283         -- Gather aggregate data based on the MBE row we're looking at now, authority axis
2284         SELECT INTO all_arecords, result_row.sees, afields
2285                 ARRAY_AGG(DISTINCT abl.bib), -- bibs to check for visibility
2286                 STRING_AGG(DISTINCT aal.source::TEXT, $$,$$), -- authority record ids
2287                 ARRAY_AGG(DISTINCT map.metabib_field) -- authority-tag-linked CMF rows
2288
2289           FROM  metabib.browse_entry_simple_heading_map mbeshm
2290                 JOIN authority.simple_heading ash ON ( mbeshm.simple_heading = ash.id )
2291                 JOIN authority.authority_linking aal ON ( ash.record = aal.source )
2292                 JOIN authority.bib_linking abl ON ( aal.target = abl.authority )
2293                 JOIN authority.control_set_auth_field_metabib_field_map_refs map ON (
2294                     ash.atag = map.authority_field
2295                     AND map.metabib_field = ANY(fields)
2296                 )
2297           WHERE mbeshm.entry = rec.id;
2298
2299
2300         -- Gather aggregate data based on the MBE row we're looking at now, bib axis
2301         SELECT INTO all_brecords, result_row.authorities, bfields
2302                 ARRAY_AGG(DISTINCT source),
2303                 STRING_AGG(DISTINCT authority::TEXT, $$,$$),
2304                 ARRAY_AGG(DISTINCT def)
2305           FROM  metabib.browse_entry_def_map
2306           WHERE entry = rec.id
2307                 AND def = ANY(fields);
2308
2309         SELECT INTO result_row.fields STRING_AGG(DISTINCT x::TEXT, $$,$$) FROM UNNEST(afields || bfields) x;
2310
2311         result_row.sources := 0;
2312         result_row.asources := 0;
2313
2314         -- Bib-linked vis checking
2315         IF ARRAY_UPPER(all_brecords,1) IS NOT NULL THEN
2316
2317             full_end := ARRAY_LENGTH(all_brecords, 1);
2318             superpage_size := COALESCE(browse_superpage_size, full_end);
2319             slice_start := 1;
2320             slice_end := superpage_size;
2321
2322             WHILE result_row.sources = 0 AND slice_start <= full_end LOOP
2323                 superpage_of_records := all_brecords[slice_start:slice_end];
2324                 qpfts_query :=
2325                     'SELECT NULL::BIGINT AS id, ARRAY[r] AS records, ' ||
2326                     '1::INT AS rel FROM (SELECT UNNEST(' ||
2327                     quote_literal(superpage_of_records) || '::BIGINT[]) AS r) rr';
2328
2329                 -- We use search.query_parser_fts() for visibility testing.
2330                 -- We're calling it once per browse-superpage worth of records
2331                 -- out of the set of records related to a given mbe, until we've
2332                 -- either exhausted that set of records or found at least 1
2333                 -- visible record.
2334
2335                 SELECT INTO result_row.sources visible
2336                     FROM search.query_parser_fts(
2337                         context_org, NULL, qpfts_query, NULL,
2338                         context_locations, 0, NULL, NULL, FALSE, staff, FALSE
2339                     ) qpfts
2340                     WHERE qpfts.rel IS NULL;
2341
2342                 slice_start := slice_start + superpage_size;
2343                 slice_end := slice_end + superpage_size;
2344             END LOOP;
2345
2346             -- Accurate?  Well, probably.
2347             result_row.accurate := browse_superpage_size IS NULL OR
2348                 browse_superpage_size >= full_end;
2349
2350         END IF;
2351
2352         -- Authority-linked vis checking
2353         IF ARRAY_UPPER(all_arecords,1) IS NOT NULL THEN
2354
2355             full_end := ARRAY_LENGTH(all_arecords, 1);
2356             superpage_size := COALESCE(browse_superpage_size, full_end);
2357             slice_start := 1;
2358             slice_end := superpage_size;
2359
2360             WHILE result_row.asources = 0 AND slice_start <= full_end LOOP
2361                 superpage_of_records := all_arecords[slice_start:slice_end];
2362                 qpfts_query :=
2363                     'SELECT NULL::BIGINT AS id, ARRAY[r] AS records, ' ||
2364                     '1::INT AS rel FROM (SELECT UNNEST(' ||
2365                     quote_literal(superpage_of_records) || '::BIGINT[]) AS r) rr';
2366
2367                 -- We use search.query_parser_fts() for visibility testing.
2368                 -- We're calling it once per browse-superpage worth of records
2369                 -- out of the set of records related to a given mbe, via
2370                 -- authority until we've either exhausted that set of records
2371                 -- or found at least 1 visible record.
2372
2373                 SELECT INTO result_row.asources visible
2374                     FROM search.query_parser_fts(
2375                         context_org, NULL, qpfts_query, NULL,
2376                         context_locations, 0, NULL, NULL, FALSE, staff, FALSE
2377                     ) qpfts
2378                     WHERE qpfts.rel IS NULL;
2379
2380                 slice_start := slice_start + superpage_size;
2381                 slice_end := slice_end + superpage_size;
2382             END LOOP;
2383
2384
2385             -- Accurate?  Well, probably.
2386             result_row.aaccurate := browse_superpage_size IS NULL OR
2387                 browse_superpage_size >= full_end;
2388
2389         END IF;
2390
2391         IF result_row.sources > 0 OR result_row.asources > 0 THEN
2392
2393             -- The function that calls this function needs row_number in order
2394             -- to correctly order results from two different runs of this
2395             -- functions.
2396             result_row.row_number := row_number;
2397
2398             -- Now, if row_counter is still less than limit, return a row.  If
2399             -- not, but it is less than next_pivot_pos, continue on without
2400             -- returning actual result rows until we find
2401             -- that next pivot, and return it.
2402
2403             IF row_counter < result_limit THEN
2404                 result_row.browse_entry := rec.id;
2405                 result_row.value := rec.value;
2406
2407                 RETURN NEXT result_row;
2408             ELSE
2409                 result_row.browse_entry := NULL;
2410                 result_row.authorities := NULL;
2411                 result_row.fields := NULL;
2412                 result_row.value := NULL;
2413                 result_row.sources := NULL;
2414                 result_row.sees := NULL;
2415                 result_row.accurate := NULL;
2416                 result_row.aaccurate := NULL;
2417                 result_row.pivot_point := rec.id;
2418
2419                 IF row_counter >= next_pivot_pos THEN
2420                     RETURN NEXT result_row;
2421                     RETURN;
2422                 END IF;
2423             END IF;
2424
2425             IF count_up_from_zero THEN
2426                 row_number := row_number + 1;
2427             ELSE
2428                 row_number := row_number - 1;
2429             END IF;
2430
2431             -- row_counter is different from row_number.
2432             -- It simply counts up from zero so that we know when
2433             -- we've reached our limit.
2434             row_counter := row_counter + 1;
2435         END IF;
2436     END LOOP;
2437 END;
2438 $p$ LANGUAGE PLPGSQL;
2439
2440
2441 CREATE OR REPLACE FUNCTION metabib.browse(
2442     search_field            INT[],
2443     browse_term             TEXT,
2444     context_org             INT DEFAULT NULL,
2445     context_loc_group       INT DEFAULT NULL,
2446     staff                   BOOL DEFAULT FALSE,
2447     pivot_id                BIGINT DEFAULT NULL,
2448     result_limit            INT DEFAULT 10
2449 ) RETURNS SETOF metabib.flat_browse_entry_appearance AS $p$
2450 DECLARE
2451     core_query              TEXT;
2452     back_query              TEXT;
2453     forward_query           TEXT;
2454     pivot_sort_value        TEXT;
2455     pivot_sort_fallback     TEXT;
2456     context_locations       INT[];
2457     browse_superpage_size   INT;
2458     results_skipped         INT := 0;
2459     back_limit              INT;
2460     back_to_pivot           INT;
2461     forward_limit           INT;
2462     forward_to_pivot        INT;
2463 BEGIN
2464     -- First, find the pivot if we were given a browse term but not a pivot.
2465     IF pivot_id IS NULL THEN
2466         pivot_id := metabib.browse_pivot(search_field, browse_term);
2467     END IF;
2468
2469     SELECT INTO pivot_sort_value, pivot_sort_fallback
2470         sort_value, value FROM metabib.browse_entry WHERE id = pivot_id;
2471
2472     -- Bail if we couldn't find a pivot.
2473     IF pivot_sort_value IS NULL THEN
2474         RETURN;
2475     END IF;
2476
2477     -- Transform the context_loc_group argument (if any) (logc at the
2478     -- TPAC layer) into a form we'll be able to use.
2479     IF context_loc_group IS NOT NULL THEN
2480         SELECT INTO context_locations ARRAY_AGG(location)
2481             FROM asset.copy_location_group_map
2482             WHERE lgroup = context_loc_group;
2483     END IF;
2484
2485     -- Get the configured size of browse superpages.
2486     SELECT INTO browse_superpage_size value     -- NULL ok
2487         FROM config.global_flag
2488         WHERE enabled AND name = 'opac.browse.holdings_visibility_test_limit';
2489
2490     -- First we're going to search backward from the pivot, then we're going
2491     -- to search forward.  In each direction, we need two limits.  At the
2492     -- lesser of the two limits, we delineate the edge of the result set
2493     -- we're going to return.  At the greater of the two limits, we find the
2494     -- pivot value that would represent an offset from the current pivot
2495     -- at a distance of one "page" in either direction, where a "page" is a
2496     -- result set of the size specified in the "result_limit" argument.
2497     --
2498     -- The two limits in each direction make four derived values in total,
2499     -- and we calculate them now.
2500     back_limit := CEIL(result_limit::FLOAT / 2);
2501     back_to_pivot := result_limit;
2502     forward_limit := result_limit / 2;
2503     forward_to_pivot := result_limit - 1;
2504
2505     -- This is the meat of the SQL query that finds browse entries.  We'll
2506     -- pass this to a function which uses it with a cursor, so that individual
2507     -- rows may be fetched in a loop until some condition is satisfied, without
2508     -- waiting for a result set of fixed size to be collected all at once.
2509     core_query := '
2510 SELECT  mbe.id,
2511         mbe.value,
2512         mbe.sort_value
2513   FROM  metabib.browse_entry mbe
2514   WHERE (
2515             EXISTS ( -- are there any bibs using this mbe via the requested fields?
2516                 SELECT  1
2517                   FROM  metabib.browse_entry_def_map mbedm
2518                   WHERE mbedm.entry = mbe.id AND mbedm.def = ANY(' || quote_literal(search_field) || ')
2519                   LIMIT 1
2520             ) OR EXISTS ( -- are there any authorities using this mbe via the requested fields?
2521                 SELECT  1
2522                   FROM  metabib.browse_entry_simple_heading_map mbeshm
2523                         JOIN authority.simple_heading ash ON ( mbeshm.simple_heading = ash.id )
2524                         JOIN authority.control_set_auth_field_metabib_field_map_refs map ON (
2525                             ash.atag = map.authority_field
2526                             AND map.metabib_field = ANY(' || quote_literal(search_field) || ')
2527                         )
2528                   WHERE mbeshm.entry = mbe.id
2529             )
2530         ) AND ';
2531
2532     -- This is the variant of the query for browsing backward.
2533     back_query := core_query ||
2534         ' mbe.sort_value <= ' || quote_literal(pivot_sort_value) ||
2535     ' ORDER BY mbe.sort_value DESC, mbe.value DESC ';
2536
2537     -- This variant browses forward.
2538     forward_query := core_query ||
2539         ' mbe.sort_value > ' || quote_literal(pivot_sort_value) ||
2540     ' ORDER BY mbe.sort_value, mbe.value ';
2541
2542     -- We now call the function which applies a cursor to the provided
2543     -- queries, stopping at the appropriate limits and also giving us
2544     -- the next page's pivot.
2545     RETURN QUERY
2546         SELECT * FROM metabib.staged_browse(
2547             back_query, search_field, context_org, context_locations,
2548             staff, browse_superpage_size, TRUE, back_limit, back_to_pivot
2549         ) UNION
2550         SELECT * FROM metabib.staged_browse(
2551             forward_query, search_field, context_org, context_locations,
2552             staff, browse_superpage_size, FALSE, forward_limit, forward_to_pivot
2553         ) ORDER BY row_number DESC;
2554
2555 END;
2556 $p$ LANGUAGE PLPGSQL;
2557
2558 CREATE OR REPLACE FUNCTION metabib.browse(
2559     search_class        TEXT,
2560     browse_term         TEXT,
2561     context_org         INT DEFAULT NULL,
2562     context_loc_group   INT DEFAULT NULL,
2563     staff               BOOL DEFAULT FALSE,
2564     pivot_id            BIGINT DEFAULT NULL,
2565     result_limit        INT DEFAULT 10
2566 ) RETURNS SETOF metabib.flat_browse_entry_appearance AS $p$
2567 BEGIN
2568     RETURN QUERY SELECT * FROM metabib.browse(
2569         (SELECT COALESCE(ARRAY_AGG(id), ARRAY[]::INT[])
2570             FROM config.metabib_field WHERE field_class = search_class),
2571         browse_term,
2572         context_org,
2573         context_loc_group,
2574         staff,
2575         pivot_id,
2576         result_limit
2577     );
2578 END;
2579 $p$ LANGUAGE PLPGSQL;
2580
2581 COMMIT;
2582