cdemodp.c

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

  fprintf(output_fp, "do_convert returned CONVERT_CONTINUE %ld times\n",
          (long)cvtcontcnt);
*/

  fprintf(output_fp, "Number of input records processed = %ld\n",
          (long)input_recnum);
}

STATICF void
finish_load(ctlp)
struct loadctl *ctlp;
{
  sword ociret;                                 /* return code from OCI call */

  /* Execute load finishing logic without freeing server data structures
   * related to the load.
   */
  OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
            OCIDirPathDataSave(ctlp->dpctx_ctl, ctlp->errhp_ctl,
                               (ub4)OCI_DIRPATH_DATASAVE_FINISH));

  /* free up server data structures for the load. */
  OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ociret, ctlp,
            OCIDirPathFinish(ctlp->dpctx_ctl, ctlp->errhp_ctl));
}

/*------------------------------ do_convert ------------------------------*/
/*
 NAME:     do_convert

 FUNCTION: Convert the data in the column array to stream format.

 RETURNS:
   CONVERT_SUCCESS:
     All data in the column array has been successfully converted.
     *cvtCntp is the number of rows successfully converted.
   CONVERT_ERROR:
     Conversion error occurred on the row after the last successfully
     converted row.
     Client Action:
       Continue converting the column array by calling this function
       again with startoff adjusted to skip over the row in error.

   CONVERT_NEED_DATA:
     All data in the column array has been converted, but the last
     column processed was marked as a partial.
   CONVERT_CONTINUE:
     Not all of the data in the column array has been converted due to
     lack of space in the stream buffer.
     Client Action:
       Load the converted stream data, reset the stream, and call this
       function again without modifying the column array and setting
       startoff to the appropriate position in the array.
 */
sword
do_convert(ctlp, startoff, rowcnt, cvtCntp, badcoffp)
struct loadctl *ctlp;             /* pointer to control structure   (IN/OUT) */
ub4             rowcnt;           /* number of rows in column array     (IN) */
ub4             startoff;     /* starting row offset into column array  (IN) */
ub4            *cvtCntp;       /* count of rows successfully converted (OUT) */
ub2            *badcoffp;   /* column offset into col array of bad col (OUT) */
{
  sword retval = CONVERT_SUCCESS;
  sword ocierr, ocierr2;
  ub2   badcol = 0;

  *cvtCntp = 0;

  if (startoff >= rowcnt)
    FATAL("DO_CONVERT: bad startoff", startoff);

  if (rowcnt)
  {
    /* convert array to stream, filter out bad records */
    ocierr = OCIDirPathColArrayToStream(ctlp->dpca_ctl, ctlp->dpctx_ctl,
                                        ctlp->dpstr_ctl, ctlp->errhp_ctl,
                                        rowcnt, startoff);
    switch (ocierr)
    {
    case OCI_SUCCESS:        /* everything succesfully converted to stream */
      retval = CONVERT_SUCCESS;
      break;

    case OCI_ERROR:            /* some error, most likely a conversion error */
      /* Tell the caller that a conversion error occurred along
       * with the number of rows successfully converted (*cvtCntp).
       * Note that the caller is responsible for adjusting startoff
       * accordingly and calling us again to resume conversion of
       * the remaining rows.
       */
      retval  = CONVERT_ERROR;                         /* conversion error */
      break;

    case OCI_CONTINUE:                              /* stream buffer is full */
      /* The stream buffer could not contain all of the data in
       * the column array.
       * The client should load the converted data, and loop
       * back to convert the remaining data in the column array.
       */
      retval  = CONVERT_CONTINUE;
      break;

    case OCI_NEED_DATA:                        /* partial column encountered */
      /* Everything converted, but have a partial column.
       * Load this stream, and return to caller for next piece.
       */
      retval = CONVERT_NEED_DATA;
      break;

    default:                                 /* unexpected OCI return value! */
      FATAL("do_convert:OCIDirPathColArrayToStream:Unexpected OCI return code",
             ocierr);
      /* NOTREACHED */
      break;
    }

    OCI_CHECK(ctlp->errhp2_ctl, OCI_HTYPE_ERROR, ocierr2, ctlp,
              OCIAttrGet((CONST dvoid *)ctlp->dpca_ctl,
                         OCI_HTYPE_DIRPATH_COLUMN_ARRAY,
                         (dvoid *)(cvtCntp), (ub4 *)0,
                         OCI_ATTR_ROW_COUNT, ctlp->errhp2_ctl));

    OCI_CHECK(ctlp->errhp2_ctl, OCI_HTYPE_ERROR, ocierr2, ctlp,
              OCIAttrGet((CONST dvoid *)ctlp->dpca_ctl,
                         OCI_HTYPE_DIRPATH_COLUMN_ARRAY,
                         (dvoid *)(&badcol), (ub4 *)0,
                         OCI_ATTR_COL_COUNT, ctlp->errhp2_ctl));
  }

  *badcoffp = badcol;

  return retval;
}

