cdemodp.c

用Oracle OCI进行Direct path insert的示例代码

     must convert the column array to stream format before calling
     this function again.

   FIELD_SET_PARTIAL:
     A field is in the partial state, the partial context is valid
     and is required to continue processing the field.  Note that
     when a field is partial, the row which contains the column
     corresponding to the field is partial also.

   FIELD_SET_ERROR:
     A read error occured on a secondary (out-of-line) data file.

 NOTES:
    Discuss how partials are handled.
 */

/* Deal with WIN32 CR-LF weirdness */
#if defined(WIN32COMMON) || defined(WIN32) || defined(_WIN32)
#define TKPIDRV_OPEN_MODE (O_RDONLY | O_BINARY)
#else
#define TKPIDRV_OPEN_MODE (O_RDONLY)
#endif

STATICF sword
field_set(ctlp, tblp, recp, rowoff)
struct loadctl    *ctlp;                           /* load control structure */
struct tbl        *tblp;                                 /* table descriptor */
text              *recp;                                     /* input record */
ub4                rowoff;                        /* column array row offset */
{
  ub1  *cval;
  ub4   thiscol;
  ub4   clen, j;                                            /* column length */
  ub1   cflg;
  sword ociret;
  int   fd;                          /* file descriptor for out-of-line data */
  char *filename;                           /* filename for out-of-line data */
  sword  partial;

  ctlp->bufoff_ctl = 0;

  if ((partial = (sword)ctlp->pctx_ctl.valid_pctx) == TRUE)
  {
    /* partial context is valid; resume where we left off */
    assert(rowoff == ctlp->pctx_ctl.row_pctx);
    thiscol = ctlp->pctx_ctl.col_pctx;
  }
  else
    thiscol = 0;

  for (/* empty */; thiscol < tblp->ncol_tbl; thiscol++)
  {
    struct col *colp =  &tblp->col_tbl[thiscol];
    struct fld *fldp =  &tblp->fld_tbl[thiscol];

    if (partial)
    {
      /* partials are always from a secondary file */
      fd       = ctlp->pctx_ctl.fd_pctx;
      filename = ctlp->pctx_ctl.fnm_pctx;
    }
    else                                                         /* !partial */
    {
      fd       = -1;
      filename = (char *)0;
      cval     = (ub1 *)recp + fldp->begpos_fld - 1;
      clen     = fldp->endpos_fld - fldp->begpos_fld + 1;

      j = 0;
      if (bit(fldp->flag_fld, FLD_STRIP_LEAD_BLANK))
      {
        /* trim leading white space */
        for (/*empty*/; j < clen; j++)
          if (!isspace((int)cval[j]))
            break;
      }

      if (j >= clen)
        clen = 0;                              /* null column, handled below */
      else
      {
        if (bit(fldp->flag_fld, FLD_STRIP_TRAIL_BLANK))
        {
          /* trim trailing white space */
          while (clen && isspace((int)cval[clen - 1]))
            clen--;
        }
        cval = cval + j;
        clen = clen - j;
      }


      if (clen)
      {
        if (bit(fldp->flag_fld, FLD_INLINE))
        {
          cflg = OCI_DIRPATH_COL_COMPLETE;
        }
        else if (bit(fldp->flag_fld, FLD_OUTOFLINE))
        {
          filename = (char *)malloc((size_t)clen+1);
          if (!filename)
          {
            perror("malloc");
            FATAL("field_set: cannot malloc buf for filename", (clen + 1));
          }
          (void) memcpy((dvoid *)filename, (dvoid *)cval, (size_t)clen);
          filename[clen] = 0;
          fd = open(filename, TKPIDRV_OPEN_MODE);
          SET_PCTX(ctlp->pctx_ctl, rowoff, thiscol, (ub4)0, fd, filename);
          LEN_PCTX(ctlp->pctx_ctl) = 0;
        }
        else
        {
          FATAL("field_set: unknown field type", fldp->flag_fld);
        }
      }
      else
      {
        cflg = OCI_DIRPATH_COL_NULL;               /* all spaces become null */
        cval = (ub1 *)0;
      }
    }

    if (bit(fldp->flag_fld, FLD_OUTOFLINE))
    {
      char *buf;
      ub4   bufsz;
      int   cnt;

      if (!ctlp->buf_ctl)
      {
        ctlp->buf_ctl   = (ub1 *)malloc((size_t)SECONDARY_BUF_SIZE);
        ctlp->bufsz_ctl = SECONDARY_BUF_SIZE;
      }

      if ((ctlp->bufsz_ctl - ctlp->bufoff_ctl) > SECONDARY_BUF_SLOP)
      {
        buf   = (char *)ctlp->buf_ctl + ctlp->bufoff_ctl;  /* buffer pointer */
        bufsz = (int)ctlp->bufsz_ctl  - ctlp->bufoff_ctl;     /* buffer size */

        if (fd == -1)
          cnt = 0;
        else
          cnt = read(fd, buf, bufsz);

        if (cnt != -1)
        {
          cval = (ub1 *)buf;
          clen = (ub4)cnt;

          if (cnt < bufsz)                    /* all file data has been read */
          {
            /* mark column as null or complete */
            if (cnt == 0 && LEN_PCTX(ctlp->pctx_ctl) == 0)
              cflg = OCI_DIRPATH_COL_NULL;
            else
              cflg = OCI_DIRPATH_COL_COMPLETE;

            field_flush(ctlp, rowoff);          /* close file, free filename */

            /* adjust offset into buffer for use by next field */
            ctlp->bufoff_ctl += cnt;
          }
          else
            cflg  = OCI_DIRPATH_COL_PARTIAL;
        }
        else
        {
          /* XXX: do something on read failure, like return an error context */
          field_flush(ctlp, rowoff);          /* close file, free filename */
          return FIELD_SET_ERROR;
        }
      }
      else
      {
        /* no room in secondary buffer, return a 0 length partial
         * and pick it up next time.
         */
        cflg = OCI_DIRPATH_COL_PARTIAL;
        clen = 0;
        cval = (ub1 *)NULL;
      }
    }

    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
              OCIDirPathColArrayEntrySet(ctlp->dpca_ctl, ctlp->errhp_ctl,
                                         rowoff, colp->id_col,
                                         cval, clen, cflg));

    if (cflg == OCI_DIRPATH_COL_PARTIAL)
    {
      /* Partials only occur for OutOfLine data
       * remember the row offset, column offset,
       * total length of the column so far,
       * and file descriptor to get data from on
       * subsequent calls to this function.
       */
      LEN_PCTX(ctlp->pctx_ctl) += clen;
      return FIELD_SET_PARTIAL;
    }
  }

  CLEAR_PCTX(ctlp->pctx_ctl);
  if (ctlp->bufoff_ctl)             /* data in secondary buffer for this row */
    return FIELD_SET_BUF;
  else
    return FIELD_SET_COMPLETE;
}


