Open-ILS/src/sql/Pg/upgrade/YYYY.function.qp_search.sql

   1 /*
   2  * Copyright (C) 2016 Equinox Software, Inc.
   3  * Mike Rylander <miker@esilibrary.com>
   4  *
   5  * This program is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU General Public License
   7  * as published by the Free Software Foundation; either version 2
   8  * of the License, or (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  *
  15  */
  16
  17
  18 BEGIN;
  19
  20 ALTER TYPE search.search_result ADD ATTRIBUTE badges TEXT, ADD ATTRIBUTE popularity NUMERIC;
  21
  22 CREATE OR REPLACE FUNCTION search.query_parser_fts (
  23
  24     param_search_ou INT,
  25     param_depth     INT,
  26     param_query     TEXT,
  27     param_statuses  INT[],
  28     param_locations INT[],
  29     param_offset    INT,
  30     param_check     INT,
  31     param_limit     INT,
  32     metarecord      BOOL,
  33     staff           BOOL,
  34     deleted_search  BOOL,
  35     param_pref_ou   INT DEFAULT NULL
  36 ) RETURNS SETOF search.search_result AS $func$
  37 DECLARE
  38
  39     current_res         search.search_result%ROWTYPE;
  40     search_org_list     INT[];
  41     luri_org_list       INT[];
  42     tmp_int_list        INT[];
  43
  44     check_limit         INT;
  45     core_limit          INT;
  46     core_offset         INT;
  47     tmp_int             INT;
  48
  49     core_result         RECORD;
  50     core_cursor         REFCURSOR;
  51     core_rel_query      TEXT;
  52
  53     total_count         INT := 0;
  54     check_count         INT := 0;
  55     deleted_count       INT := 0;
  56     visible_count       INT := 0;
  57     excluded_count      INT := 0;
  58
  59     luri_as_copy        BOOL;
  60 BEGIN
  61
  62     check_limit := COALESCE( param_check, 1000 );
  63     core_limit  := COALESCE( param_limit, 25000 );
  64     core_offset := COALESCE( param_offset, 0 );
  65
  66     SELECT COALESCE( enabled, FALSE ) INTO luri_as_copy FROM config.global_flag WHERE name = 'opac.located_uri.act_as_copy';
  67
  68     -- core_skip_chk := COALESCE( param_skip_chk, 1 );
  69
  70     IF param_search_ou > 0 THEN
  71         IF param_depth IS NOT NULL THEN
  72             SELECT ARRAY_AGG(distinct id) INTO search_org_list FROM actor.org_unit_descendants( param_search_ou, param_depth );
  73         ELSE
  74             SELECT ARRAY_AGG(distinct id) INTO search_org_list FROM actor.org_unit_descendants( param_search_ou );
  75         END IF;
  76
  77         IF luri_as_copy THEN
  78             SELECT ARRAY_AGG(distinct id) INTO luri_org_list FROM actor.org_unit_full_path( param_search_ou );
  79         ELSE
  80             SELECT ARRAY_AGG(distinct id) INTO luri_org_list FROM actor.org_unit_ancestors( param_search_ou );
  81         END IF;
  82
  83     ELSIF param_search_ou < 0 THEN
  84         SELECT ARRAY_AGG(distinct org_unit) INTO search_org_list FROM actor.org_lasso_map WHERE lasso = -param_search_ou;
  85
  86         FOR tmp_int IN SELECT * FROM UNNEST(search_org_list) LOOP
  87
  88             IF luri_as_copy THEN
  89                 SELECT ARRAY_AGG(distinct id) INTO tmp_int_list FROM actor.org_unit_full_path( tmp_int );
  90             ELSE
  91                 SELECT ARRAY_AGG(distinct id) INTO tmp_int_list FROM actor.org_unit_ancestors( tmp_int );
  92             END IF;
  93
  94             luri_org_list := luri_org_list || tmp_int_list;
  95         END LOOP;
  96
  97         SELECT ARRAY_AGG(DISTINCT x.id) INTO luri_org_list FROM UNNEST(luri_org_list) x(id);
  98
  99     ELSIF param_search_ou = 0 THEN
 100         -- reserved for user lassos (ou_buckets/type='lasso') with ID passed in depth ... hack? sure.
 101     END IF;
 102
 103     IF param_pref_ou IS NOT NULL THEN
 104             IF luri_as_copy THEN
 105                 SELECT ARRAY_AGG(distinct id) INTO tmp_int_list FROM actor.org_unit_full_path( param_pref_ou );
 106             ELSE
 107                 SELECT ARRAY_AGG(distinct id) INTO tmp_int_list FROM actor.org_unit_ancestors( param_pref_ou );
 108             END IF;
 109
 110         luri_org_list := luri_org_list || tmp_int_list;
 111     END IF;
 112
 113     OPEN core_cursor FOR EXECUTE param_query;
 114
 115     LOOP
 116
 117         FETCH core_cursor INTO core_result;
 118         EXIT WHEN NOT FOUND;
 119         EXIT WHEN total_count >= core_limit;
 120
 121         total_count := total_count + 1;
 122
 123         CONTINUE WHEN total_count NOT BETWEEN  core_offset + 1 AND check_limit + core_offset;
 124
 125         check_count := check_count + 1;
 126
 127         IF NOT deleted_search THEN
 128
 129             PERFORM 1 FROM biblio.record_entry b WHERE NOT b.deleted AND b.id IN ( SELECT * FROM unnest( core_result.records ) );
 130             IF NOT FOUND THEN
 131                 -- RAISE NOTICE ' % were all deleted ... ', core_result.records;
 132                 deleted_count := deleted_count + 1;
 133                 CONTINUE;
 134             END IF;
 135
 136             PERFORM 1
 137               FROM  biblio.record_entry b
 138                     JOIN config.bib_source s ON (b.source = s.id)
 139               WHERE s.transcendant
 140                     AND b.id IN ( SELECT * FROM unnest( core_result.records ) );
 141
 142             IF FOUND THEN
 143                 -- RAISE NOTICE ' % were all transcendant ... ', core_result.records;
 144                 visible_count := visible_count + 1;
 145
 146                 current_res.id = core_result.id;
 147                 current_res.rel = core_result.rel;
 148                 current_res.badges = core_result.badges;
 149                 current_res.popularity = core_result.popularity;
 150
 151                 tmp_int := 1;
 152                 IF metarecord THEN
 153                     SELECT COUNT(DISTINCT s.source) INTO tmp_int FROM metabib.metarecord_source_map s WHERE s.metarecord = core_result.id;
 154                 END IF;
 155
 156                 IF tmp_int = 1 THEN
 157                     current_res.record = core_result.records[1];
 158                 ELSE
 159                     current_res.record = NULL;
 160                 END IF;
 161
 162                 RETURN NEXT current_res;
 163
 164                 CONTINUE;
 165             END IF;
 166
 167             PERFORM 1
 168               FROM  asset.call_number cn
 169                     JOIN asset.uri_call_number_map map ON (map.call_number = cn.id)
 170                     JOIN asset.uri uri ON (map.uri = uri.id)
 171               WHERE NOT cn.deleted
 172                     AND cn.label = '##URI##'
 173                     AND uri.active
 174                     AND ( param_locations IS NULL OR array_upper(param_locations, 1) IS NULL )
 175                     AND cn.record IN ( SELECT * FROM unnest( core_result.records ) )
 176                     AND cn.owning_lib IN ( SELECT * FROM unnest( luri_org_list ) )
 177               LIMIT 1;
 178
 179             IF FOUND THEN
 180                 -- RAISE NOTICE ' % have at least one URI ... ', core_result.records;
 181                 visible_count := visible_count + 1;
 182
 183                 current_res.id = core_result.id;
 184                 current_res.rel = core_result.rel;
 185                 current_res.badges = core_result.badges;
 186                 current_res.popularity = core_result.popularity;
 187
 188                 tmp_int := 1;
 189                 IF metarecord THEN
 190                     SELECT COUNT(DISTINCT s.source) INTO tmp_int FROM metabib.metarecord_source_map s WHERE s.metarecord = core_result.id;
 191                 END IF;
 192
 193                 IF tmp_int = 1 THEN
 194                     current_res.record = core_result.records[1];
 195                 ELSE
 196                     current_res.record = NULL;
 197                 END IF;
 198
 199                 RETURN NEXT current_res;
 200
 201                 CONTINUE;
 202             END IF;
 203
 204             IF param_statuses IS NOT NULL AND array_upper(param_statuses, 1) > 0 THEN
 205
 206                 PERFORM 1
 207                   FROM  asset.call_number cn
 208                         JOIN asset.copy cp ON (cp.call_number = cn.id)
 209                   WHERE NOT cn.deleted
 210                         AND NOT cp.deleted
 211                         AND cp.status IN ( SELECT * FROM unnest( param_statuses ) )
 212                         AND cn.record IN ( SELECT * FROM unnest( core_result.records ) )
 213                         AND cp.circ_lib IN ( SELECT * FROM unnest( search_org_list ) )
 214                   LIMIT 1;
 215
 216                 IF NOT FOUND THEN
 217                     PERFORM 1
 218                       FROM  biblio.peer_bib_copy_map pr
 219                             JOIN asset.copy cp ON (cp.id = pr.target_copy)
 220                       WHERE NOT cp.deleted
 221                             AND cp.status IN ( SELECT * FROM unnest( param_statuses ) )
 222                             AND pr.peer_record IN ( SELECT * FROM unnest( core_result.records ) )
 223                             AND cp.circ_lib IN ( SELECT * FROM unnest( search_org_list ) )
 224                       LIMIT 1;
 225
 226                     IF NOT FOUND THEN
 227                     -- RAISE NOTICE ' % and multi-home linked records were all status-excluded ... ', core_result.records;
 228                         excluded_count := excluded_count + 1;
 229                         CONTINUE;
 230                     END IF;
 231                 END IF;
 232
 233             END IF;
 234
 235             IF param_locations IS NOT NULL AND array_upper(param_locations, 1) > 0 THEN
 236
 237                 PERFORM 1
 238                   FROM  asset.call_number cn
 239                         JOIN asset.copy cp ON (cp.call_number = cn.id)
 240                   WHERE NOT cn.deleted
 241                         AND NOT cp.deleted
 242                         AND cp.location IN ( SELECT * FROM unnest( param_locations ) )
 243                         AND cn.record IN ( SELECT * FROM unnest( core_result.records ) )
 244                         AND cp.circ_lib IN ( SELECT * FROM unnest( search_org_list ) )
 245                   LIMIT 1;
 246
 247                 IF NOT FOUND THEN
 248                     PERFORM 1
 249                       FROM  biblio.peer_bib_copy_map pr
 250                             JOIN asset.copy cp ON (cp.id = pr.target_copy)
 251                       WHERE NOT cp.deleted
 252                             AND cp.location IN ( SELECT * FROM unnest( param_locations ) )
 253                             AND pr.peer_record IN ( SELECT * FROM unnest( core_result.records ) )
 254                             AND cp.circ_lib IN ( SELECT * FROM unnest( search_org_list ) )
 255                       LIMIT 1;
 256
 257                     IF NOT FOUND THEN
 258                         -- RAISE NOTICE ' % and multi-home linked records were all copy_location-excluded ... ', core_result.records;
 259                         excluded_count := excluded_count + 1;
 260                         CONTINUE;
 261                     END IF;
 262                 END IF;
 263
 264             END IF;
 265
 266             IF staff IS NULL OR NOT staff THEN
 267
 268                 PERFORM 1
 269                   FROM  asset.opac_visible_copies
 270                   WHERE circ_lib IN ( SELECT * FROM unnest( search_org_list ) )
 271                         AND record IN ( SELECT * FROM unnest( core_result.records ) )
 272                   LIMIT 1;
 273
 274                 IF NOT FOUND THEN
 275                     PERFORM 1
 276                       FROM  biblio.peer_bib_copy_map pr
 277                             JOIN asset.opac_visible_copies cp ON (cp.copy_id = pr.target_copy)
 278                       WHERE cp.circ_lib IN ( SELECT * FROM unnest( search_org_list ) )
 279                             AND pr.peer_record IN ( SELECT * FROM unnest( core_result.records ) )
 280                       LIMIT 1;
 281
 282                     IF NOT FOUND THEN
 283
 284                         -- RAISE NOTICE ' % and multi-home linked records were all visibility-excluded ... ', core_result.records;
 285                         excluded_count := excluded_count + 1;
 286                         CONTINUE;
 287                     END IF;
 288                 END IF;
 289
 290             ELSE
 291
 292                 PERFORM 1
 293                   FROM  asset.call_number cn
 294                         JOIN asset.copy cp ON (cp.call_number = cn.id)
 295                   WHERE NOT cn.deleted
 296                         AND NOT cp.deleted
 297                         AND cp.circ_lib IN ( SELECT * FROM unnest( search_org_list ) )
 298                         AND cn.record IN ( SELECT * FROM unnest( core_result.records ) )
 299                   LIMIT 1;
 300
 301                 IF NOT FOUND THEN
 302
 303                     PERFORM 1
 304                       FROM  biblio.peer_bib_copy_map pr
 305                             JOIN asset.copy cp ON (cp.id = pr.target_copy)
 306                       WHERE NOT cp.deleted
 307                             AND cp.circ_lib IN ( SELECT * FROM unnest( search_org_list ) )
 308                             AND pr.peer_record IN ( SELECT * FROM unnest( core_result.records ) )
 309                       LIMIT 1;
 310
 311                     IF NOT FOUND THEN
 312
 313                         PERFORM 1
 314                           FROM  asset.call_number cn
 315                                 JOIN asset.copy cp ON (cp.call_number = cn.id)
 316                           WHERE cn.record IN ( SELECT * FROM unnest( core_result.records ) )
 317                                 AND NOT cp.deleted
 318                           LIMIT 1;
 319
 320                         IF NOT FOUND THEN
 321                             -- Recheck Located URI visibility in the case of no "foreign" copies
 322                             PERFORM 1
 323                               FROM  asset.call_number cn
 324                                     JOIN asset.uri_call_number_map map ON (map.call_number = cn.id)
 325                                     JOIN asset.uri uri ON (map.uri = uri.id)
 326                               WHERE NOT cn.deleted
 327                                     AND cn.label = '##URI##'
 328                                     AND uri.active
 329                                     AND cn.record IN ( SELECT * FROM unnest( core_result.records ) )
 330                                     AND cn.owning_lib NOT IN ( SELECT * FROM unnest( luri_org_list ) )
 331                               LIMIT 1;
 332
 333                             IF FOUND THEN
 334                                 -- RAISE NOTICE ' % were excluded for foreign located URIs... ', core_result.records;
 335                                 excluded_count := excluded_count + 1;
 336                                 CONTINUE;
 337                             END IF;
 338                         ELSE
 339                             -- RAISE NOTICE ' % and multi-home linked records were all visibility-excluded ... ', core_result.records;
 340                             excluded_count := excluded_count + 1;
 341                             CONTINUE;
 342                         END IF;
 343                     END IF;
 344
 345                 END IF;
 346
 347             END IF;
 348
 349         END IF;
 350
 351         visible_count := visible_count + 1;
 352
 353         current_res.id = core_result.id;
 354         current_res.rel = core_result.rel;
 355         current_res.badges = core_result.badges;
 356         current_res.popularity = core_result.popularity;
 357
 358         tmp_int := 1;
 359         IF metarecord THEN
 360             SELECT COUNT(DISTINCT s.source) INTO tmp_int FROM metabib.metarecord_source_map s WHERE s.metarecord = core_result.id;
 361         END IF;
 362
 363         IF tmp_int = 1 THEN
 364             current_res.record = core_result.records[1];
 365         ELSE
 366             current_res.record = NULL;
 367         END IF;
 368
 369         RETURN NEXT current_res;
 370
 371         IF visible_count % 1000 = 0 THEN
 372             -- RAISE NOTICE ' % visible so far ... ', visible_count;
 373         END IF;
 374
 375     END LOOP;
 376
 377     current_res.id = NULL;
 378     current_res.rel = NULL;
 379     current_res.record = NULL;
 380     current_res.badges = NULL;
 381     current_res.popularity = NULL;
 382     current_res.total = total_count;
 383     current_res.checked = check_count;
 384     current_res.deleted = deleted_count;
 385     current_res.visible = visible_count;
 386     current_res.excluded = excluded_count;
 387
 388     CLOSE core_cursor;
 389
 390     RETURN NEXT current_res;
 391
 392 END;
 393 $func$ LANGUAGE PLPGSQL;
 394
 395 CREATE OR REPLACE FUNCTION metabib.staged_browse(
 396     query                   TEXT,
 397     fields                  INT[],
 398     context_org             INT,
 399     context_locations       INT[],
 400     staff                   BOOL,
 401     browse_superpage_size   INT,
 402     count_up_from_zero      BOOL,   -- if false, count down from -1
 403     result_limit            INT,
 404     next_pivot_pos          INT
 405 ) RETURNS SETOF metabib.flat_browse_entry_appearance AS $p$
 406 DECLARE
 407     curs                    REFCURSOR;
 408     rec                     RECORD;
 409     qpfts_query             TEXT;
 410     aqpfts_query            TEXT;
 411     afields                 INT[];
 412     bfields                 INT[];
 413     result_row              metabib.flat_browse_entry_appearance%ROWTYPE;
 414     results_skipped         INT := 0;
 415     row_counter             INT := 0;
 416     row_number              INT;
 417     slice_start             INT;
 418     slice_end               INT;
 419     full_end                INT;
 420     all_records             BIGINT[];
 421     all_brecords             BIGINT[];
 422     all_arecords            BIGINT[];
 423     superpage_of_records    BIGINT[];
 424     superpage_size          INT;
 425 BEGIN
 426     IF count_up_from_zero THEN
 427         row_number := 0;
 428     ELSE
 429         row_number := -1;
 430     END IF;
 431
 432     OPEN curs FOR EXECUTE query;
 433
 434     LOOP
 435         FETCH curs INTO rec;
 436         IF NOT FOUND THEN
 437             IF result_row.pivot_point IS NOT NULL THEN
 438                 RETURN NEXT result_row;
 439             END IF;
 440             RETURN;
 441         END IF;
 442
 443
 444         -- Gather aggregate data based on the MBE row we're looking at now, authority axis
 445         SELECT INTO all_arecords, result_row.sees, afields
 446                 ARRAY_AGG(DISTINCT abl.bib), -- bibs to check for visibility
 447                 STRING_AGG(DISTINCT aal.source::TEXT, $$,$$), -- authority record ids
 448                 ARRAY_AGG(DISTINCT map.metabib_field) -- authority-tag-linked CMF rows
 449
 450           FROM  metabib.browse_entry_simple_heading_map mbeshm
 451                 JOIN authority.simple_heading ash ON ( mbeshm.simple_heading = ash.id )
 452                 JOIN authority.authority_linking aal ON ( ash.record = aal.source )
 453                 JOIN authority.bib_linking abl ON ( aal.target = abl.authority )
 454                 JOIN authority.control_set_auth_field_metabib_field_map_refs map ON (
 455                     ash.atag = map.authority_field
 456                     AND map.metabib_field = ANY(fields)
 457                 )
 458           WHERE mbeshm.entry = rec.id;
 459
 460
 461         -- Gather aggregate data based on the MBE row we're looking at now, bib axis
 462         SELECT INTO all_brecords, result_row.authorities, bfields
 463                 ARRAY_AGG(DISTINCT source),
 464                 STRING_AGG(DISTINCT authority::TEXT, $$,$$),
 465                 ARRAY_AGG(DISTINCT def)
 466           FROM  metabib.browse_entry_def_map
 467           WHERE entry = rec.id
 468                 AND def = ANY(fields);
 469
 470         SELECT INTO result_row.fields STRING_AGG(DISTINCT x::TEXT, $$,$$) FROM UNNEST(afields || bfields) x;
 471
 472         result_row.sources := 0;
 473         result_row.asources := 0;
 474
 475         -- Bib-linked vis checking
 476         IF ARRAY_UPPER(all_brecords,1) IS NOT NULL THEN
 477
 478             full_end := ARRAY_LENGTH(all_brecords, 1);
 479             superpage_size := COALESCE(browse_superpage_size, full_end);
 480             slice_start := 1;
 481             slice_end := superpage_size;
 482
 483             WHILE result_row.sources = 0 AND slice_start <= full_end LOOP
 484                 superpage_of_records := all_brecords[slice_start:slice_end];
 485                 qpfts_query :=
 486                     'SELECT NULL::BIGINT AS id, ARRAY[r] AS records, ' ||
 487                     'NULL AS badges, NULL::NUMERIC AS popularity, ' ||
 488                     '1::NUMERIC AS rel FROM (SELECT UNNEST(' ||
 489                     quote_literal(superpage_of_records) || '::BIGINT[]) AS r) rr';
 490
 491                 -- We use search.query_parser_fts() for visibility testing.
 492                 -- We're calling it once per browse-superpage worth of records
 493                 -- out of the set of records related to a given mbe, until we've
 494                 -- either exhausted that set of records or found at least 1
 495                 -- visible record.
 496
 497                 SELECT INTO result_row.sources visible
 498                     FROM search.query_parser_fts(
 499                         context_org, NULL, qpfts_query, NULL,
 500                         context_locations, 0, NULL, NULL, FALSE, staff, FALSE
 501                     ) qpfts
 502                     WHERE qpfts.rel IS NULL;
 503
 504                 slice_start := slice_start + superpage_size;
 505                 slice_end := slice_end + superpage_size;
 506             END LOOP;
 507
 508             -- Accurate?  Well, probably.
 509             result_row.accurate := browse_superpage_size IS NULL OR
 510                 browse_superpage_size >= full_end;
 511
 512         END IF;
 513
 514         -- Authority-linked vis checking
 515         IF ARRAY_UPPER(all_arecords,1) IS NOT NULL THEN
 516
 517             full_end := ARRAY_LENGTH(all_arecords, 1);
 518             superpage_size := COALESCE(browse_superpage_size, full_end);
 519             slice_start := 1;
 520             slice_end := superpage_size;
 521
 522             WHILE result_row.asources = 0 AND slice_start <= full_end LOOP
 523                 superpage_of_records := all_arecords[slice_start:slice_end];
 524                 qpfts_query :=
 525                     'SELECT NULL::BIGINT AS id, ARRAY[r] AS records, ' ||
 526                     'NULL AS badges, NULL::NUMERIC AS popularity, ' ||
 527                     '1::NUMERIC AS rel FROM (SELECT UNNEST(' ||
 528                     quote_literal(superpage_of_records) || '::BIGINT[]) AS r) rr';
 529
 530                 -- We use search.query_parser_fts() for visibility testing.
 531                 -- We're calling it once per browse-superpage worth of records
 532                 -- out of the set of records related to a given mbe, via
 533                 -- authority until we've either exhausted that set of records
 534                 -- or found at least 1 visible record.
 535
 536                 SELECT INTO result_row.asources visible
 537                     FROM search.query_parser_fts(
 538                         context_org, NULL, qpfts_query, NULL,
 539                         context_locations, 0, NULL, NULL, FALSE, staff, FALSE
 540                     ) qpfts
 541                     WHERE qpfts.rel IS NULL;
 542
 543                 slice_start := slice_start + superpage_size;
 544                 slice_end := slice_end + superpage_size;
 545             END LOOP;
 546
 547
 548             -- Accurate?  Well, probably.
 549             result_row.aaccurate := browse_superpage_size IS NULL OR
 550                 browse_superpage_size >= full_end;
 551
 552         END IF;
 553
 554         IF result_row.sources > 0 OR result_row.asources > 0 THEN
 555
 556             -- The function that calls this function needs row_number in order
 557             -- to correctly order results from two different runs of this
 558             -- functions.
 559             result_row.row_number := row_number;
 560
 561             -- Now, if row_counter is still less than limit, return a row.  If
 562             -- not, but it is less than next_pivot_pos, continue on without
 563             -- returning actual result rows until we find
 564             -- that next pivot, and return it.
 565
 566             IF row_counter < result_limit THEN
 567                 result_row.browse_entry := rec.id;
 568                 result_row.value := rec.value;
 569
 570                 RETURN NEXT result_row;
 571             ELSE
 572                 result_row.browse_entry := NULL;
 573                 result_row.authorities := NULL;
 574                 result_row.fields := NULL;
 575                 result_row.value := NULL;
 576                 result_row.sources := NULL;
 577                 result_row.sees := NULL;
 578                 result_row.accurate := NULL;
 579                 result_row.aaccurate := NULL;
 580                 result_row.pivot_point := rec.id;
 581
 582                 IF row_counter >= next_pivot_pos THEN
 583                     RETURN NEXT result_row;
 584                     RETURN;
 585                 END IF;
 586             END IF;
 587
 588             IF count_up_from_zero THEN
 589                 row_number := row_number + 1;
 590             ELSE
 591                 row_number := row_number - 1;
 592             END IF;
 593
 594             -- row_counter is different from row_number.
 595             -- It simply counts up from zero so that we know when
 596             -- we've reached our limit.
 597             row_counter := row_counter + 1;
 598         END IF;
 599     END LOOP;
 600 END;
 601 $p$ LANGUAGE PLPGSQL;
 602
 603 COMMIT;
 604