bcp.c

ncbi源码

                        ** 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++;
					}
				}
			}	/* Not TDS  7.0 or 8.0 */

			rows_written_so_far++;

			if (dbproc->bcpbatch > 0 && rows_written_so_far == dbproc->bcpbatch) {
				rows_written_so_far = 0;

				tds_flush_packet(tds);

				do {
					marker = tds_get_byte(tds);
					if (marker == TDS_DONE_TOKEN) {
						tds_process_end(tds, marker, NULL,NULL);
						rows_copied_this_batch = tds->rows_affected;
						rows_copied_so_far += rows_copied_this_batch;
                    } 
                    else {
						tds_process_default_tokens(tds, marker);
					}
				} while (marker != TDS_DONE_TOKEN);

				_bcp_start_new_batch(dbproc);

			}
		}
	}

	if (fclose(hostfile) != 0) {
        _bcp_err_handler(dbproc, BCPEBCUC);
		return (FAIL);
	}

	tds_flush_packet(tds);

	do {
		marker = tds_get_byte(tds);
		if (marker == TDS_DONE_TOKEN) {
			tds_process_end(tds, marker, NULL,NULL);
			rows_copied_this_batch = tds->rows_affected;
			rows_copied_so_far += rows_copied_this_batch;
			*rows_copied = rows_copied_so_far;
        } 
        else {
			tds_process_default_tokens(tds, marker);
		}
	} while (marker != TDS_DONE_TOKEN);

	return SUCCEED;
}

RETCODE bcp_exec(DBPROCESS *dbproc, DBINT *rows_copied)
{

RETCODE ret;

	if (dbproc->bcp_direction == 0) {
        _bcp_err_handler(dbproc, BCPEBCPI);
		return FAIL;
	}
	if (dbproc->bcp_hostfile) {
		if (dbproc->bcp_direction == DB_OUT) {
			ret = _bcp_exec_out(dbproc, rows_copied);
        } 
        else if (dbproc->bcp_direction==DB_IN) {
			ret = _bcp_exec_in(dbproc, rows_copied);
		}
		_bcp_clear_storage(dbproc);
		return ret;
    }
    else {
        _bcp_err_handler(dbproc, BCPEBCVH );
		return FAIL;
	}
}