/*------------------------------ do_load ------------------------------*/
/*
 NAME:     do_load

 FUNCTION: Load a direct path stream.

 PARAMETERS:
   ctlp:     Pointer to control structure.  Note that errhp_ctl
             contains error information when an error is returned
             from this function.

   loadCntp: Count of rows loaded on this call.

 RETURNS:
   LOAD_SUCCESS:
     All data loaded succesfully.
     Client Action:
       Supply another stream and call again, or finish the load.

   LOAD_ERROR:
     Error while loading occured.  *loadCntp is the number of
     rows successfully loaded this call.
     Client Action:
       Use *loadCntp to compute current column array position and
       map the column array position to the input record and reject
       the record.

       if (this is a continuation of a row)
       {
         /o server has data for this row buffered o/
         flush the row data
       }

       if (end-of-stream has not been reached)
       {
         call this function again,
         stream loading will resume with the next row in the stream.
       }
       else if (end-of-stream has been reached)
       {
         build another stream and call this function again,
         or finish the load.
       }

   LOAD_NEED_DATA:
     Last row was not complete.
     Client Action:
       Caller needs to supply more data for the row (a column is
       being pieced.)  Note that the row offset can be determined
       by either the cvtCnt returned from do_convert, or from the
       loadCntp returned by do_load.  The column offset for the
       column being pieced is available as an attribute of
       the column array.
 */
sword
do_load(ctlp, loadCntp)
struct loadctl *ctlp;             /* pointer to control structure   (IN/OUT) */
ub4            *loadCntp;                     /* number of rows loaded (OUT) */
{
  sword ocierr;                                          /* OCI return value */
  sword retval;                           /* return value from this function */
  sword getRowCnt = FALSE;                       /* return row count if TRUE */

  if (loadCntp != (ub4 *)0)
  {
    *loadCntp = 0;
    getRowCnt = TRUE;
  }

  /* Load the stream.
   * Note that the position in the stream is maintained internally to
   * the stream handle, along with offset information for the column
   * array which produced the stream.  When the conversion to stream
   * format is done, the data is appended to the stream.  It is the
   * responsibility of the caller to reset the stream when appropriate.
   * On errors, the position is moved to the next row, or the end of
   * the stream if the error occurs on the last row.  The next LoadStream
   * call will start on the next row, if any.
   * If a LoadStream call is made, and end of stream has been reached,
   * OCI_NO_DATA is returned.
   */

#if 1
  ocierr = OCIDirPathLoadStream(ctlp->dpctx_ctl, ctlp->dpstr_ctl,
                                ctlp->errhp_ctl);
#else
  {
    ub1 *bufaddr;
    ub4  buflen;

    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ocierr, ctlp,
              OCIAttrGet((CONST dvoid *)(ctlp->dpstr_ctl),
                         OCI_HTYPE_DIRPATH_STREAM,
                         (dvoid *)&bufaddr, (ub4 *)0,
                         OCI_ATTR_BUF_ADDR, ctlp->errhp_ctl));
    OCI_CHECK(ctlp->errhp_ctl, OCI_HTYPE_ERROR, ocierr, ctlp,
              OCIAttrGet((CONST dvoid *)(ctlp->dpstr_ctl),
                         OCI_HTYPE_DIRPATH_STREAM,
                         (dvoid *)&buflen, (ub4 *)0,
                         OCI_ATTR_BUF_SIZE, ctlp->errhp_ctl));
    write(1, (char *)bufaddr, (int)buflen);
    fprintf(output_fp, "Wrote %d bytes from stream\n", (int)buflen);
    getRowCnt = FALSE;
  }
