cdemodp.c

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

                       OCI_DTYPE_PARAM,
                       (dvoid *)&parmtyp, (ub4 *)0,
                       OCI_ATTR_PTYPE, ctlp->errhp_ctl));

  if (parmtyp != OCI_PTYPE_LIST)
  {
    fprintf(output_fp, "ERROR: expected parmtyp of OCI_PTYPE_LIST, got %d\n",
            (int)parmtyp);
  }

  /* Now set the attributes of each column by getting a parameter
   * handle on each column, then setting attributes on the parameter
   * handle for the column.
   * Note that positions within a column list descriptor are 1-based.
   */
  for (i = 0, pos = 1, colp = tblp->col_tbl, fldp = tblp->fld_tbl;
       i < tblp->ncol_tbl;
       i++, pos++, colp++, fldp++)
  {
    /* get parameter handle on the column */
    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
              OCIParamGet((CONST dvoid *)ctlp->colLstDesc_ctl,
                          (ub4)OCI_DTYPE_PARAM, ctlp->errhp_ctl,
                          (dvoid **)&colDesc, pos));


    colp->id_col = i;                            /* position in column array */

    /* set external attributes on the column */

    /* column name */
    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
              OCIAttrSet((dvoid *)colDesc, (ub4)OCI_DTYPE_PARAM,
                         (dvoid *)colp->name_col,
                         (ub4)strlen((const char *)colp->name_col),
                         (ub4)OCI_ATTR_NAME, ctlp->errhp_ctl));

    /* column type */
    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
              OCIAttrSet((dvoid *)colDesc, (ub4)OCI_DTYPE_PARAM,
                         (dvoid *)&colp->exttyp_col, (ub4)0,
                         (ub4)OCI_ATTR_DATA_TYPE, ctlp->errhp_ctl));

    /* max data size */
    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
              OCIAttrSet((dvoid *)colDesc, (ub4)OCI_DTYPE_PARAM,
                         (dvoid *)&fldp->maxlen_fld, (ub4)0,
                         (ub4)OCI_ATTR_DATA_SIZE, ctlp->errhp_ctl));

    /* If column is chrdate or date, set column (input field) date mask
     * to trigger client library to check string for a valid date.
     * Note: OCIAttrSet() may be called here w/ a null ptr or null string.
     */
    if (colp->date_col)
    {
      OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
                OCIAttrSet((dvoid *)colDesc, (ub4)OCI_DTYPE_PARAM,
                         (dvoid *)colp->datemask_col,
                         (colp->datemask_col) ?
                           (ub4)strlen((const char *)colp->datemask_col) : 0,
                         (ub4)OCI_ATTR_DATEFORMAT, ctlp->errhp_ctl));
    }

    if (colp->prec_col)
    {
      OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
                OCIAttrSet((dvoid *)colDesc, (ub4)OCI_DTYPE_PARAM,
                         (dvoid *)&colp->prec_col, (ub4)0,
                         (ub4)OCI_ATTR_PRECISION, ctlp->errhp_ctl));
    }

    if (colp->scale_col)
    {
      OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
                OCIAttrSet((dvoid *)colDesc, (ub4)OCI_DTYPE_PARAM,
                         (dvoid *)&colp->scale_col, (ub4)0,
                         (ub4)OCI_ATTR_SCALE, ctlp->errhp_ctl));
    }

    if (colp->csid_col)
    {
      OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
                OCIAttrSet((dvoid *)colDesc, (ub4)OCI_DTYPE_PARAM,
                         (dvoid *)&colp->csid_col, (ub4)0,
                         (ub4)OCI_ATTR_CHARSET_ID, ctlp->errhp_ctl));
    }

    /* free the parameter handle to the column descriptor */
    OCI_CHECK((dvoid *)0, 0, ociret, ctlp,
              OCIDescriptorFree((dvoid *)colDesc, OCI_DTYPE_PARAM));
  }

  /* read back some of the attributes for purpose of illustration */
  for (i = 0, pos = 1, colp = tblp->col_tbl, fldp = tblp->fld_tbl;
       i < tblp->ncol_tbl;
       i++, pos++, colp++, fldp++)
  {
    text *s;
    ub4   slen;
    ub4   maxdsz;
    ub2   dty;

    /* get parameter handle on the column */
    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
              OCIParamGet((CONST dvoid *)ctlp->colLstDesc_ctl,
                          (ub4)OCI_DTYPE_PARAM, ctlp->errhp_ctl,
                          (dvoid **)&colDesc, pos));


    /* get column name */
    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
              OCIAttrGet((dvoid *)colDesc,
                         OCI_DTYPE_PARAM,
                         (dvoid *)&s, (ub4 *)&slen,
                         OCI_ATTR_NAME, ctlp->errhp_ctl));

    /* check string length */
    if (slen != (ub4)strlen((const char *)colp->name_col))
    {
      fprintf(output_fp,
        "*** ERROR *** bad col name len in column parameter\n");
      fprintf(output_fp, "\texpected %d, got %d\n",
             (int)strlen((const char *)colp->name_col), (int)slen);
    }

    if (strncmp((const char *)s, (const char *)colp->name_col, (size_t)slen))
    {
      fprintf(output_fp,"*** ERROR *** bad column name in column parameter\n");
      fprintf(output_fp, "\texpected %s, got %s\n",
              (char *)colp->name_col, (char *)s);
    }

    /* get column type */
    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
              OCIAttrGet((dvoid *)colDesc,
                         OCI_DTYPE_PARAM,
                         (dvoid *)&dty, (ub4 *)0,
                         OCI_ATTR_DATA_TYPE, ctlp->errhp_ctl));
    if (dty != colp->exttyp_col)
    {
      fprintf(output_fp,
        "*** ERROR *** bad OCI_ATTR_DATA_TYPE in col param\n");
      fprintf(output_fp, "\tColumn name %s\n", colp->name_col);
      fprintf(output_fp, "\t\texpected %d, got %d\n",
              (int)colp->exttyp_col, (int)dty);
    }

    /* get the max data size */
    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
              OCIAttrGet((dvoid *)colDesc,
                         OCI_DTYPE_PARAM,
                         (dvoid *)&maxdsz, (ub4 *)0,
                         OCI_ATTR_DATA_SIZE, ctlp->errhp_ctl));
    if (maxdsz != fldp->maxlen_fld)
    {
      fprintf(output_fp,
        "*** ERROR *** bad OCI_ATTR_DATA_SIZE in col param\n");
      fprintf(output_fp, "\tColumn name %s\n", colp->name_col);
      fprintf(output_fp, "\t\texpected %d, got %d\n",
              (int)fldp->maxlen_fld, (int)maxdsz);
    }

    /* free the parameter handle to the column descriptor */
    OCI_CHECK((dvoid *)0, 0, ociret, ctlp,
              OCIDescriptorFree((dvoid *)colDesc, OCI_DTYPE_PARAM));
  }

  {
    char *vbuf;

    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
              OCIAttrSet((dvoid *)dpctx, (ub4)OCI_HTYPE_DIRPATH_CTX,
                         (dvoid *)&tblp->xfrsz_tbl,
                         (ub4)0, (ub4)OCI_ATTR_BUF_SIZE, ctlp->errhp_ctl));

    /* minimize read system calls */
    vbuf = (char *)malloc((size_t)tblp->xfrsz_tbl);
    if (vbuf != (char *)0)
      (void)setvbuf(stdin, vbuf, _IOFBF, (size_t)tblp->xfrsz_tbl);
  }

  /* prepare the load */
  OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
            OCIDirPathPrepare(dpctx, ctlp->svchp_ctl, ctlp->errhp_ctl));

  /* Allocate column array and stream handles.
   * Note that for the column array and stream handles
   * the parent handle is the direct path context.
   * Also note that Oracle errors are returned via the
   * environment handle associated with the direct path context.
   */
  OCI_CHECK(ctlp->envhp_ctl, OCI_HTYPE_ENV, ociret, ctlp,
            OCIHandleAlloc((dvoid *)ctlp->dpctx_ctl, (dvoid **)&ctlp->dpca_ctl,
                           (ub4)OCI_HTYPE_DIRPATH_COLUMN_ARRAY,
                           (size_t)0, (dvoid **)0));

  OCI_CHECK(ctlp->envhp_ctl, OCI_HTYPE_ENV, ociret, ctlp,
            OCIHandleAlloc((dvoid *)ctlp->dpctx_ctl,(dvoid **)&ctlp->dpstr_ctl,
                           (ub4)OCI_HTYPE_DIRPATH_STREAM,
                           (size_t)0, (dvoid **)0));

  /* get number of rows in the column array just allocated */
  OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
            OCIAttrGet((CONST dvoid *)(ctlp->dpca_ctl),
                       OCI_HTYPE_DIRPATH_COLUMN_ARRAY,
                       (dvoid *)(&ctlp->nrow_ctl), (ub4 *)0,
                       OCI_ATTR_NUM_ROWS, ctlp->errhp_ctl));

  /* get number of columns in the column array just allocated */
  OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
            OCIAttrGet((CONST dvoid *)(ctlp->dpca_ctl),
                       OCI_HTYPE_DIRPATH_COLUMN_ARRAY,
                       (dvoid *)(&ctlp->ncol_ctl), (ub4 *)0,
                       OCI_ATTR_NUM_COLS, ctlp->errhp_ctl));


  /* allocate buffer for input records */
  ctlp->inbuf_ctl = (ub1 *)malloc(ctlp->nrow_ctl * sessp->maxreclen_sess);
  if (ctlp->inbuf_ctl == (ub1 *)0)
  {
    perror("malloc");
    FATAL("init_load:malloc:inbuf_ctl alloc failure",
          ctlp->nrow_ctl * sessp->maxreclen_sess);
  }

  /* allocate Offset-TO-Record number mapping array */
  ctlp->otor_ctl = (ub4 *)malloc(ctlp->nrow_ctl * sizeof(ub4));
  if (ctlp->otor_ctl == (ub4 *)0)
  {
    perror("malloc");
    FATAL("init_load:malloc:otor_ctl alloc failure",
          ctlp->nrow_ctl * sizeof(ub4));
  }

  CLEAR_PCTX(ctlp->pctx_ctl);                  /* initialize partial context */