static RETCODE _bcp_start_copy_in(DBPROCESS *dbproc)
{

TDSSOCKET *tds = dbproc->tds_socket;
BCP_COLINFO *bcpcol;
    TDSCOLINFO      *curcol;
    TDSRESULTINFO   *resinfo;

int i;
int marker;
    char colstatus;

#ifdef NCBI_FTDS
    char query[2048];
    char colclause[2048];
#else
    char query[256];
    char colclause[256];
#endif
int firstcol;

	if (IS_TDS7_PLUS(tds)) {
		int erc;
		char *hint;
        firstcol = 1;
#ifdef NCBI_FTDS
        colclause[0]= '\0';
        for(i= 0; i < dbproc->bcp_colcount; i++) {
            if((!is_blob_type(dbproc->bcp_columns[i]->db_type)) &&
               (dbproc->host_columns[i]->datatype != 0)){
                _bcp_build_bulk_insert_stmt(colclause, dbproc->bcp_columns[i], firstcol);
                firstcol = 0;
            }
        }

        for(i= 0; i < dbproc->bcp_colcount; i++) {
            if(is_blob_type(dbproc->bcp_columns[i]->db_type) &&
               (dbproc->host_columns[i]->datatype != 0)) {
                _bcp_build_bulk_insert_stmt(colclause, dbproc->bcp_columns[i], firstcol);
                firstcol = 0;
            }
        }
#else
		strcpy(colclause, "");

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

			if (IS_TDS7_PLUS(tds)) {
				_bcp_build_bulk_insert_stmt(colclause, bcpcol, firstcol);
				firstcol = 0;
			}
		}
#endif
		if (dbproc->bcp_hint) {
			if (asprintf(&hint, " with (%s)", dbproc->bcp_hint) < 0) {
				return FAIL;
			}
		} else {
			hint = strdup("");
		}
		if (!hint) return FAIL;

		sprintf(query, "insert bulk %s (%s)%s", dbproc->bcp_tablename, colclause, hint);

		free(hint); 

	} 
   else {
       sprintf(query, "insert bulk %s", dbproc->bcp_tablename);
	}

	tds_submit_query(tds, query);

	/* save the statement for later... */

    dbproc->bcp_insert_stmt = (char *) malloc(strlen(query)+1);
    strcpy(dbproc->bcp_insert_stmt, query);

	/* In TDS 5 we get the column information as a */
	/* result set from the "insert bulk" command.  */
	/* we're going to ignore this....              */

	if (IS_TDS50(tds)) {
        if (tds_process_result_tokens(tds) == TDS_FAIL) {
			return FAIL;
		}
		if (!tds->res_info) {
			return FAIL;
		}

        while (tds_process_row_tokens(tds) == SUCCEED); 

    }
    else {
		marker = tds_get_byte(tds);
		tds_process_default_tokens(tds, marker);
		if (!is_end_token(marker)) {
			return FAIL;
		}
	}

	/* work out the number of "variable" columns -         */
	/* either varying length type e.g. varchar or nullable */

	dbproc->var_cols = 0;

	for (i = 0; i < dbproc->bcp_colcount; i++) {

		bcpcol = dbproc->bcp_columns[i];
        if (is_nullable_type(bcpcol->db_type) ||
            bcpcol->db_nullable) {
			dbproc->var_cols++;
		}
	}

	return SUCCEED;

}

static RETCODE _bcp_build_bulk_insert_stmt(char *clause, BCP_COLINFO *bcpcol, int first)
{
	char column_type[32];

	switch (bcpcol->db_type_save) {
    case SYBINT1:      strcpy(column_type,"tinyint");
		break;
    case SYBBIT:       strcpy(column_type,"bit");
		break;
    case SYBINT2:      strcpy(column_type,"smallint");
		break;
    case SYBINT4:      strcpy(column_type,"int");
		break;
    case SYBDATETIME:  strcpy(column_type,"datetime");
		break;
    case SYBDATETIME4: strcpy(column_type,"smalldatetime");
		break;
    case SYBREAL:      strcpy(column_type,"real");
		break;
    case SYBMONEY:     strcpy(column_type,"money");
		break;
    case SYBMONEY4:    strcpy(column_type,"smallmoney");
		break;
    case SYBFLT8:      strcpy(column_type,"float");
		break;
    case SYBINT8:      strcpy(column_type,"bigint");
		break;

    case SYBINTN:      switch (bcpcol->db_length) {
    case 1: strcpy(column_type,"tinyint");
			break;
    case 2: strcpy(column_type,"smallint");
			break;
    case 4: strcpy(column_type,"int");
			break;
    case 8: strcpy(column_type,"bigint");
			break;
		}
		break;

    case SYBBITN:      strcpy(column_type,"bit");
		break;
    case SYBFLTN:      switch (bcpcol->db_length) {
    case 4: strcpy(column_type,"real");
			break;
    case 8: strcpy(column_type,"float");
			break;
		}
		break;
    case SYBMONEYN:    switch (bcpcol->db_length) {
    case 4: strcpy(column_type,"smallmoney");
			break;
    case 8: strcpy(column_type,"money");
			break;
		}
		break;
    case SYBDATETIMN:  switch (bcpcol->db_length) {
    case 4: strcpy(column_type,"smalldatetime");
			break;
    case 8: strcpy(column_type,"datetime");
			break;
		}
		break;
    case SYBDECIMAL:   sprintf(column_type,"decimal(%d,%d)", bcpcol->db_prec, bcpcol->db_scale);
		break;
    case SYBNUMERIC:   sprintf(column_type,"numeric(%d,%d)", bcpcol->db_prec, bcpcol->db_scale);
		break;

    case XSYBVARBINARY: sprintf(column_type,"varbinary(%d)", bcpcol->db_length);
		break;
    case XSYBVARCHAR  : sprintf(column_type,"varchar(%d)", bcpcol->db_length);
		break;
    case XSYBBINARY   : sprintf(column_type,"binary(%d)", bcpcol->db_length);
		break;
    case XSYBCHAR     : sprintf(column_type,"char(%d)", bcpcol->db_length);
		break;
    case SYBTEXT      : sprintf(column_type,"text");
		break;
    case SYBIMAGE     : sprintf(column_type,"image");
		break;
    case XSYBNVARCHAR : sprintf(column_type,"nvarchar(%d)", bcpcol->db_length);
		break;
    case XSYBNCHAR    : sprintf(column_type,"nchar(%d)", bcpcol->db_length);
		break;
    case SYBNTEXT     : sprintf(column_type,"ntext");
		break;
	}

	if (!first)
		strcat(clause, ", ");

	strcat(clause, bcpcol->db_name);
	strcat(clause, " ");

	strcat(clause, column_type);

#ifdef NCBI_FTDS
	return SUCCEED;
#endif
}