#endif

  switch (ocierr)
  {
    case OCI_SUCCESS:
      /* all data succcesfully loaded */
      retval    = LOAD_SUCCESS;
      break;

    case OCI_ERROR:
      /* Error occurred while loading: could be a partition mapping
       * error, null constraint violation, or an out of space
       * condition.  In any case, we return the number of rows
       * processed (successfully loaded).
       */
      retval    = LOAD_ERROR;
      break;

    case OCI_NEED_DATA:
      /* Last row was not complete.
       * The caller needs to supply another piece.
       */
      retval    = LOAD_NEED_DATA;
      break;

    case OCI_NO_DATA:
      /* the stream was empty */
      retval    = LOAD_NO_DATA;
      break;

    default:
      FATAL("do_load:OCIDirPathLoadStream:Unexpected OCI return code", ocierr);
      /* NOTREACHED */
      break;
  }

  if (getRowCnt)
  {
    sword  ocierr2;

    OCI_CHECK(ctlp->errhp2_ctl, OCI_HTYPE_ERROR, ocierr2, ctlp,
              OCIAttrGet((CONST dvoid *)(ctlp->dpstr_ctl),
                         OCI_HTYPE_DIRPATH_STREAM,
                         (dvoid *)loadCntp, (ub4 *)0,
                         OCI_ATTR_ROW_COUNT, ctlp->errhp2_ctl));
  }

  return retval;
}


/*------------------------------ field_flush ------------------------------*/
/*
 NAME      field_flush

 FUNCTION: Helper function which cleans up the partial context
           state, and clears it.
 */
STATICF void
field_flush(ctlp, rowoff)
struct loadctl *ctlp;
ub4             rowoff;
{
  if (ctlp->pctx_ctl.valid_pctx)
  {
    /* Partial context is valid; make sure the request is
     * for the context corresponding to the current row.
     */
    assert(rowoff == ctlp->pctx_ctl.row_pctx);
    (void) close(ctlp->pctx_ctl.fd_pctx);
    free((void *)ctlp->pctx_ctl.fnm_pctx);
  }
  CLEAR_PCTX(ctlp->pctx_ctl);
}


/*------------------------------ field_set ------------------------------*/
/*
 NAME:     field_set

 FUNCTION: Set the input data fields to their corresponding
           data columns.

   Simple field setting.
   Computes address and length of fields in the input record,
   and sets the corresponding column array entry for each
   input field.

   This function only deals with positional fields.

   Leading white space is trimmed from the field if FLD_STRIP_LEAD_BLANK
   is set.

   Trailing white space is trimmed from the field if FLD_STRIP_TRAIL_BLANK
   is set.

   Fields which consist of all spaces are loaded as null columns.

   Fields which are marked as FLD_OUTOFLINE are interpreted
   as being file names which can be passed directly to an
   open system call.

 RETURNS:
   FIELD_SET_COMPLETE:
     All fields are complete, the partial context is not valid.

   FIELD_SET_BUF
     All fields are complete, the partial context is not valid, but
     data is buffered in a secondary buffer and the column array has
     one or more pointers into the secondary buffer.  The caller