/*
  fprintf(output_fp, "init_load: %ld column array rows\n",
          (long)ctlp->nrow_ctl);
*/

  return;
}

/*------------------------------ simple_load ------------------------------*/
/*
  This function reads input records from 'inputfp', parses the input
  records into fields according to the field description given by
  tblp->fld_tbl, and loads the data into the database.

  LOBs can be loaded with this function in a piecewise manner.  This
  function is written as a state machine, which cycles through the
  following states:
    RESET, GET_RECORD, FIELD_SET, DO_CONVERT, DO_LOAD, END_OF_INPUT

  The normal case of all scalar data, where multiple records fit
  entirely in memory, cycles through the following states:
    RESET, [[GET_RECORD, FIELD_SET]+, DO_CONVERT, DO_LOAD]+, RESET

  The case of loading one or more LOB columns, which do not fit entirely
  in memory, has the following state transitions:
    RESET, GET_RECORD, [FIELD_SET, DO_CONVERT, DO_LOAD]+, RESET
  Note that the second and subsequent transitions to the FIELD_SET
  state have a partial record context.

  A mapping of column array offset to input record number (otor_ctl[])
  is maintained by this function for error reporting and recovery.
 */
STATICF void
simple_load(ctlp, tblp, sessp, inputfp)
struct loadctl *ctlp;
struct tbl     *tblp;
struct sess    *sessp;
FILE           *inputfp;
{
  sword  fsetrv;                              /* return value from field_set */
  sword  cvtrv;                              /* return value from do_convert */
  sword  ldrv;                                  /* return value from do_load */
  ub4    startoff;                     /* starting row offset for conversion */
  ub4    nxtLoadOff;                /* column array offset to be loaded next */
  ub4    rowCnt;                  /* count of rows populated in column array */
  ub4    cvtCnt;                                  /* count of rows converted */
  ub4    lastoff;                    /* last row offset used in column array */
  sword  state;                               /* current state machine state */
  sword  done;                          /* set to TRUE when load is complete */
  ub4    input_recnum;                        /* current input record number */
  ub4    load_recnum;   /* record number corresponding to last record loaded */
  ub4    err_recnum;                 /* record number corresponding to error */
  text   *recp;
  ub4    cvtcontcnt;                 /* # of times CONVERT_CONTINUE returned */

  /* set initial state */
  input_recnum = 0;
  load_recnum  = UB4MAXVAL;
  err_recnum   = 0;
  state        = RESET;
  fsetrv       = FIELD_SET_COMPLETE;
  cvtrv        = CONVERT_SUCCESS;
  ldrv         = LOAD_SUCCESS;
  done         = FALSE;
  cvtcontcnt   = 0;

  while (!done)
  {
    switch (state)
    {
    case RESET:    /* Reset column array and direct stream state to be empty */
    {
      startoff   = 0;             /* reset starting offset into column array */
      lastoff    = 0;                      /* last entry set of column array */
      rowCnt     = 0;                  /* count of rows partial and complete */
      cvtCnt     = 0;                             /* count of converted rows */
      nxtLoadOff = 0;

      /* Reset column array state in case a previous conversion needed
       * to be continued, or a row is expecting more data.
       */
      (void) OCIDirPathColArrayReset(ctlp->dpca_ctl, ctlp->errhp_ctl);

      /* Reset the stream state since we are starting a new stream
       * (i.e. don't want to append to existing data in the stream.)
       */
      (void) OCIDirPathStreamReset(ctlp->dpstr_ctl,  ctlp->errhp_ctl);

      state = GET_RECORD;                     /* get some more input records */
      /* FALLTHROUGH */
    }

    case GET_RECORD:
    {
      assert(lastoff < ctlp->nrow_ctl);                /* array bounds check */

      recp = (text *)(ctlp->inbuf_ctl + (lastoff * sessp->maxreclen_sess));

      if (fgets((char *)recp, (int)sessp->maxreclen_sess, inputfp)
                != (char *)NULL)
      {
        /* set column array offset to input record number map */
        ctlp->otor_ctl[lastoff] = ++input_recnum;
        if ((input_recnum % 10000) == 0)
          fprintf(output_fp, "record number: %d\n", (int)input_recnum);
        state = FIELD_SET;
        /* FALLTHROUGH */
      }
      else
      {
        if (lastoff)
          lastoff--;