static RETCODE _bcp_start_new_batch(DBPROCESS *dbproc)
{

TDSSOCKET *tds = dbproc->tds_socket;
int marker;

	if (IS_TDS50(tds)) {

		tds_submit_query(tds, dbproc->bcp_insert_stmt);

        if (tds_process_result_tokens(tds) == TDS_FAIL) {
			return FAIL;
		}
		if (!tds->res_info) {
			return FAIL;
		}

        while (tds_process_row_tokens(tds) == SUCCEED); 

    }   
    else {
		tds_submit_query(tds, dbproc->bcp_insert_stmt);

		marker = tds_get_byte(tds);
		tds_process_default_tokens(tds, marker);
		if (!is_end_token(marker)) {
			return FAIL;
		}
	}

	tds->out_flag = 0x07;

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

static RETCODE _bcp_send_colmetadata(DBPROCESS *dbproc)
{

TDSSOCKET *tds = dbproc->tds_socket;
    int          marker;
unsigned char colmetadata_token = 0x81;
BCP_COLINFO *bcpcol;
char unicode_string[256];


int i;
#ifdef NCBI_FTDS
    int n;
#endif


	if (IS_TDS7_PLUS(tds)) {
		/* Deep joy! - for TDS 8 we have to send a 
           colmetadata message followed by row data 
		 */

		tds_put_byte(tds, colmetadata_token);	/* 0x81 */

#ifdef NCBI_FTDS
        for (i = 0, n= 0; i < dbproc->bcp_colcount; i++) {
            if(dbproc->host_columns[i]->datatype != 0) n++;
        }
        tds_put_smallint(tds, n);
#else
		tds_put_smallint(tds, dbproc->bcp_colcount);
#endif

#ifdef NCBI_FTDS
        for(n= 2; n--;)
#endif
        for (i = 0; i < dbproc->bcp_colcount; i++) {
#ifdef NCBI_FTDS
            if(dbproc->host_columns[i]->datatype == 0) continue;
            if(n) {
                if(is_blob_type(dbproc->bcp_columns[i]->db_type)) continue;
            }
            else
                if(!is_blob_type(dbproc->bcp_columns[i]->db_type)) continue;
#endif

			bcpcol = dbproc->bcp_columns[i];
			tds_put_smallint(tds, bcpcol->db_usertype);
			tds_put_smallint(tds, bcpcol->db_flags);
			tds_put_byte(tds, bcpcol->db_type_save);

			switch (bcpcol->db_varint_size) {
            case 4: tds_put_int(tds, bcpcol->db_length);
				break;
            case 2: tds_put_smallint(tds, bcpcol->db_length);
				break;
            case 1: tds_put_byte(tds, bcpcol->db_length);
				break;
            case 0: break;
			}

			if (is_numeric_type(bcpcol->db_type)) {
				tds_put_byte(tds, bcpcol->db_prec);
				tds_put_byte(tds, bcpcol->db_scale);
			}
#ifndef NCBI_FTDS
            if (IS_TDS80(tds) && is_collate_type(bcpcol->db_type_save)) {
#else
            if (IS_TDS80(tds) && 
                (((bcpcol->db_flags & 0x2) != 0) || is_collate_type(bcpcol->db_type_save))) {
#endif
				tds_put_n(tds, bcpcol->db_collate, 5);
			}
#ifdef NCBI_FTDS
            if(is_blob_type(bcpcol->db_type)) {
                tds_put_byte(tds, 0x0); tds_put_byte(tds, 0x0); /* don't know what it is */
            }
#endif
			tds_put_byte(tds, strlen(bcpcol->db_name));
			tds7_ascii2unicode(tds, bcpcol->db_name, unicode_string, 255);
			tds_put_n(tds, unicode_string, strlen(bcpcol->db_name) * 2);

		}
	}
	return SUCCEED;
}


RETCODE bcp_readfmt(DBPROCESS *dbproc, char *filename)
{
FILE *ffile;
char buffer[1024];

float lf_version = 0.0;
int li_numcols = 0;
int colinfo_count = 0;

    struct fflist {
	struct fflist *nextptr;
	BCP_HOSTCOLINFO colinfo;
};

struct fflist *topptr = (struct fflist *) NULL;
struct fflist *curptr = (struct fflist *) NULL;

BCP_HOSTCOLINFO *hostcol;

	if (dbproc->bcp_direction == 0) {
        _bcp_err_handler(dbproc, BCPEBCPI);
		return FAIL;
	}

	if ((ffile = fopen(filename, "r")) == (FILE *) NULL) {
        _bcp_err_handler(dbproc, BCPEBUOF);
		return (FAIL);
	}

	if ((fgets(buffer, sizeof(buffer), ffile)) != (char *) NULL) {
		buffer[strlen(buffer) - 1] = '\0';	/* discard newline */
		lf_version = atof(buffer);
	}

	if ((fgets(buffer, sizeof(buffer), ffile)) != (char *) NULL) {
		buffer[strlen(buffer) - 1] = '\0';	/* discard newline */
		li_numcols = atoi(buffer);
	}

	while ((fgets(buffer, sizeof(buffer), ffile)) != (char *) NULL) {

		buffer[strlen(buffer) - 1] = '\0';	/* discard newline */


		if (topptr == (struct fflist *) NULL) {	/* first time */
			if ((topptr = (struct fflist *) malloc(sizeof(struct fflist))) == (struct fflist *) NULL) {
				fprintf(stderr, "out of memory\n");
				return (FAIL);
			}
			curptr = topptr;
			curptr->nextptr = NULL;
			if (_bcp_readfmt_colinfo(dbproc, buffer, &(curptr->colinfo)))
				colinfo_count++;
			else
				return (FAIL);
		} else {
			if ((curptr->nextptr = (struct fflist *) malloc(sizeof(struct fflist))) == (struct fflist *) NULL) {
				fprintf(stderr, "out of memory\n");
				return (FAIL);
			}
			curptr = curptr->nextptr;
			curptr->nextptr = NULL;
			if (_bcp_readfmt_colinfo(dbproc, buffer, &(curptr->colinfo)))
				colinfo_count++;
			else
				return (FAIL);
		}

	}
	if (fclose(ffile) != 0) {
        _bcp_err_handler(dbproc, BCPEBUCF);
		return (FAIL);
	}

	if (colinfo_count != li_numcols)
		return (FAIL);