granicus.if.org Git - postgresql/blob - src/backend/utils/adt/arrayfuncs.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * arrayfuncs.c--
   4  *    Special functions for arrays.
   5  *
   6  * Copyright (c) 1994, Regents of the University of California
   7  *
   8  *
   9  * IDENTIFICATION
  10  *    $Header: /cvsroot/pgsql/src/backend/utils/adt/arrayfuncs.c,v 1.10 1996/11/08 05:59:38 momjian Exp $
  11  *
  12  *-------------------------------------------------------------------------
  13  */
  14
  15 #include <ctype.h>
  16 #include <stdio.h>
  17 #include <string.h>
  18
  19 #include "postgres.h"
  20
  21 #include "catalog/catalog.h"
  22 #include "catalog/pg_type.h"
  23
  24 #include "utils/syscache.h"
  25 #include "utils/memutils.h"
  26 #include "storage/fd.h"         /* for SEEK_ */
  27 #include "fmgr.h"
  28 #include "utils/array.h"
  29
  30 #include "libpq/libpq-fs.h"
  31 #include "libpq/be-fsstubs.h"
  32
  33 #define ASSGN    "="
  34
  35 /* An array has the following internal structure:
  36  *    <nbytes>            - total number of bytes
  37  *     <ndim>                - number of dimensions of the array
  38  *    <flags>                - bit mask of flags
  39  *    <dim>                - size of each array axis
  40  *    <dim_lower>            - lower boundary of each dimension
  41  *    <actual data>        - whatever is the stored data
  42 */
  43
  44 /*-=-=--=-=-=-=-=-=-=-=--=-=-=-=-=-=-=-=--=-=-=-=-=-=-=-=--=-=-=-=-=-=-=-*/
  45 static int _ArrayCount(char *str, int dim[], int typdelim);
  46 static char *_ReadArrayStr(char *arrayStr, int nitems, int ndim, int dim[],
  47                            func_ptr inputproc, Oid typelem, char typdelim,
  48                            int typlen, bool typbyval, char typalign,
  49                            int *nbytes);
  50 #ifdef LOARRAY
  51 static char *_ReadLOArray(char *str, int *nbytes, int *fd, bool *chunkFlag,
  52                           int ndim, int dim[], int baseSize);
  53 #endif
  54 static void _CopyArrayEls(char **values, char *p, int nitems, int typlen,
  55                           char typalign, bool typbyval);
  56 static void system_cache_lookup(Oid element_type, bool input, int *typlen,
  57                 bool *typbyval, char *typdelim, Oid *typelem, Oid *proc,
  58                 char *typalign);
  59 static Datum _ArrayCast(char *value, bool byval, int len);
  60 #ifdef LOARRAY
  61 static char *_AdvanceBy1word(char *str, char **word);
  62 #endif
  63 static void _ArrayRange(int st[], int endp[], int bsize, char *destPtr,
  64                         ArrayType *array, int from);
  65 static int _ArrayClipCount(int stI[], int endpI[], ArrayType *array);
  66 static void _LOArrayRange(int st[], int endp[], int bsize, int srcfd,
  67           int destfd, ArrayType *array, int isSrcLO, bool *isNull);
  68 static void _ReadArray (int st[], int endp[], int bsize, int srcfd, int destfd,
  69                        ArrayType *array, int isDestLO, bool *isNull);
  70 static ArrayCastAndSet(char *src, bool typbyval, int typlen, char *dest);
  71
  72
  73 /*---------------------------------------------------------------------
  74  * array_in :
  75  *        converts an array from the external format in "string" to
  76  *        it internal format.
  77  * return value :
  78  *        the internal representation of the input array
  79  *--------------------------------------------------------------------
  80  */
  81 char *
  82 array_in(char *string,          /* input array in external form */
  83          Oid element_type)      /* type OID of an array element */
  84 {
  85     int typlen;
  86     bool typbyval, done;
  87     char typdelim;
  88     Oid typinput;
  89     Oid typelem;
  90     char *string_save, *p, *q, *r;
  91     func_ptr inputproc;
  92     int i, nitems, dummy;
  93     int32 nbytes;
  94     char *dataPtr;
  95     ArrayType *retval = NULL;
  96     int ndim, dim[MAXDIM], lBound[MAXDIM];
  97     char typalign;
  98
  99     system_cache_lookup(element_type, true, &typlen, &typbyval, &typdelim,
 100                         &typelem, &typinput, &typalign);
 101
 102     fmgr_info(typinput, &inputproc, &dummy);
 103
 104     string_save = (char *) palloc(strlen(string) + 3);
 105     strcpy(string_save, string);
 106
 107     /* --- read array dimensions  ---------- */
 108     p = q = string_save; done = false;
 109     for ( ndim = 0; !done; ) {
 110         while (isspace(*p)) p++;
 111         if (*p == '[' ) {
 112             p++;
 113             if ((r = (char *)strchr(p, ':')) == (char *)NULL)
 114                 lBound[ndim] = 1;
 115             else {
 116                 *r = '\0';
 117                 lBound[ndim] = atoi(p);
 118                 p = r + 1;
 119             }
 120             for (q = p; isdigit(*q); q++);
 121             if (*q != ']')
 122                 elog(WARN, "array_in: missing ']' in array declaration");
 123             *q = '\0';
 124             dim[ndim] = atoi(p);
 125             if ((dim[ndim] < 0) || (lBound[ndim] < 0))
 126                 elog(WARN,"array_in: array dimensions need to be positive");
 127             dim[ndim] = dim[ndim] - lBound[ndim] + 1;
 128             if (dim[ndim] < 0)
 129                 elog(WARN, "array_in: upper_bound cannot be < lower_bound");
 130             p = q + 1; ndim++;
 131         } else {
 132             done = true;
 133         }
 134     }
 135
 136     if (ndim == 0) {
 137         if (*p == '{') {
 138             ndim = _ArrayCount(p, dim, typdelim);
 139             for (i = 0; i < ndim; lBound[i++] = 1);
 140         } else {
 141             elog(WARN,"array_in: Need to specify dimension");
 142         }
 143     } else {
 144         while (isspace(*p)) p++;
 145         if (strncmp(p, ASSGN, strlen(ASSGN)))
 146             elog(WARN, "array_in: missing assignment operator");
 147         p+= strlen(ASSGN);
 148         while (isspace(*p)) p++;
 149     }
 150
 151     nitems = getNitems( ndim, dim);
 152     if (nitems == 0) {
 153         char *emptyArray = palloc(sizeof(ArrayType));
 154         memset(emptyArray, 0, sizeof(ArrayType));
 155         * (int32 *) emptyArray = sizeof(ArrayType);
 156         return emptyArray;
 157     }
 158
 159     if (*p == '{') {
 160         /* array not a large object */
 161         dataPtr =
 162             (char *) _ReadArrayStr(p, nitems,  ndim, dim, inputproc, typelem,
 163                                    typdelim, typlen, typbyval, typalign,
 164                                    &nbytes );
 165         nbytes += ARR_OVERHEAD(ndim);
 166         retval = (ArrayType *) palloc(nbytes);
 167         memset(retval,0, nbytes);
 168         memmove(retval, (char *)&nbytes, sizeof(int));
 169         memmove((char*)ARR_NDIM_PTR(retval), (char *)&ndim, sizeof(int));
 170         SET_LO_FLAG (false, retval);
 171         memmove((char *)ARR_DIMS(retval), (char *)dim, ndim*sizeof(int));
 172         memmove((char *)ARR_LBOUND(retval), (char *)lBound,
 173                 ndim*sizeof(int));
 174         /* dataPtr is an array of arbitraystuff even though its type is char*
 175            cast to char** to pass to _CopyArrayEls for now  - jolly */
 176         _CopyArrayEls((char**)dataPtr,
 177                       ARR_DATA_PTR(retval), nitems,
 178                       typlen, typalign, typbyval);
 179     } else  {
 180 #ifdef LOARRAY
 181         int dummy, bytes;
 182         bool chunked = false;
 183
 184         dataPtr = _ReadLOArray(p, &bytes, &dummy, &chunked, ndim,
 185                                dim, typlen );
 186         nbytes = bytes + ARR_OVERHEAD(ndim);
 187         retval = (ArrayType *) palloc(nbytes);
 188         memset(retval, 0,nbytes);
 189         memmove(retval, (char *)&nbytes, sizeof(int));
 190         memmove((char *)ARR_NDIM_PTR(retval), (char *)&ndim, sizeof(int));
 191         SET_LO_FLAG (true, retval);
 192         SET_CHUNK_FLAG (chunked, retval);
 193         memmove((char *)ARR_DIMS(retval), (char *)dim, ndim*sizeof(int));
 194         memmove((char *)ARR_LBOUND(retval),(char *)lBound, ndim*sizeof(int));
 195         memmove(ARR_DATA_PTR(retval), dataPtr, bytes);
 196 #endif
 197         elog(WARN, "large object arrays not supported");
 198     }
 199     pfree(string_save);
 200     return((char *)retval);
 201 }
 202
 203 /*-----------------------------------------------------------------------------
 204  * _ArrayCount --
 205  *   Counts the number of dimensions and the dim[] array for an array string.
 206  *       The syntax for array input is C-like nested curly braces
 207  *-----------------------------------------------------------------------------
 208  */
 209 static int
 210 _ArrayCount(char *str, int dim[], int typdelim)
 211 {
 212     int nest_level = 0, i;
 213     int ndim = 0, temp[MAXDIM];
 214     bool scanning_string = false;
 215     bool eoArray = false;
 216     char *q;
 217
 218     for (i = 0; i < MAXDIM; ++i) {
 219         temp[i] = dim[i] = 0;
 220     }
 221
 222     if (strncmp (str, "{}", 2) == 0) return(0);
 223
 224     q = str;
 225     while (eoArray != true) {
 226         bool done = false;
 227         while (!done) {
 228             switch (*q) {
 229             case '\\':
 230                 /* skip escaped characters (\ and ") inside strings */
 231                 if (scanning_string && *(q+1)) {
 232                     q++;
 233                 }
 234                 break;
 235             case '\0':
 236                 /* Signal a premature end of the string.  DZ - 2-9-1996 */
 237                 elog(WARN, "malformed array constant: %s", str);
 238                 break;
 239             case '\"':
 240                 scanning_string = ! scanning_string;
 241                 break;
 242             case '{':
 243                 if (!scanning_string)  {
 244                     temp[nest_level] = 0;
 245                     nest_level++;
 246                 }
 247                 break;
 248             case '}':
 249                 if (!scanning_string) {
 250                     if (!ndim) ndim = nest_level;
 251                     nest_level--;
 252                     if (nest_level) temp[nest_level-1]++;
 253                     if (nest_level == 0) eoArray = done = true;
 254                 }
 255                 break;
 256             default:
 257                 if (!ndim) ndim = nest_level;
 258                 if (*q == typdelim && !scanning_string )
 259                     done = true;
 260                 break;
 261             }
 262             if (!done) q++;
 263         }
 264         temp[ndim-1]++;
 265         q++;
 266         if (!eoArray)
 267             while (isspace(*q)) q++;
 268     }
 269     for (i = 0; i < ndim; ++i) {
 270         dim[i] = temp[i];
 271     }
 272
 273     return(ndim);
 274 }
 275
 276 /*---------------------------------------------------------------------------
 277  * _ReadArrayStr :
 278  *   parses the array string pointed by "arrayStr" and converts it in the
 279  *   internal format. The external format expected is like C array
 280  *   declaration. Unspecified elements are initialized to zero for fixed length
 281  *   base types and to empty varlena structures for variable length base
 282  *   types.
 283  * result :
 284  *   returns the internal representation of the array elements
 285  *   nbytes is set to the size of the array in its internal representation.
 286  *---------------------------------------------------------------------------
 287  */
 288 static char *
 289 _ReadArrayStr(char *arrayStr,
 290               int nitems,
 291               int ndim,
 292               int dim[],
 293               func_ptr inputproc, /* function used for the conversion */
 294               Oid typelem,
 295               char typdelim,
 296               int typlen,
 297               bool typbyval,
 298               char typalign,
 299               int *nbytes)
 300 {
 301     int i, nest_level = 0;
 302     char *p, *q, *r, **values;
 303     bool scanning_string = false;
 304     int indx[MAXDIM], prod[MAXDIM];
 305     bool eoArray = false;
 306
 307     mda_get_prod(ndim, dim, prod);
 308     for (i = 0; i < ndim; indx[i++] = 0);
 309     /* read array enclosed within {} */
 310     values = (char **) palloc(nitems * sizeof(char *));
 311     memset(values, 0, nitems * sizeof(char *));
 312     q = p = arrayStr;
 313
 314     while ( ! eoArray ) {
 315         bool done = false;
 316         int i = -1;
 317
 318         while (!done) {
 319             switch (*q) {
 320             case '\\':
 321                 /* Crunch the string on top of the backslash. */
 322                 for (r = q; *r != '\0'; r++) *r = *(r+1);
 323                 break;
 324             case '\"':
 325                 if (!scanning_string ) {
 326                     while (p != q) p++;
 327                     p++; /* get p past first doublequote */
 328                 } else
 329                     *q = '\0';
 330                 scanning_string = ! scanning_string;
 331                 break;
 332             case '{':
 333                 if (!scanning_string) {
 334                     p++;
 335                     nest_level++;
 336                     if (nest_level > ndim)
 337                         elog(WARN, "array_in: illformed array constant");
 338                     indx[nest_level - 1] = 0;
 339                     indx[ndim - 1] = 0;
 340                 }
 341                 break;
 342             case '}':
 343                 if (!scanning_string) {
 344                     if (i == -1)
 345                         i = tuple2linear(ndim, indx, prod);
 346                     nest_level--;
 347                     if (nest_level == 0)
 348                         eoArray = done = true;
 349                     else {
 350                         *q = '\0';
 351                         indx[nest_level - 1]++;
 352                     }
 353                 }
 354                 break;
 355             default:
 356                 if (*q == typdelim && !scanning_string ) {
 357                     if (i == -1)
 358                         i = tuple2linear(ndim, indx, prod);
 359                     done = true;
 360                     indx[ndim - 1]++;
 361                 }
 362                 break;
 363             }
 364             if (!done)
 365                 q++;
 366         }
 367         *q = '\0';
 368         if (i >= nitems)
 369             elog(WARN, "array_in: illformed array constant");
 370         values[i] = (*inputproc) (p, typelem);
 371         p = ++q;
 372         if (!eoArray)
 373             /*
 374              * if not at the end of the array skip white space
 375              */
 376             while (isspace(*q)) {
 377                 p++;
 378                 q++;
 379             }
 380     }
 381     if (typlen > 0) {
 382         *nbytes = nitems * typlen;
 383         if (!typbyval)
 384             for (i = 0; i < nitems; i++)
 385                 if (!values[i]) {
 386                     values[i] = palloc(typlen);
 387                     memset(values[i], 0, typlen);
 388                 }
 389     } else {
 390         for (i = 0, *nbytes = 0; i < nitems; i++) {
 391             if (values[i]) {
 392                 if (typalign=='d') {
 393                     *nbytes += DOUBLEALIGN(* (int32 *) values[i]);
 394                 } else {
 395                     *nbytes += INTALIGN(* (int32 *) values[i]);
 396                 }
 397             } else {
 398                 *nbytes += sizeof(int32);
 399                 values[i] = palloc(sizeof(int32));
 400                 *(int32 *)values[i] = sizeof(int32);
 401             }
 402         }
 403     }
 404     return((char *)values);
 405 }
 406
 407
 408 /*----------------------------------------------------------------------------
 409  * Read data about an array to be stored as a large object
 410  *----------------------------------------------------------------------------
 411  */
 412 #ifdef LOARRAY
 413 static char *
 414 _ReadLOArray(char *str,
 415              int *nbytes,
 416              int *fd,
 417              bool *chunkFlag,
 418              int ndim,
 419              int dim[],
 420              int baseSize)
 421 {
 422     char *inputfile, *accessfile = NULL, *chunkfile = NULL;
 423     char *retStr, *_AdvanceBy1word();
 424     Oid lobjId;
 425
 426     str = _AdvanceBy1word(str, &inputfile);
 427
 428     while (str != NULL) {
 429         char *word;
 430
 431         str = _AdvanceBy1word(str, &word);
 432
 433         if (!strcmp (word, "-chunk")) {
 434             if (str == NULL)
 435                 elog(WARN, "array_in: access pattern file required");
 436             str = _AdvanceBy1word(str, &accessfile);
 437         }
 438         else if (!strcmp (word, "-noreorg"))  {
 439             if (str == NULL)
 440                 elog(WARN, "array_in: chunk file required");
 441             str = _AdvanceBy1word(str, &chunkfile);
 442         } else {
 443             elog(WARN, "usage: <input file> -chunk DEFAULT/<access pattern file> -invert/-native [-noreorg <chunk file>]");
 444         }
 445     }
 446
 447     if (inputfile == NULL)
 448         elog(WARN, "array_in: missing file name");
 449     lobjId = lo_creat(0);
 450     *fd = lo_open(lobjId, INV_READ);
 451     if ( *fd < 0 )
 452         elog(WARN, "Large object create failed");
 453     retStr = inputfile;
 454     *nbytes = strlen(retStr) + 2;
 455
 456     if ( accessfile ) {
 457         FILE *afd;
 458         if ((afd = fopen (accessfile, "r")) == NULL)
 459             elog(WARN, "unable to open access pattern file");
 460         *chunkFlag = true;
 461         retStr = _ChunkArray(*fd, afd, ndim, dim, baseSize, nbytes,
 462                              chunkfile);
 463     }
 464     return(retStr);
 465 }
 466 #endif
 467
 468 static void
 469 _CopyArrayEls(char **values,
 470               char *p,
 471               int nitems,
 472               int typlen,
 473               char typalign,
 474               bool typbyval)
 475 {
 476     int i;
 477
 478     for (i = 0; i < nitems; i++) {
 479         int inc;
 480         inc = ArrayCastAndSet(values[i], typbyval, typlen, p);
 481         p += inc;
 482         if (!typbyval)
 483             pfree(values[i]);
 484     }
 485     pfree(values);
 486 }
 487
 488 /*-------------------------------------------------------------------------
 489  * array_out :
 490  *         takes the internal representation of an array and returns a string
 491  *        containing the array in its external format.
 492  *-------------------------------------------------------------------------
 493  */
 494 char *
 495 array_out(ArrayType *v, Oid element_type)
 496 {
 497     int typlen;
 498     bool typbyval;
 499     char typdelim;
 500     Oid typoutput, typelem;
 501     func_ptr outputproc;
 502     char typalign;
 503
 504     char *p, *retval, **values, delim[2];
 505     int nitems, overall_length, i, j, k, indx[MAXDIM];
 506     bool dummy_bool;
 507     int dummy_int;
 508     int ndim, *dim;
 509
 510     if (v == (ArrayType *) NULL)
 511         return ((char *) NULL);
 512
 513     if (ARR_IS_LO(v) == true)  {
 514         char *p, *save_p;
 515         int nbytes;
 516
 517         /* get a wide string to print to */
 518         p = array_dims(v, &dummy_bool);
 519         nbytes = strlen(ARR_DATA_PTR(v)) + 4 + *(int *)p;
 520
 521         save_p = (char *) palloc(nbytes);
 522
 523         strcpy(save_p, p + sizeof(int));
 524         strcat(save_p, ASSGN);
 525         strcat(save_p, ARR_DATA_PTR(v));
 526         pfree(p);
 527         return (save_p);
 528     }
 529
 530     system_cache_lookup(element_type, false, &typlen, &typbyval,
 531                         &typdelim, &typelem, &typoutput, &typalign);
 532     fmgr_info(typoutput, & outputproc, &dummy_int);
 533     sprintf(delim, "%c", typdelim);
 534     ndim = ARR_NDIM(v);
 535     dim = ARR_DIMS(v);
 536     nitems = getNitems(ndim, dim);
 537
 538     if (nitems == 0) {
 539         char *emptyArray = palloc(3);
 540         emptyArray[0] = '{';
 541         emptyArray[1] = '}';
 542         emptyArray[2] = '\0';
 543         return emptyArray;
 544     }
 545
 546     p = ARR_DATA_PTR(v);
 547     overall_length = 1; /* [TRH] don't forget to count \0 at end. */
 548     values = (char **) palloc(nitems * sizeof (char *));
 549     for (i = 0; i < nitems; i++) {
 550         if (typbyval) {
 551             switch(typlen) {
 552             case 1:
 553                 values[i] = (*outputproc) (*p, typelem);
 554                 break;
 555             case 2:
 556                 values[i] = (*outputproc) (* (int16 *) p, typelem);
 557                 break;
 558             case 3:
 559             case 4:
 560                 values[i] = (*outputproc) (* (int32 *) p, typelem);
 561                 break;
 562             }
 563             p += typlen;
 564         } else {
 565             values[i] = (*outputproc) (p, typelem);
 566             if (typlen > 0)
 567                 p += typlen;
 568             else
 569                 p += INTALIGN(* (int32 *) p);
 570             /*
 571              * For the pair of double quotes
 572              */
 573             overall_length += 2;
 574         }
 575         overall_length += (strlen(values[i]) + 1);
 576     }
 577
 578     /*
 579      * count total number of curly braces in output string
 580      */
 581     for (i = j = 0, k = 1; i < ndim; k *= dim[i++], j += k);
 582
 583     p = (char *) palloc(overall_length + 2*j);
 584     retval = p;
 585
 586     strcpy(p, "{");
 587     for (i = 0; i < ndim; indx[i++] = 0);
 588     j = 0; k = 0;
 589     do {
 590         for (i = j; i < ndim - 1; i++)
 591             strcat(p, "{");
 592         /*
 593          * Surround anything that is not passed by value in double quotes.
 594          * See above for more details.
 595          */
 596         if (!typbyval) {
 597             strcat(p, "\"");
 598             strcat(p, values[k]);
 599             strcat(p, "\"");
 600         } else
 601             strcat(p, values[k]);
 602         pfree(values[k++]);
 603
 604         for (i = ndim - 1; i >= 0; i--) {
 605             indx[i] = (indx[i] + 1)%dim[i];
 606             if (indx[i]) {
 607                 strcat (p, delim);
 608                 break;
 609             } else
 610                 strcat (p, "}");
 611         }
 612         j = i;
 613     } while (j  != -1);
 614
 615     pfree(values);
 616     return(retval);
 617 }
 618
 619 /*-----------------------------------------------------------------------------
 620  * array_dims :
 621  *        returns the dimension of the array pointed to by "v"
 622  *----------------------------------------------------------------------------
 623  */
 624 char *
 625 array_dims(ArrayType *v, bool *isNull)
 626 {
 627     char *p, *save_p;
 628     int nbytes, i;
 629     int *dimv, *lb;
 630
 631     if (v == (ArrayType *) NULL) RETURN_NULL;
 632     nbytes = ARR_NDIM(v)*33;
 633     /*
 634      * 33 since we assume 15 digits per number + ':' +'[]'
 635      */
 636     save_p = p =  (char *) palloc(nbytes + 4);
 637     memset(save_p, 0, nbytes + 4);
 638     dimv = ARR_DIMS(v); lb = ARR_LBOUND(v);
 639     p += 4;
 640     for (i = 0; i < ARR_NDIM(v); i++) {
 641         sprintf(p, "[%d:%d]", lb[i], dimv[i]+lb[i]-1);
 642         p += strlen(p);
 643     }
 644     nbytes = strlen(save_p + 4) + 4;
 645     memmove(save_p, &nbytes,4);
 646     return (save_p);
 647 }
 648
 649 /*---------------------------------------------------------------------------
 650  * array_ref :
 651  *    This routing takes an array pointer and an index array and returns
 652  *    a pointer to the referred element if element is passed by
 653  *    reference otherwise returns the value of the referred element.
 654  *---------------------------------------------------------------------------
 655  */
 656 Datum
 657 array_ref(ArrayType *array,
 658           int n,
 659           int indx[],
 660           int reftype,
 661           int elmlen,
 662           int arraylen,
 663           bool *isNull)
 664 {
 665     int i, ndim, *dim, *lb, offset, nbytes;
 666     struct varlena *v = NULL;
 667     char *retval = NULL;
 668
 669     if (array == (ArrayType *) NULL) RETURN_NULL;
 670     if (arraylen > 0) {
 671         /*
 672          * fixed length arrays -- these are assumed to be 1-d
 673          */
 674         if (indx[0]*elmlen > arraylen)
 675             elog(WARN, "array_ref: array bound exceeded");
 676         retval = (char *)array + indx[0]*elmlen;
 677         return _ArrayCast(retval, reftype, elmlen);
 678     }
 679     dim = ARR_DIMS(array);
 680     lb = ARR_LBOUND(array);
 681     ndim = ARR_NDIM(array);
 682     nbytes = (* (int32 *) array) - ARR_OVERHEAD(ndim);
 683
 684     if (!SanityCheckInput(ndim, n,  dim, lb, indx))
 685         RETURN_NULL;
 686
 687     offset = GetOffset(n, dim, lb, indx);
 688
 689     if (ARR_IS_LO(array)) {
 690         char * lo_name;
 691         int fd = 0;
 692
 693         /* We are assuming fixed element lengths here */
 694         offset *= elmlen;
 695         lo_name = (char *)ARR_DATA_PTR(array);
 696 #ifdef LOARRAY
 697         if ((fd = LOopen(lo_name, ARR_IS_INV(array)?INV_READ:O_RDONLY)) < 0)
 698             RETURN_NULL;
 699 #endif
 700         if (ARR_IS_CHUNKED(array))
 701             v = _ReadChunkArray1El(indx, elmlen, fd, array, isNull);
 702         else {
 703             if (lo_lseek(fd, offset, SEEK_SET) < 0)
 704                 RETURN_NULL;
 705 #ifdef LOARRAY
 706             v = (struct varlena *) LOread(fd, elmlen);
 707 #endif
 708         }
 709         if (*isNull) RETURN_NULL;
 710         if (VARSIZE(v) - 4 < elmlen)
 711             RETURN_NULL;
 712         (void) lo_close(fd);
 713         retval  = (char *)_ArrayCast((char *)VARDATA(v), reftype, elmlen);
 714         if ( reftype == 0) { /* not by value */
 715             char * tempdata = palloc (elmlen);
 716             memmove(tempdata, retval, elmlen);
 717             retval = tempdata;
 718         }
 719         pfree(v);
 720         return (Datum) retval;
 721     }
 722
 723     if (elmlen >  0) {
 724         offset = offset * elmlen;
 725         /*  off the end of the array */
 726         if (nbytes - offset < 1) RETURN_NULL;
 727         retval = ARR_DATA_PTR (array) + offset;
 728         return _ArrayCast(retval, reftype, elmlen);
 729     } else {
 730         bool done = false;
 731         char *temp;
 732         int bytes = nbytes;
 733         temp = ARR_DATA_PTR (array);
 734         i = 0;
 735         while (bytes > 0 && !done) {
 736             if (i == offset) {
 737                 retval = temp;
 738                 done = true;
 739             }
 740             bytes -= INTALIGN(* (int32 *) temp);
 741             temp += INTALIGN(* (int32 *) temp);
 742             i++;
 743         }
 744         if (! done)
 745             RETURN_NULL;
 746         return (Datum) retval;
 747     }
 748 }
 749
 750 /*-----------------------------------------------------------------------------
 751  * array_clip :
 752  *        This routine takes an array and a range of indices (upperIndex and
 753  *         lowerIndx), creates a new array structure for the referred elements
 754  *         and returns a pointer to it.
 755  *-----------------------------------------------------------------------------
 756  */
 757 Datum
 758 array_clip(ArrayType *array,
 759            int n,
 760            int upperIndx[],
 761            int lowerIndx[],
 762            int reftype,
 763            int len,
 764            bool *isNull)
 765 {
 766     int i, ndim, *dim, *lb, nbytes;
 767     ArrayType *newArr;
 768     int bytes, span[MAXDIM];
 769
 770     /* timer_start(); */
 771     if (array == (ArrayType *) NULL)
 772         RETURN_NULL;
 773     dim = ARR_DIMS(array);
 774     lb = ARR_LBOUND(array);
 775     ndim = ARR_NDIM(array);
 776     nbytes = (* (int32 *) array) - ARR_OVERHEAD(ndim);
 777
 778     if (!SanityCheckInput(ndim, n,  dim, lb, upperIndx))
 779         RETURN_NULL;
 780
 781     if (!SanityCheckInput(ndim, n, dim, lb, lowerIndx))
 782         RETURN_NULL;
 783
 784     for (i = 0; i < n; i++)
 785         if (lowerIndx[i] > upperIndx[i])
 786             elog(WARN, "lowerIndex cannot be larger than upperIndx");
 787     mda_get_range(n, span, lowerIndx, upperIndx);
 788
 789     if (ARR_IS_LO(array)) {
 790 #ifdef LOARRAY
 791         char * lo_name;
 792 #endif
 793         char *newname = NULL;
 794         int fd = 0, newfd = 0, isDestLO = true, rsize;
 795
 796         if (len < 0)
 797             elog(WARN, "array_clip: array of variable length objects not supported");
 798 #ifdef LOARRAY
 799         lo_name = (char *)ARR_DATA_PTR(array);
 800         if ((fd = LOopen(lo_name, ARR_IS_INV(array)?INV_READ:O_RDONLY)) < 0)
 801             RETURN_NULL;
 802         newname = _array_newLO( &newfd, Unix );
 803 #endif
 804         bytes = strlen(newname) + 1 + ARR_OVERHEAD(n);
 805         newArr = (ArrayType *) palloc(bytes);
 806         memmove(newArr, array, sizeof(ArrayType));
 807         memmove(newArr, &bytes, sizeof(int));
 808         memmove(ARR_DIMS(newArr), span, n*sizeof(int));
 809         memmove(ARR_LBOUND(newArr), lowerIndx, n*sizeof(int));
 810         strcpy(ARR_DATA_PTR(newArr), newname);
 811
 812         rsize = compute_size (lowerIndx, upperIndx, n, len);
 813         if (rsize < MAX_BUFF_SIZE) {
 814             char *buff;
 815             rsize += 4;
 816             buff = palloc(rsize);
 817             if ( buff )
 818                 isDestLO = false;
 819             if (ARR_IS_CHUNKED(array)) {
 820                 _ReadChunkArray(lowerIndx, upperIndx, len, fd, &(buff[4]),
 821                                 array,0,isNull);
 822             } else {
 823                 _ReadArray(lowerIndx, upperIndx, len, fd, (int)&(buff[4]),
 824                            array,
 825                            0,isNull);
 826             }
 827             memmove(buff, &rsize, 4);
 828 #ifdef LOARRAY
 829             if (! *isNull)
 830                 bytes = LOwrite(newfd, (struct varlena *)buff);
 831 #endif
 832             pfree (buff);
 833         }
 834         if (isDestLO)
 835             if (ARR_IS_CHUNKED(array)) {
 836                 _ReadChunkArray(lowerIndx, upperIndx, len, fd, (char*)newfd, array,
 837                                 1,isNull);
 838             } else {
 839                 _ReadArray(lowerIndx, upperIndx, len, fd, newfd, array, 1,isNull);
 840             }
 841 #ifdef LOARRAY
 842         (void) LOclose(fd);
 843         (void) LOclose(newfd);
 844 #endif
 845         if (*isNull) {
 846             pfree(newArr);
 847             newArr = NULL;
 848         }
 849         /* timer_end(); */
 850         return ((Datum) newArr);
 851     }
 852
 853     if (len >  0) {
 854         bytes = getNitems(n, span);
 855         bytes = bytes*len + ARR_OVERHEAD(n);
 856     } else {
 857         bytes = _ArrayClipCount(lowerIndx, upperIndx, array);
 858         bytes += ARR_OVERHEAD(n);
 859     }
 860     newArr = (ArrayType *) palloc(bytes);
 861     memmove(newArr, array, sizeof(ArrayType));
 862     memmove(newArr, &bytes, sizeof(int));
 863     memmove(ARR_DIMS(newArr), span, n*sizeof(int));
 864     memmove(ARR_LBOUND(newArr), lowerIndx, n*sizeof(int));
 865     _ArrayRange(lowerIndx, upperIndx, len, ARR_DATA_PTR(newArr), array, 1);
 866     return (Datum) newArr;
 867 }
 868
 869 /*-----------------------------------------------------------------------------
 870  * array_set  :
 871  *        This routine sets the value of an array location (specified by an index array)
 872  *        to a new value specified by "dataPtr".
 873  * result :
 874  *        returns a pointer to the modified array.
 875  *-----------------------------------------------------------------------------
 876  */
 877 char *
 878 array_set(ArrayType *array,
 879           int n,
 880           int indx[],
 881           char *dataPtr,
 882           int reftype,
 883           int elmlen,
 884           int arraylen,
 885           bool *isNull)
 886 {
 887     int ndim, *dim, *lb, offset, nbytes;
 888     char *pos;
 889
 890     if (array == (ArrayType *) NULL)
 891         RETURN_NULL;
 892     if (arraylen > 0) {
 893         /*
 894          * fixed length arrays -- these are assumed to be 1-d
 895          */
 896         if (indx[0]*elmlen > arraylen)
 897             elog(WARN, "array_ref: array bound exceeded");
 898         pos = (char *)array + indx[0]*elmlen;
 899         ArrayCastAndSet(dataPtr, (bool) reftype, elmlen, pos);
 900         return((char *)array);
 901     }
 902     dim = ARR_DIMS(array);
 903     lb = ARR_LBOUND(array);
 904     ndim = ARR_NDIM(array);
 905     nbytes = (* (int32 *) array) - ARR_OVERHEAD(ndim);
 906
 907     if (!SanityCheckInput(ndim, n,  dim, lb, indx))
 908     {
 909         elog(WARN, "array_set: array bound exceeded");
 910         return((char *)array);
 911     }
 912     offset = GetOffset( n, dim, lb, indx);
 913
 914     if (ARR_IS_LO(array)) {
 915         int fd = 0;
 916         struct varlena *v;
 917
 918         /* We are assuming fixed element lengths here */
 919         offset *= elmlen;
 920 #ifdef LOARRAY
 921         char * lo_name;
 922
 923         lo_name = ARR_DATA_PTR(array);
 924         if ((fd = LOopen(lo_name, ARR_IS_INV(array)?INV_WRITE:O_WRONLY)) < 0)
 925             return((char *)array);
 926 #endif
 927         if (lo_lseek(fd, offset, SEEK_SET) < 0)
 928             return((char *)array);
 929         v = (struct varlena *) palloc(elmlen + 4);
 930         VARSIZE (v) = elmlen + 4;
 931         ArrayCastAndSet(dataPtr, (bool) reftype, elmlen, VARDATA(v));
 932 #ifdef LOARRAY
 933         n =  LOwrite(fd, v);
 934 #endif
 935         /* if (n < VARSIZE(v) - 4)
 936            RETURN_NULL;
 937            */
 938         pfree(v);
 939         (void) lo_close(fd);
 940         return((char *)array);
 941     }
 942     if (elmlen >  0) {
 943         offset = offset * elmlen;
 944         /*  off the end of the array */
 945         if (nbytes - offset < 1) return((char *)array);
 946         pos = ARR_DATA_PTR (array) + offset;
 947     } else {
 948         ArrayType *newarray;
 949         char *elt_ptr;
 950         int oldsize, newsize, oldlen, newlen, lth0, lth1, lth2;
 951
 952         elt_ptr = array_seek(ARR_DATA_PTR(array), -1, offset);
 953         oldlen = INTALIGN(*(int32 *)elt_ptr);
 954         newlen = INTALIGN(*(int32 *)dataPtr);
 955
 956         if (oldlen == newlen) {
 957             /* new element with same size, overwrite old data */
 958             ArrayCastAndSet(dataPtr, (bool)reftype, elmlen, elt_ptr);
 959             return((char *)array);
 960         }
 961
 962         /* new element with different size, reallocate the array */
 963         oldsize = array->size;
 964         lth0 = ARR_OVERHEAD(n);
 965         lth1 = (int)(elt_ptr - ARR_DATA_PTR(array));
 966         lth2 = (int)(oldsize - lth0 - lth1 - oldlen);
 967         newsize = lth0 + lth1 + newlen + lth2;
 968
 969         newarray = (ArrayType *)palloc(newsize);
 970         memmove((char *)newarray, (char *)array, lth0+lth1);
 971         newarray->size = newsize;
 972         newlen = ArrayCastAndSet(dataPtr, (bool)reftype, elmlen,
 973                                  (char *)newarray+lth0+lth1);
 974         memmove((char *)newarray+lth0+lth1+newlen,
 975                 (char *)array+lth0+lth1+oldlen, lth2);
 976
 977         /* ??? who should free this storage ??? */
 978         return((char *)newarray);
 979     }
 980     ArrayCastAndSet(dataPtr, (bool) reftype, elmlen, pos);
 981     return((char *)array);
 982 }
 983
 984 /*----------------------------------------------------------------------------
 985  * array_assgn :
 986  *        This routine sets the value of a range of array locations (specified
 987  *        by upper and lower index values ) to new values passed as
 988  *        another array
 989  * result :
 990  *        returns a pointer to the modified array.
 991  *----------------------------------------------------------------------------
 992  */
 993 char *
 994 array_assgn(ArrayType *array,
 995             int n,
 996             int upperIndx[],
 997             int lowerIndx[],
 998             ArrayType *newArr,
 999             int reftype,
1000             int len,
1001             bool *isNull)
1002 {
1003     int i, ndim, *dim, *lb;
1004
1005     if (array == (ArrayType *) NULL)
1006         RETURN_NULL;
1007     if (len < 0)
1008         elog(WARN,"array_assgn:updates on arrays of variable length elements not allowed");
1009
1010     dim = ARR_DIMS(array);
1011     lb = ARR_LBOUND(array);
1012     ndim = ARR_NDIM(array);
1013
1014     if (!SanityCheckInput(ndim, n,  dim, lb, upperIndx) ||
1015         !SanityCheckInput(ndim, n, dim, lb, lowerIndx)) {
1016         return((char *)array);
1017     }
1018
1019     for (i = 0; i < n; i++)
1020         if (lowerIndx[i] > upperIndx[i])
1021             elog(WARN, "lowerIndex larger than upperIndx");
1022
1023     if (ARR_IS_LO(array)) {
1024         int fd = 0, newfd = 0;
1025
1026 #ifdef LOARRAY
1027         char * lo_name;
1028
1029         lo_name = (char *)ARR_DATA_PTR(array);
1030         if ((fd = LOopen(lo_name,  ARR_IS_INV(array)?INV_WRITE:O_WRONLY)) < 0)
1031             return((char *)array);
1032 #endif
1033         if (ARR_IS_LO(newArr)) {
1034 #ifdef LOARRAY
1035             lo_name = (char *)ARR_DATA_PTR(newArr);
1036             if ((newfd = LOopen(lo_name, ARR_IS_INV(newArr)?INV_READ:O_RDONLY)) < 0)
1037                 return((char *)array);
1038 #endif
1039             _LOArrayRange(lowerIndx, upperIndx, len, fd, newfd, array, 1, isNull);
1040             (void) lo_close(newfd);
1041         } else {
1042             _LOArrayRange(lowerIndx, upperIndx, len, fd, (int)ARR_DATA_PTR(newArr),
1043                           array, 0, isNull);
1044         }
1045         (void) lo_close(fd);
1046         return ((char *) array);
1047     }
1048     _ArrayRange(lowerIndx, upperIndx, len, ARR_DATA_PTR(newArr), array, 0);
1049     return (char *) array;
1050 }
1051
1052 /*-----------------------------------------------------------------------------
1053  * array_eq :
1054  *        compares two arrays for equality
1055  * result :
1056  *        returns 1 if the arrays are equal, 0 otherwise.
1057  *-----------------------------------------------------------------------------
1058  */
1059 int
1060 array_eq (ArrayType *array1, ArrayType *array2)
1061 {
1062     if ((array1 == NULL) || (array2 == NULL))
1063         return(0);
1064     if (*(int *)array1 != *(int *)array2)
1065         return (0);
1066     if (memcmp(array1, array2, *(int *)array1))
1067         return(0);
1068     return(1);
1069 }
1070
1071 /***************************************************************************/
1072 /******************|          Support  Routines           |*****************/
1073 /***************************************************************************/
1074 static void
1075 system_cache_lookup(Oid element_type,
1076                     bool input,
1077                     int *typlen,
1078                     bool *typbyval,
1079                     char *typdelim,
1080                     Oid *typelem,
1081                     Oid *proc,
1082                     char *typalign)
1083 {
1084     HeapTuple typeTuple;
1085     TypeTupleForm typeStruct;
1086
1087     typeTuple = SearchSysCacheTuple(TYPOID, ObjectIdGetDatum(element_type),
1088                                     0,0,0);
1089
1090     if (!HeapTupleIsValid(typeTuple)) {
1091         elog(WARN, "array_out: Cache lookup failed for type %d\n",
1092              element_type);
1093         return;
1094     }
1095     typeStruct = (TypeTupleForm) GETSTRUCT(typeTuple);
1096     *typlen    = typeStruct->typlen;
1097     *typbyval  = typeStruct->typbyval;
1098     *typdelim  = typeStruct->typdelim;
1099     *typelem   = typeStruct->typelem;
1100     *typalign  = typeStruct->typalign;
1101     if (input) {
1102         *proc = typeStruct->typinput;
1103     } else {
1104         *proc = typeStruct->typoutput;
1105     }
1106 }
1107
1108 static Datum
1109 _ArrayCast(char *value, bool byval, int len)
1110 {
1111     if (byval) {
1112         switch (len) {
1113         case 1:
1114             return((Datum) * value);
1115         case 2:
1116             return((Datum) * (int16 *) value);
1117         case 3:
1118         case 4:
1119             return((Datum) * (int32 *) value);
1120         default:
1121             elog(WARN, "array_ref: byval and elt len > 4!");
1122             break;
1123         }
1124     } else {
1125         return (Datum) value;
1126     }
1127  return 0;
1128 }
1129
1130
1131 static int
1132 ArrayCastAndSet(char *src,
1133                 bool typbyval,
1134                 int typlen,
1135                 char *dest)
1136 {
1137     int inc;
1138
1139     if (typlen > 0) {
1140         if (typbyval) {
1141             switch(typlen) {
1142             case 1:
1143                 *dest = DatumGetChar(src);
1144                 break;
1145             case 2:
1146                 * (int16 *) dest = DatumGetInt16(src);
1147                 break;
1148             case 4:
1149                 * (int32 *) dest = (int32)src;
1150                 break;
1151             }
1152         } else {
1153             memmove(dest, src, typlen);
1154         }
1155         inc = typlen;
1156     } else {
1157         memmove(dest, src, *(int32 *)src);
1158         inc = (INTALIGN(* (int32 *) src));
1159     }
1160     return(inc);
1161 }
1162
1163 #ifdef LOARRAY
1164 static char *
1165 _AdvanceBy1word(char *str, char **word)
1166 {
1167     char *retstr, *space;
1168
1169     *word = NULL;
1170     if (str == NULL) return str;
1171     while (isspace(*str)) str++;
1172     *word = str;
1173     if ((space = (char *)strchr(str, ' ')) != (char *) NULL) {
1174         retstr = space + 1;
1175         *space = '\0';
1176     }
1177     else
1178         retstr = NULL;
1179     return retstr;
1180 }
1181 #endif
1182
1183 int
1184 SanityCheckInput(int ndim, int n, int dim[], int lb[], int indx[])
1185 {
1186     int i;
1187     /* Do Sanity check on input */
1188     if (n != ndim) return 0;
1189     for (i = 0; i < ndim; i++)
1190         if ((lb[i] > indx[i]) || (indx[i] >= (dim[i] + lb[i])))
1191             return 0;
1192     return 1;
1193 }
1194
1195 static void
1196 _ArrayRange(int st[],
1197             int endp[],
1198             int bsize,
1199             char *destPtr,
1200             ArrayType *array,
1201             int from)
1202 {
1203     int n, *dim, *lb, st_pos, prod[MAXDIM];
1204     int span[MAXDIM], dist[MAXDIM], indx[MAXDIM];
1205     int i, j, inc;
1206     char *srcPtr, *array_seek();
1207
1208     n = ARR_NDIM(array); dim = ARR_DIMS(array);
1209     lb = ARR_LBOUND(array); srcPtr = ARR_DATA_PTR(array);
1210     for (i = 0; i < n; st[i] -= lb[i], endp[i] -= lb[i], i++);
1211     mda_get_prod(n, dim, prod);
1212     st_pos = tuple2linear(n, st, prod);
1213     srcPtr = array_seek(srcPtr, bsize, st_pos);
1214     mda_get_range(n, span, st, endp);
1215     mda_get_offset_values(n, dist, prod, span);
1216     for (i=0; i < n; indx[i++]=0);
1217     i = j = n-1; inc = bsize;
1218     do {
1219         srcPtr = array_seek(srcPtr, bsize,  dist[j]);
1220         if (from)
1221             inc = array_read(destPtr, bsize, 1, srcPtr);
1222         else
1223             inc = array_read(srcPtr, bsize, 1, destPtr);
1224         destPtr += inc; srcPtr += inc;
1225     } while ((j = next_tuple(i+1, indx, span)) != -1);
1226 }
1227
1228 static int
1229 _ArrayClipCount(int stI[], int endpI[], ArrayType *array)
1230 {
1231     int n, *dim, *lb, st_pos, prod[MAXDIM];
1232     int span[MAXDIM], dist[MAXDIM], indx[MAXDIM];
1233     int i, j, inc, st[MAXDIM], endp[MAXDIM];
1234     int count = 0;
1235     char *ptr, *array_seek();
1236
1237     n = ARR_NDIM(array); dim = ARR_DIMS(array);
1238     lb = ARR_LBOUND(array); ptr = ARR_DATA_PTR(array);
1239     for (i = 0; i < n; st[i] = stI[i]-lb[i], endp[i]=endpI[i]-lb[i], i++);
1240     mda_get_prod(n, dim, prod);
1241     st_pos = tuple2linear(n, st, prod);
1242     ptr = array_seek(ptr, -1, st_pos);
1243     mda_get_range(n, span, st, endp);
1244     mda_get_offset_values(n, dist, prod, span);
1245     for (i=0; i < n; indx[i++]=0);
1246     i = j = n-1;
1247     do {
1248         ptr = array_seek(ptr, -1,  dist[j]);
1249         inc =  INTALIGN(* (int32 *) ptr);
1250         ptr += inc; count += inc;
1251     } while ((j = next_tuple(i+1, indx, span)) != -1);
1252     return count;
1253 }
1254
1255 char *
1256 array_seek(char *ptr, int eltsize, int nitems)
1257 {
1258     int i;
1259
1260     if (eltsize > 0)
1261         return(ptr + eltsize*nitems);
1262     for (i = 0; i < nitems; i++)
1263         ptr += INTALIGN(* (int32 *) ptr);
1264     return(ptr);
1265 }
1266
1267 int
1268 array_read(char *destptr, int eltsize, int nitems, char *srcptr)
1269 {
1270     int i, inc, tmp;
1271
1272     if (eltsize > 0)  {
1273         memmove(destptr, srcptr, eltsize*nitems);
1274         return(eltsize*nitems);
1275     }
1276     for (i = inc = 0; i < nitems; i++) {
1277         tmp = (INTALIGN(* (int32 *) srcptr));
1278         memmove(destptr, srcptr, tmp);
1279         srcptr += tmp;
1280         destptr += tmp;
1281         inc += tmp;
1282     }
1283     return(inc);
1284 }
1285
1286 static void
1287 _LOArrayRange(int st[],
1288               int endp[],
1289               int bsize,
1290               int srcfd,
1291               int destfd,
1292               ArrayType *array,
1293               int isSrcLO,
1294               bool *isNull)
1295 {
1296     int n, *dim, st_pos, prod[MAXDIM];
1297     int span[MAXDIM], dist[MAXDIM], indx[MAXDIM];
1298     int i, j, inc, tmp, *lb, offset;
1299
1300     n = ARR_NDIM(array); dim = ARR_DIMS(array);
1301     lb = ARR_LBOUND(array);
1302     for (i = 0; i < n; st[i] -= lb[i], endp[i] -= lb[i], i++);
1303
1304     mda_get_prod(n, dim, prod);
1305     st_pos = tuple2linear(n, st, prod);
1306     offset = st_pos*bsize;
1307     if (lo_lseek(srcfd, offset, SEEK_SET) < 0)
1308         return;
1309     mda_get_range(n, span, st, endp);
1310     mda_get_offset_values(n, dist, prod, span);
1311     for (i=0; i < n; indx[i++]=0);
1312     for (i = n-1, inc = bsize; i >= 0; inc *= span[i--])
1313         if (dist[i])
1314             break;
1315     j = n-1;
1316     do {
1317         offset += (dist[j]*bsize);
1318         if (lo_lseek(srcfd,  offset, SEEK_SET) < 0)
1319             return;
1320         tmp = _LOtransfer((char**)&srcfd, inc, 1, (char**)&destfd, isSrcLO, 1);
1321         if ( tmp < inc )
1322             return;
1323         offset += inc;
1324     } while ((j = next_tuple(i+1, indx, span)) != -1);
1325 }
1326
1327
1328 static void
1329 _ReadArray (int st[],
1330             int endp[],
1331             int bsize,
1332             int srcfd,
1333             int destfd,
1334             ArrayType *array,
1335             int isDestLO,
1336             bool *isNull)
1337 {
1338     int n, *dim, st_pos, prod[MAXDIM];
1339     int span[MAXDIM], dist[MAXDIM], indx[MAXDIM];
1340     int i, j, inc, tmp, *lb, offset;
1341
1342     n = ARR_NDIM(array); dim = ARR_DIMS(array);
1343     lb = ARR_LBOUND(array);
1344     for (i = 0; i < n; st[i] -= lb[i], endp[i] -= lb[i], i++);
1345
1346     mda_get_prod(n, dim, prod);
1347     st_pos = tuple2linear(n, st, prod);
1348     offset = st_pos*bsize;
1349     if (lo_lseek(srcfd, offset, SEEK_SET) < 0)
1350         return;
1351     mda_get_range(n, span, st, endp);
1352     mda_get_offset_values(n, dist, prod, span);
1353     for (i=0; i < n; indx[i++]=0);
1354     for (i = n-1, inc = bsize; i >= 0; inc *= span[i--])
1355         if (dist[i])
1356             break;
1357     j = n-1;
1358     do {
1359         offset += (dist[j]*bsize);
1360         if (lo_lseek(srcfd,  offset, SEEK_SET) < 0)
1361             return;
1362         tmp = _LOtransfer((char**)&destfd, inc, 1, (char**)&srcfd, 1, isDestLO);
1363         if ( tmp < inc )
1364             return;
1365         offset += inc;
1366     } while ((j = next_tuple(i+1, indx, span)) != -1);
1367 }
1368
1369
1370 int
1371 _LOtransfer(char **destfd,
1372             int size,
1373             int nitems,
1374             char **srcfd,
1375             int isSrcLO,
1376             int isDestLO)
1377 {
1378 #define MAX_READ (512 * 1024)
1379 #define min(a, b) (a < b ? a : b)
1380     struct varlena *v = NULL;
1381     int tmp, inc, resid;
1382
1383     inc = nitems*size;
1384     if (isSrcLO && isDestLO && inc > 0)
1385         for (tmp = 0, resid = inc;
1386              resid > 0 && (inc = min(resid, MAX_READ)) > 0; resid -= inc) {
1387 #ifdef LOARRAY
1388             v = (struct varlena *) LOread((int) *srcfd, inc);
1389             if (VARSIZE(v) - 4 < inc)
1390                 {pfree(v); return(-1);}
1391             tmp += LOwrite((int) *destfd, v);
1392 #endif
1393             pfree(v);
1394
1395         }
1396     else if (!isSrcLO && isDestLO) {
1397         tmp = lo_write((int) *destfd, *srcfd, inc);
1398         *srcfd = *srcfd + tmp;
1399     }
1400     else if (isSrcLO && !isDestLO) {
1401         tmp = lo_read((int) *srcfd, *destfd, inc);
1402         *destfd = *destfd + tmp;
1403     }
1404     else {
1405         memmove(*destfd, *srcfd, inc);
1406         tmp = inc;
1407         *srcfd += inc;
1408         *destfd += inc;
1409     }
1410     return(tmp);
1411 #undef MAX_READ
1412 }
1413
1414 char *
1415 _array_newLO(int *fd, int flag)
1416 {
1417     char *p;
1418     char saveName[NAME_LEN];
1419
1420     p = (char *) palloc(NAME_LEN);
1421     sprintf(p, "/Arry.%d", newoid());
1422     strcpy (saveName, p);
1423 #ifdef LOARRAY
1424     if ( (*fd = LOcreat (saveName, 0600, flag)) < 0)
1425         elog(WARN, "Large object create failed");
1426 #endif
1427     return (p);
1428 }
1429