gpAux/extensions/gps3ext/src/gps3ext.cpp
/*
 * Import data into GPDB.
 * invoked by GPDB, be careful with C++ exceptions.
 */
Import data into GPDB.
invoked by GPDB, be careful with C++ exceptions.
Import data into GPDB . invoked by GPDB be careful with C ++ exceptions .
Datum s3_import(PG_FUNCTION_ARGS) {
    if (!CALLED_AS_EXTPROTOCOL(fcinfo))
        elog(ERROR, "extprotocol_import: not called by external protocol manager");
    GPReader *gpreader = (GPReader *)EXTPROTOCOL_GET_USER_CTX(fcinfo);
    if (EXTPROTOCOL_IS_LAST_CALL(fcinfo)) {
        thread_cleanup();
        if (!reader_cleanup(&gpreader)) {
            ereport(ERROR,
                    (0, errmsg("Failed to cleanup S3 extension: %s", s3extErrorMessage.c_str())));
        }
        EXTPROTOCOL_SET_USER_CTX(fcinfo, NULL);
        PG_RETURN_INT32(0);
    }
    if (gpreader == NULL) {
        const char *url_with_options = EXTPROTOCOL_GET_URL(fcinfo);
        thread_setup();
        gpreader = reader_init(url_with_options);
        if (!gpreader) {
            ereport(ERROR, (0, errmsg("Failed to init S3 extension, segid = %d, "
                                      "segnum = %d, please check your "
                                      "configurations and net connection: %s",
                                      s3ext_segid, s3ext_segnum, s3extErrorMessage.c_str())));
        }
        EXTPROTOCOL_SET_USER_CTX(fcinfo, gpreader);
    }
    char *data_buf = EXTPROTOCOL_GET_DATABUF(fcinfo);
    int32 data_len = EXTPROTOCOL_GET_DATALEN(fcinfo);
    if (!reader_transfer_data(gpreader, data_buf, data_len)) {
        ereport(ERROR,
                (0, errmsg("s3_import: could not read data: %s", s3extErrorMessage.c_str())));
    }
    PG_RETURN_INT32(data_len);
}
Datum s3_import ( PG_FUNCTION_ARGS ) { if ( ! CALLED_AS_EXTPROTOCOL ( fcinfo ) ) elog ( ERROR , " " ) ; GPReader * gpreader = ( GPReader * ) EXTPROTOCOL_GET_USER_CTX ( fcinfo ) ; if ( EXTPROTOCOL_IS_LAST_CALL ( fcinfo ) ) { thread_cleanup ( ) ; if ( ! reader_cleanup ( & gpreader ) ) { ereport ( ERROR , ( 0 , errmsg ( " " , s3extErrorMessage . c_str ( ) ) ) ) ; } EXTPROTOCOL_SET_USER_CTX ( fcinfo , NULL ) ; PG_RETURN_INT32 ( 0 ) ; } if ( gpreader == NULL ) { const char * url_with_options = EXTPROTOCOL_GET_URL ( fcinfo ) ; thread_setup ( ) ; gpreader = reader_init ( url_with_options ) ; if ( ! gpreader ) { ereport ( ERROR , ( 0 , errmsg ( " " " " " " , s3ext_segid , s3ext_segnum , s3extErrorMessage . c_str ( ) ) ) ) ; } EXTPROTOCOL_SET_USER_CTX ( fcinfo , gpreader ) ; } char * data_buf = EXTPROTOCOL_GET_DATABUF ( fcinfo ) ; int32 data_len = EXTPROTOCOL_GET_DATALEN ( fcinfo ) ; if ( ! reader_transfer_data ( gpreader , data_buf , data_len ) ) { ereport ( ERROR , ( 0 , errmsg ( " " , s3extErrorMessage . c_str ( ) ) ) ) ; } PG_RETURN_INT32 ( data_len ) ; }
Import data into GPDB.
Import data into GPDB .
/* Must be called via the external table format manager */ /* Get our internal description of the protocol */ /* last call. destroy reader */ /* first call. do any desired init */