STATICF void
errprint(errhp, htype, errcodep)
dvoid  *errhp;
ub4     htype;
sb4    *errcodep;
{
  text errbuf[512];

  if (errhp)
  {
    sb4  errcode;

    if (errcodep == (sb4 *)0)
      errcodep = &errcode;

    (void) OCIErrorGet((dvoid *)errhp, (ub4) 1, (text *) NULL, errcodep,
                       errbuf, (ub4) sizeof(errbuf), htype);
    (void) fprintf(output_fp, "Error - %.*s\n", 512, errbuf);
  }
}

STATICF void
checkerr(errhp, htype, status, note, code, file, line)
dvoid *errhp;
ub4    htype;
sword  status;
text  *note;
sb4    code;
text  *file;
sb4    line;
{
  sb4 errcode = 0;

  if ((status != OCI_SUCCESS))
    (void) fprintf(output_fp, "OCI Error %ld occurred at File %s:%ld\n",
                   (long)status, (char *)file, (long)line);

  if (note)
    (void) fprintf(output_fp, "File %s:%ld (code=%ld)  %s\n",
                   (char *)file, (long)line, (long)code, (char *)note);

  switch (status)
  {
  case OCI_SUCCESS:
    break;
  case OCI_SUCCESS_WITH_INFO:
    (void) fprintf(output_fp, "Error - OCI_SUCCESS_WITH_INFO\n");
    errprint(errhp, htype, &errcode);
    break;
  case OCI_NEED_DATA:
    (void) fprintf(output_fp, "Error - OCI_NEED_DATA\n");
    break;
  case OCI_NO_DATA:
    (void) fprintf(output_fp, "Error - OCI_NODATA\n");
    break;
  case OCI_ERROR:
    errprint(errhp, htype, &errcode);
    break;
  case OCI_INVALID_HANDLE:
    (void) fprintf(output_fp, "Error - OCI_INVALID_HANDLE\n");
    break;
  case OCI_STILL_EXECUTING:
    (void) fprintf(output_fp, "Error - OCI_STILL_EXECUTE\n");
    break;
  case OCI_CONTINUE:
    (void) fprintf(output_fp, "Error - OCI_CONTINUE\n");
    break;
  default:
    break;
  }
}


/* cleanup
 *   Free up handles and exit with the supplied exit status code.
 */
STATICF void
cleanup(ctlp, ex_status)
struct loadctl *ctlp;
sb4    ex_status;
{
  sword ociret;

  /* Free the column array and stream handles if they have been
   * allocated.  We don't need to do this since freeing the direct
   * path context will free the heap which these child handles have
   * been allocated from.  I'm doing this just to exercise the code
   * path to free these handles.
   */
  if (ctlp->dpca_ctl)
  {
    ociret = OCIHandleFree((dvoid *)ctlp->dpca_ctl,
                           OCI_HTYPE_DIRPATH_COLUMN_ARRAY);
    if (ociret != OCI_SUCCESS)
      CHECKERR(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret);
  }

  if (ctlp->dpstr_ctl)
  {
    ociret = OCIHandleFree((dvoid *)ctlp->dpstr_ctl,
                           OCI_HTYPE_DIRPATH_STREAM);
    if (ociret != OCI_SUCCESS)
      CHECKERR(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret);
  }

  if (ctlp->dpctx_ctl)
  {
    ociret = OCIHandleFree((dvoid *)ctlp->dpctx_ctl, OCI_HTYPE_DIRPATH_CTX);
    if (ociret != OCI_SUCCESS)
      CHECKERR(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret);
  }

  if (ctlp->errhp_ctl && ctlp->srvhp_ctl)
  {
    (void) OCIServerDetach(ctlp->srvhp_ctl, ctlp->errhp_ctl, OCI_DEFAULT );
    ociret = OCIHandleFree((dvoid *)ctlp->srvhp_ctl, OCI_HTYPE_SERVER);
    if (ociret != OCI_SUCCESS)
      CHECKERR(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret);
  }

  if (ctlp->svchp_ctl)
    (void) OCIHandleFree((dvoid *) ctlp->svchp_ctl, OCI_HTYPE_SVCCTX);
  if (ctlp->errhp_ctl)
    (void) OCIHandleFree((dvoid *) ctlp->errhp_ctl, OCI_HTYPE_ERROR);

  if ((output_fp != stdout) && (output_fp != stderr))
    fclose(output_fp);

  exit((int)ex_status);
}


/* end of file cdemodp.c */