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