bcp.c

ncbi源码

			 */

			row_pos += bcpcol->db_length;
		}
	}
	return row_pos;
}

/*
** Add variable size columns to the row
*/

static int _bcp_add_variable_columns(DBPROCESS *dbproc, BYTE *rowbuffer, int start)
{
	TDSSOCKET *tds = dbproc->tds_socket;
	BCP_COLINFO *bcpcol;
	TDS_NUMERIC *num;
	int row_pos = start;
	int cpbytes;
	int eod_ptr;
	BYTE offset_table[256];
	int offset_pos = 0;
	BYTE adjust_table[256];
	int adjust_pos = 0;
	int num_cols = 0;

	int i;

	for (i = 0; i < dbproc->bcp_colcount; i++) {
		bcpcol = dbproc->bcp_columns[i];

        if (is_nullable_type(bcpcol->db_type) || 
            bcpcol->db_nullable) {

			if (!(bcpcol->db_nullable) && bcpcol->data_size == 0) {
                _bcp_err_handler(dbproc, BCPEBCNN);
				return FAIL;
			}

			if (is_blob_type(bcpcol->db_type)) {

				/* no need to copy they are all zero bytes */

				cpbytes = 16;
				/* save for data write */
				bcpcol->txptr_offset = row_pos;
            } 
            else if (is_numeric_type(bcpcol->db_type)) {
				num = (TDS_NUMERIC *) bcpcol->data;
                cpbytes = g__numeric_bytes_per_prec[num->precision];
				memcpy(&rowbuffer[row_pos], num->array, cpbytes);
            }
            else {
				/* compute the length to copy to the row ** buffer */

                cpbytes = bcpcol->data_size > bcpcol->db_length 
                    ? bcpcol->db_length : bcpcol->data_size;
				memcpy(&rowbuffer[row_pos], bcpcol->data, cpbytes);

			}

			/* update offset and adjust tables (if necessary) */
			offset_table[offset_pos++] = row_pos % 256;
            if (row_pos > 255 && 
                (adjust_pos==0 || row_pos/256 != adjust_table[adjust_pos])) {
				adjust_table[adjust_pos++] = row_pos / 256;
			}

			num_cols++;
			row_pos += cpbytes;
		}
	}

	eod_ptr = row_pos;


	/* write the marker */
	if (IS_TDS50(tds))
		rowbuffer[row_pos++] = num_cols + 1;

	/* write the adjust table (right to left) */
	for (i = adjust_pos - 1; i >= 0; i--) {
		/* fprintf(stderr,"adjust %d\n",adjust_table[i]); */
		rowbuffer[row_pos++] = adjust_table[i];
	}

	/* the EOD (end of data) pointer */
	rowbuffer[row_pos++] = eod_ptr;

	/* write the offset table (right to left) */
	for (i = offset_pos - 1; i >= 0; i--) {
		/* fprintf(stderr,"offset %d\n",offset_table[i]); */
		rowbuffer[row_pos++] = offset_table[i];
	}

	return row_pos;
}

RETCODE bcp_sendrow(DBPROCESS *dbproc)
{

TDSSOCKET *tds = dbproc->tds_socket;
BCP_COLINFO *bcpcol;
    BCP_HOSTCOLINFO *hostcol;
    TDSCOLINFO      *curcol;

int i;
    int collen;
/* FIX ME -- calculate dynamically */
unsigned char rowbuffer[ROWBUF_SIZE];
int row_pos;
int row_sz_pos;
TDS_SMALLINT row_size;
    TDS_SMALLINT row_size_saved;

    int marker;
int blob_cols = 0;

unsigned char CHARBIN_NULL[] = { 0xff, 0xff };
unsigned char GEN_NULL = 0x00;

TDS_NUMERIC *num;
TDS_TINYINT numeric_size;

unsigned char row_token = 0xd1;

	if (dbproc->bcp_direction == 0) {
        _bcp_err_handler(dbproc, BCPEBCPI);
		return FAIL;
	}
	if (dbproc->bcp_hostfile != (char *) NULL) {
        _bcp_err_handler(dbproc, BCPEBCPB);
		return FAIL;
	}

	if (dbproc->bcp_direction != DB_IN) {
        _bcp_err_handler(dbproc, BCPEBCPN);
		return FAIL;
	}

	/* The first time sendrow is called after bcp_init,  */
	/* there is a certain amount of initialisation to be */
	/* done...                                           */

	if (!(dbproc->sendrow_init)) {

		/* first call the start_copy function, which will */
		/* retrieve details of the database table columns */

		if (_bcp_start_copy_in(dbproc) == FAIL)
			return (FAIL);

#ifndef NCBI_FTDS
        /* Next, allocate the storage to hold data about the */
        /* program variable data - these must be the same in */
        /* number as the table columns...                    */

        dbproc->host_colcount = dbproc->bcp_colcount;
        dbproc->host_columns  = (BCP_HOSTCOLINFO **) malloc(dbproc->host_colcount * sizeof(BCP_HOSTCOLINFO *));

        for ( i = 0; i < dbproc->host_colcount; i++ ) {
            dbproc->host_columns[i] = (BCP_HOSTCOLINFO *) malloc(sizeof(BCP_HOSTCOLINFO));
            memset(dbproc->host_columns[i], '\0', sizeof(BCP_HOSTCOLINFO));
        }
#endif

		/* set packet type to send bulk data */
		tds->out_flag = 0x07;

		if (IS_TDS7_PLUS(tds)) {
			_bcp_send_colmetadata(dbproc);
		}

		dbproc->sendrow_init = 1;

	}


	if (_bcp_get_prog_data(dbproc) == SUCCEED) {

		if (IS_TDS7_PLUS(tds)) {
			tds_put_byte(tds, row_token);	/* 0xd1 */

#ifdef NCBI_FTDS
            dbproc->curr_text_col= 0;
#endif
			for (i = 0; i < dbproc->bcp_colcount; i++) {
				bcpcol = dbproc->bcp_columns[i];
#ifdef NCBI_FTDS
                if(dbproc->host_columns[i]->datatype == 0) continue;

                if(is_blob_type(bcpcol->db_type)) continue;
#endif

                if (bcpcol->data_size == 0 ) /* Are we trying to insert a NULL ? */ {
                    if(bcpcol->db_nullable)   /* is the column nullable ? */ {
                        if (bcpcol->db_type_save == XSYBCHAR || 
                            bcpcol->db_type_save == XSYBVARCHAR ||
                            bcpcol->db_type_save == XSYBBINARY ||
                            bcpcol->db_type_save == XSYBVARBINARY|| 
                            bcpcol->db_type_save == XSYBNCHAR || 
                            bcpcol->db_type_save == XSYBNVARCHAR) {
						tds_put_n(tds, CHARBIN_NULL, 2);
                        }
                        else {
						tds_put_byte(tds, GEN_NULL);
					}
                    }
                    else {
                        _bcp_err_handler(dbproc, BCPEBCNN);
                        return FAIL;
                    }
                }
                else {   
					switch (bcpcol->db_varint_size) {
                    case 4: tds_put_int(tds, bcpcol->data_size);
						break;
                    case 2: tds_put_smallint(tds, bcpcol->data_size);
						break;
                    case 1: if (is_numeric_type(bcpcol->db_type)) {
                        numeric_size = g__numeric_bytes_per_prec[bcpcol->db_prec];
							tds_put_byte(tds, numeric_size);
                    }
                    else
							tds_put_byte(tds, bcpcol->data_size);
						break;
                    case 0: break;
					}

#if WORDS_BIGENDIAN
               if (tds->emul_little_endian) {
                    tds_swap_datatype(tds_get_conversion_type(bcpcol->db_type, bcpcol->db_length),
                                      bcpcol->data);
               }
#endif

					if (is_numeric_type(bcpcol->db_type)) {
						num = (TDS_NUMERIC *) bcpcol->data;
#ifdef NCBI_FTDS
                        tds_put_n(tds, num->array, g__numeric_bytes_per_prec[bcpcol->db_prec]);
#else
                        tds_put_n(tds, num->array, g__numeric_bytes_per_prec[num->precision]);
#endif
                    }
#if 0
                    else if(is_blob_type(bcpcol->db_type)) { //add blob data
                        int ss;
                        char blob_buff[256];
                        for(marker= 0; marker < 256;) {
                            blob_buff[marker++]= 'T';
                            blob_buff[marker++]= 'E';
                            blob_buff[marker++]= 'S';
                            blob_buff[marker++]= 't';
                        }
                        for(marker = bcpcol->data_size; marker > 0; marker-= ss) {
                            ss= (marker > 256)? 256 : marker;
                            tds_put_n(tds, blob_buff, ss);
                        }
                    }
#endif
                    else
						tds_put_n(tds, bcpcol->data, bcpcol->data_size);
				}
			}
			return SUCCEED;
        }
        else    /* Not TDS 7 or 8 */    {
			memset(rowbuffer, '\0', ROWBUF_SIZE);	/* zero the rowbuffer */

			/* offset 0 = number of var columns */
			/* offset 1 = row number.  zeroed (datasever assigns) */
			row_pos = 2;

			rowbuffer[0] = dbproc->var_cols;

            if (row_pos = _bcp_add_fixed_columns(dbproc, rowbuffer, row_pos) == FAIL)
				return (FAIL);


			row_sz_pos = row_pos;


			if (dbproc->var_cols) {

				row_pos += 2;	/* skip over row length */

				if ((row_pos = _bcp_add_variable_columns(dbproc, rowbuffer, row_pos)) == FAIL)
					return (FAIL);
			}

			row_size = row_pos;


			if (dbproc->var_cols)
				memcpy(&rowbuffer[row_sz_pos], &row_size, sizeof(row_size));

			tds_put_smallint(tds, row_size);
			tds_put_n(tds, rowbuffer, row_size);

			/* row is done, now handle any text/image data */

			blob_cols = 0;

			for (i = 0; i < dbproc->bcp_colcount; i++) {
				bcpcol = dbproc->bcp_columns[i];
				if (is_blob_type(bcpcol->db_type)) {
					/* unknown but zero */
					tds_put_smallint(tds, 0);
					tds_put_byte(tds, bcpcol->db_type);
					tds_put_byte(tds, 0xff - blob_cols);
					/* offset of txptr we stashed during variable
                    ** column processing */
					tds_put_smallint(tds, bcpcol->txptr_offset);
					tds_put_int(tds, bcpcol->data_size);
					tds_put_n(tds, bcpcol->data, bcpcol->data_size);
					blob_cols++;
				}
			}
			return SUCCEED;
		}
	}

	return FAIL;
}


static RETCODE _bcp_exec_in(DBPROCESS *dbproc, DBINT *rows_copied)
{
	FILE *hostfile, *errfile;
	TDSSOCKET *tds = dbproc->tds_socket;
	BCP_COLINFO *bcpcol;

	/* FIX ME -- calculate dynamically */
	unsigned char rowbuffer[ROWBUF_SIZE];
	int row_pos;

	/* end of data pointer...the last byte of var data before the adjust table */

	int row_sz_pos;
	TDS_SMALLINT row_size;

	const unsigned char CHARBIN_NULL[] = { 0xff, 0xff };
	const unsigned char GEN_NULL = 0x00;
	const unsigned char row_token = 0xd1;

	TDS_NUMERIC *num;
	TDS_TINYINT numeric_size;

	int i;
	int marker;
	int blob_cols = 0;
	int row_of_hostfile;
	int rows_written_so_far;
	int rows_copied_this_batch = 0;
	int rows_copied_so_far = 0;

	if (!(hostfile = fopen(dbproc->bcp_hostfile, "r"))) {
        _bcp_err_handler(dbproc, BCPEBCUO);
		return FAIL;
	}

	if (dbproc->bcp_errorfile) {
		if (!(errfile = fopen(dbproc->bcp_errorfile, "w"))) {
			_bcp_err_handler(dbproc, SYBEBUOE);
			return FAIL;
		}
	}

	if (_bcp_start_copy_in(dbproc) == FAIL)
		return (FAIL);

	tds->out_flag = 0x07;

	if (IS_TDS7_PLUS(tds)) {
		_bcp_send_colmetadata(dbproc);
	}

	row_of_hostfile = 0;

	rows_written_so_far = 0;

	while (_bcp_read_hostfile(dbproc, hostfile/*, errfile*/) == SUCCEED) {

		row_of_hostfile++;

		if ((dbproc->firstrow == 0 && dbproc->lastrow == 0) ||
		    ((dbproc->firstrow > 0 && row_of_hostfile >= dbproc->firstrow)
		     && (dbproc->lastrow > 0 && row_of_hostfile <= dbproc->lastrow))
			) {

			if (IS_TDS7_PLUS(tds)) {
				tds_put_byte(tds, row_token);	/* 0xd1 */

				for (i = 0; i < dbproc->bcp_colcount; i++) {
					bcpcol = dbproc->bcp_columns[i];

					if (bcpcol->data_size == 0) {	/* Are we trying to insert a NULL ? */
						if (!bcpcol->db_nullable){	
							/* too bad if the column is not nullable */
							fprintf(errfile, "Column is not nullable. Row %d, Column %d\n", 
								row_of_hostfile, i);
							fwrite(bcpcol->data, bcpcol->data_size, 1, errfile);
							_bcp_err_handler(dbproc, BCPEBCNN);
							return FAIL;
						}
						
						switch (bcpcol->db_type_save) {
						case XSYBCHAR:
						case XSYBVARCHAR:
						case XSYBBINARY:
						case XSYBVARBINARY:
						case XSYBNCHAR:
						case XSYBNVARCHAR:
							tds_put_n(tds, CHARBIN_NULL, 2);
							break;
						default :
							tds_put_byte(tds, GEN_NULL);
							break;
						}
					} else {
						switch (bcpcol->db_varint_size) {
						case 4:
							tds_put_int(tds, bcpcol->data_size);
							break;
						case 2:
							tds_put_smallint(tds, bcpcol->data_size);
							break;
						case 1:
							if (is_numeric_type(bcpcol->db_type)) {
								numeric_size = tds_numeric_bytes_per_prec[bcpcol->db_prec];
								tds_put_byte(tds, numeric_size);
							} else
								tds_put_byte(tds, bcpcol->data_size);
							break;
						case 0:
							break;
						default:
                      /*assert(bcpcol->db_varint_size <= 4);*/
							break;
						}

#ifdef WORDS_BIGENDIAN
						tds_swap_datatype
							(tds_get_conversion_type(bcpcol->db_type, bcpcol->db_length), bcpcol->data);
#else
						if (is_numeric_type(bcpcol->db_type)) {
							tds_swap_datatype
								(tds_get_conversion_type
								 (bcpcol->db_type, bcpcol->db_length), bcpcol->data);
						}
#endif

						if (is_numeric_type(bcpcol->db_type)) {
							num = (TDS_NUMERIC *) bcpcol->data;
                            tds_put_n(tds, num->array, g__numeric_bytes_per_prec[num->precision]);
                        }
                        else
							tds_put_n(tds, bcpcol->data, bcpcol->data_size);
					}
				}
            }
            else    /* Not TDS 7 or 8 */    {
				memset(rowbuffer, '\0', ROWBUF_SIZE);	/* zero the rowbuffer */

				/* offset 0 = number of var columns */
				/* offset 1 = row number.  zeroed (datasever assigns) */

				row_pos = 2;

				rowbuffer[0] = dbproc->var_cols;

				if ((row_pos = _bcp_add_fixed_columns(dbproc, rowbuffer, row_pos)) == FAIL)
					return (FAIL);

				row_sz_pos = row_pos;

				if (dbproc->var_cols) {

					row_pos += 2;	/* skip over row length */

                    if ((row_pos = _bcp_add_variable_columns(dbproc, rowbuffer, row_pos)) == FAIL)
						return (FAIL);
				}

				row_size = row_pos;


				if (dbproc->var_cols)
					memcpy(&rowbuffer[row_sz_pos], &row_size, sizeof(row_size));

				tds_put_smallint(tds, row_size);
				tds_put_n(tds, rowbuffer, row_size);

				/* row is done, now handle any text/image data */

				blob_cols = 0;

				for (i = 0; i < dbproc->bcp_colcount; i++) {
					bcpcol = dbproc->bcp_columns[i];
					if (is_blob_type(bcpcol->db_type)) {
						/* unknown but zero */
						tds_put_smallint(tds, 0);
						tds_put_byte(tds, bcpcol->db_type);
						tds_put_byte(tds, 0xff - blob_cols);
						/* offset of txptr we stashed during variable