| /*! |
| * \file mfvsketch.c |
| |
| \brief CountMin sketch for Most Frequent Value estimation |
| \implementation |
| This is basically a CountMin sketch that keeps track of most frequent values |
| as it goes. This is easy to do, because at any point during a scan, |
| it can use the CM sketch to quickly get the count of any value so far. |
| |
| It only uses CountMin sketches for value counting, and doesn't need the "dyadic" range trick. |
| As a result it's not limited to integers, and the implementation works |
| for any Postgres data type. |
| |
| |
| The parallel method (<c>mfvsketch_quick_histogram</c>) is a heuristic with no |
| such guarantees, but it will likely work well in most cases. As an example |
| of a case where it will fail, consider a scenario where the top <i>n</i> values on node 1 are very infrequent on |
| node 2, and the top <i>n</i> values on node 2 are infrequent on node 1. But the <i>n</i>+1'th value |
| is the same on both nodes and the most frequent value in toto. It will get |
| surpressed incorrectly by the parallel heuristic, but get chosen by the standard method. |
| |
| However, we're probably OK here most of the time. What we're interested in are |
| values whose frequencies are unusually high. For |
| columns with very flat distributions, we likely don't care about the results much. |
| Otherwise, The results of this heuristic will likely be |
| unusually frequent values, if not precisely the *most* frequent values. |
| */ |
| |
| |
| |
| #include <postgres.h> |
| #include <utils/array.h> |
| #include <utils/elog.h> |
| #include <utils/builtins.h> |
| #include <utils/lsyscache.h> |
| #include <utils/numeric.h> |
| #include <utils/typcache.h> |
| #include <utils/datum.h> |
| #include <nodes/execnodes.h> |
| #include <fmgr.h> |
| #include <catalog/pg_type.h> |
| #include <funcapi.h> |
| #include "sketch_support.h" |
| #include "countmin.h" |
| |
| #include <ctype.h> |
| |
| /* check whether the content in the given bytea is safe for mfvtransval */ |
| void check_mfvtransval(bytea *storage) { |
| size_t left_len = VARSIZE(storage); |
| Oid outFuncOid; |
| bool typIsVarLen; |
| |
| mfvtransval *mfv = NULL; |
| |
| if (left_len < VARHDRSZ + sizeof(mfvtransval)) { |
| elog(ERROR, "invalid transition state for mfvsketch"); |
| } |
| mfv = (mfvtransval*)VARDATA(storage); |
| left_len -= VARHDRSZ + sizeof(mfvtransval); |
| |
| if (mfv->next_mfv > mfv->max_mfvs) { |
| elog(ERROR, "invalid transition state for mfvsketch"); |
| } |
| |
| if (mfv->next_offset + VARHDRSZ > VARSIZE(storage)) { |
| elog(ERROR, "invalid transition state for mfvsketch"); |
| } |
| |
| if (InvalidOid == mfv->typOid) { |
| elog(ERROR, "invalid transition state for mfvsketch"); |
| } |
| |
| getTypeOutputInfo(mfv->typOid, &outFuncOid, &typIsVarLen); |
| if (mfv->outFuncOid != outFuncOid |
| || mfv->typLen != get_typlen(mfv->typOid) |
| || mfv->typByVal != get_typbyval(mfv->typOid)) { |
| elog(ERROR, "invalid transition state for mfvsketch"); |
| } |
| |
| if (left_len < sizeof(offsetcnt)*mfv->max_mfvs) { |
| elog(ERROR, "invalid transition state for mfvsketch"); |
| } |
| } |
| |
| PG_FUNCTION_INFO_V1(__mfvsketch_trans); |
| |
| /*! |
| * transition function to maintain a CountMin sketch with |
| * Most-Frequent Values |
| */ |
| Datum __mfvsketch_trans(PG_FUNCTION_ARGS) |
| { |
| bytea * transblob = PG_GETARG_BYTEA_P(0); |
| Datum newdatum = PG_GETARG_DATUM(1); |
| int max_mfvs = PG_GETARG_INT32(2); |
| mfvtransval *transval; |
| uint64 tmpcnt; |
| int i; |
| Datum md5_datum; |
| |
| /* |
| * This function makes destructive updates to its arguments. |
| * Make sure it's being called in an agg context. |
| */ |
| if (!(fcinfo->context && |
| (IsA(fcinfo->context, AggState) |
| #ifdef NOTGP |
| || IsA(fcinfo->context, WindowAggState) |
| #endif |
| ))) |
| elog(ERROR, |
| "destructive pass by reference outside agg"); |
| /* initialize if this is first call */ |
| if (VARSIZE(transblob) <= sizeof(MFV_TRANSVAL_SZ(0))) { |
| Oid typOid = get_fn_expr_argtype(fcinfo->flinfo, 1); |
| transblob = mfv_init_transval(max_mfvs, typOid); |
| } |
| // else { |
| // check_mfvtransval(transblob); |
| // } |
| |
| /* ignore NULL inputs */ |
| if (PG_ARGISNULL(1) || PG_ARGISNULL(2)) |
| PG_RETURN_DATUM(PointerGetDatum(transblob)); |
| |
| transval = (mfvtransval *)VARDATA(transblob); |
| // if (transval->typOid != get_fn_expr_argtype(fcinfo->flinfo, 1)) { |
| // elog(ERROR, "cannot aggregate on elements with different types"); |
| // } |
| |
| /* insert into the countmin sketch */ |
| md5_datum = countmin_trans_c(transval->sketch, |
| newdatum, |
| transval->outFuncOid, |
| transval->typOid); |
| |
| tmpcnt = cmsketch_count_md5_datum(transval->sketch, |
| (bytea *)DatumGetPointer(md5_datum), |
| transval->outFuncOid); |
| i = mfv_find(transblob, newdatum); |
| |
| if (i > -1) { |
| transval->mfvs[i].cnt = tmpcnt; |
| } |
| else { |
| /* try to insert as either a new or replacement entry */ |
| for (i = 0; i < (int)transval->max_mfvs; i++) { |
| if (i == (int)transval->next_mfv) { |
| /* room for new */ |
| transblob = mfv_transval_append(transblob, newdatum); |
| transval = (mfvtransval *)VARDATA(transblob); |
| transval->mfvs[i].cnt = tmpcnt; |
| break; |
| } |
| else if (transval->mfvs[i].cnt < tmpcnt) { |
| /* arg beats this mfv */ |
| transblob = mfv_transval_replace(transblob, newdatum, i); |
| transval = (mfvtransval *)VARDATA(transblob); |
| transval->mfvs[i].cnt = tmpcnt; |
| break; |
| } |
| /* else this is not a frequent value */ |
| } |
| } |
| PG_RETURN_DATUM(PointerGetDatum(transblob)); |
| } |
| |
| /*! |
| * look to see if the mfvsketch currently has <c>val</c> |
| * stored as one of its most-frequent values. |
| * Returns the offset in the <c>mfvs</c> array, or -1 |
| * if not found. |
| * NOTE: a 0 return value means the item <i>was found</i> |
| * at offset 0! |
| * \param blob a bytea holding an mfv transval |
| * \param val the datum to search for |
| */ |
| int mfv_find(bytea *blob, Datum val) |
| { |
| mfvtransval *transval = (mfvtransval *)VARDATA(blob); |
| unsigned i; |
| void *datp; |
| Datum iDat; |
| |
| /* look for existing entry for this value */ |
| for (i = 0; i < transval->next_mfv; i++) { |
| /* if they're the same */ |
| datp = mfv_transval_getval(blob,i); |
| iDat = PointerExtractDatum(datp, transval->typByVal); |
| |
| if (datumIsEqual(iDat, val, transval->typByVal, transval->typLen)) { |
| /* arg is an mfv */ |
| return(i); |
| } |
| } |
| return(-1); |
| } |
| |
| /*! |
| * Initialize an mfv sketch |
| * \param max_mfvs the number of "bins" in the histogram |
| * \param typOid the type ID for the column |
| */ |
| bytea *mfv_init_transval(int max_mfvs, Oid typOid) |
| { |
| int initial_size; |
| bool typIsVarLen; |
| bytea * transblob; |
| mfvtransval *transval; |
| |
| if (max_mfvs <= 0) { |
| elog(ERROR, "Invalid entry for number of MFV values"); |
| } |
| /* |
| * initialize mfvtransval, using palloc0 to zero it out. |
| * if typlen is positive (fixed), size chosen accurately. |
| * Else we'll do a conservative estimate of 16 bytes, and repalloc as needed. |
| */ |
| if ((initial_size = get_typlen(typOid)) > 0) |
| initial_size *= max_mfvs*get_typlen(typOid); |
| else /* guess */ |
| initial_size = max_mfvs*16; |
| |
| transblob = (bytea *)palloc0(MFV_TRANSVAL_SZ(max_mfvs) + initial_size); |
| |
| SET_VARSIZE(transblob, MFV_TRANSVAL_SZ(max_mfvs) + initial_size); |
| transval = (mfvtransval *)VARDATA(transblob); |
| transval->max_mfvs = max_mfvs; |
| transval->next_mfv = 0; |
| transval->next_offset = MFV_TRANSVAL_SZ(max_mfvs)-VARHDRSZ; |
| transval->typOid = typOid; |
| getTypeOutputInfo(transval->typOid, |
| &(transval->outFuncOid), |
| &(typIsVarLen)); |
| transval->typLen = get_typlen(transval->typOid); |
| transval->typByVal = get_typbyval(transval->typOid); |
| if (!transval->outFuncOid) { |
| /* no outFunc for this type! */ |
| elog(ERROR, "no outFunc for type %d", transval->typOid); |
| } |
| return(transblob); |
| } |
| |
| /*! |
| * \param blob a bytea holding an mfv transval |
| * \param i index of the mfv to look up |
| * \returns pointer to the datum associated with the i'th mfv |
| */ |
| void *mfv_transval_getval(bytea *blob, uint32 i) |
| { |
| mfvtransval *tvp = (mfvtransval *)VARDATA(blob); |
| void * retval = (void *)(((char*)tvp) + tvp->mfvs[i].offset); |
| Datum dat = PointerExtractDatum(retval, tvp->typByVal); |
| |
| if (i >= tvp->next_mfv) |
| elog(ERROR, |
| "attempt to get frequent value at illegal index %d in mfv sketch", |
| i); |
| if (tvp->mfvs[i].offset > VARSIZE(blob) - VARHDRSZ |
| || tvp->mfvs[i].offset < MFV_TRANSVAL_SZ(tvp->max_mfvs)-VARHDRSZ) |
| elog(ERROR, "illegal offset %u in mfv sketch", tvp->mfvs[i].offset); |
| /* |
| * call ExtractDatumLen to make sure enough space, this checking is unnecessary, |
| * it is used to prevent gcc from optimizing out the ExtractDatumLen function call. |
| */ |
| if (tvp->mfvs[i].offset |
| + ExtractDatumLen(dat, tvp->typLen, tvp->typByVal, VARSIZE(blob) - VARHDRSZ - tvp->mfvs[i].offset) |
| > VARSIZE(blob) - VARHDRSZ) |
| elog(ERROR, "value overruns size of mfv sketch"); |
| |
| return (retval); |
| } |
| |
| /*! |
| * copy datum <c>dat</c> into the offset of position <c>index</c> of |
| * the mfv sketch stored in <c>transblob</c>. |
| * |
| * <i>Caller beware: this helper return assumes that |
| * <c>dat</c> is small enough to fit in the storage |
| * currently used by the datum at position <c>index</c>.</i> |
| * |
| * \param transblob a bytea holding and mfv transval |
| * \param index the index of the destination for copying |
| * \param dat the datum to be copied into the transval |
| */ |
| void mfv_copy_datum(bytea *transblob, int index, Datum dat) |
| { |
| mfvtransval *transval = (mfvtransval *)VARDATA(transblob); |
| size_t datumLen = ExtractDatumLen(dat, transval->typLen, transval->typByVal, -1); |
| void * curval = (char*)transval + transval->mfvs[index].offset; |
| |
| memmove(curval, (void *)DatumExtractPointer(dat, transval->typByVal), datumLen); |
| } |
| |
| /*! |
| * insert a value at position i of the mfv sketch |
| * |
| * we do not overwrite the previous value at position i. |
| * instead we place the new value at the next_offset. |
| * |
| * <i>Note: we do not currently garbage collection the old value's storage. |
| * This wastes space, with the worst-case scenario being a column with values |
| * of increasing size and frequency!</i> |
| * |
| * \param transblob the transition value packed into a bytea |
| * \param dat the value to be inserted |
| * \param i the position to insert at |
| */ |
| bytea *mfv_transval_insert_at(bytea *transblob, Datum dat, uint32 i) |
| { |
| mfvtransval *transval = (mfvtransval *)VARDATA(transblob); |
| bytea * tmpblob; |
| size_t datumLen = ExtractDatumLen(dat, transval->typLen, transval->typByVal, -1); |
| |
| // In order to compare pass-by-val Datum's where length < 8 bytes in |
| // an arch/platform independent way, each Datum needs to be aligned |
| // on an 8-byte boundary (and be padded with zero's from palloc0) |
| if ((transval->typByVal) && (datumLen < sizeof(Datum))) { |
| datumLen = sizeof(Datum); |
| } |
| |
| if (i > transval->next_mfv) |
| elog( |
| ERROR, |
| "attempt to insert frequent value at illegal index %d in mfv sketch", |
| i); |
| if (MFV_TRANSVAL_CAPACITY(transblob) < datumLen) { |
| /* allocate a copy with room for this, and double the current space for values */ |
| size_t curspace = VARSIZE(transblob) - transval->mfvs[0].offset - |
| VARHDRSZ; |
| tmpblob = (bytea *)palloc0(VARSIZE(transblob) + curspace + datumLen); |
| memmove(tmpblob, transblob, VARSIZE(transblob)); |
| SET_VARSIZE(tmpblob, VARSIZE(transblob) + curspace + datumLen); |
| /* |
| * PG won't let us pfree the old transblob |
| * pfree(transblob); |
| */ |
| transblob = tmpblob; |
| transval = (mfvtransval *)VARDATA(transblob); |
| } |
| transval->mfvs[i].offset = transval->next_offset; |
| mfv_copy_datum(transblob, i, dat); |
| |
| transval->next_offset += datumLen; |
| |
| return(transblob); |
| } |
| |
| /*! |
| * insert a value into the mfvsketch |
| * \param transblob the transition value packed into a bytea |
| * \param dat the value to be inserted |
| */ |
| bytea *mfv_transval_append(bytea *transblob, Datum dat) |
| { |
| mfvtransval *transval = (mfvtransval *)VARDATA(transblob); |
| bytea * retval; |
| |
| if (transval->next_mfv == transval->max_mfvs) { |
| elog(ERROR, "attempt to append to a full mfv sketch"); |
| } |
| retval = mfv_transval_insert_at(transblob, |
| dat, |
| transval->next_mfv); |
| (((mfvtransval *)VARDATA(retval))->next_mfv)++; |
| |
| return(retval); |
| } |
| |
| /*! |
| * replace the value at position i of the mfvsketch with dat |
| * |
| * \param transblob the transition value packed into a bytea |
| * \param dat the value to be inserted |
| * \param i the position to replace |
| */ |
| bytea *mfv_transval_replace(bytea *transblob, Datum dat, int i) |
| { |
| /* |
| * if new value is smaller than old, we overwrite at the old offset. |
| * otherwise we call mfv_transval_insert_at which will take care of |
| * space allocation for the new value |
| */ |
| mfvtransval *transval = (mfvtransval *)VARDATA(transblob); |
| size_t datumLen = ExtractDatumLen(dat, transval->typLen, transval->typByVal, -1); |
| void * tmpp = mfv_transval_getval(transblob,i); |
| Datum oldDat = PointerExtractDatum(tmpp, transval->typByVal); |
| size_t oldLen = ExtractDatumLen(oldDat, transval->typLen, transval->typByVal, -1); |
| |
| if (datumLen <= oldLen) { |
| mfv_copy_datum(transblob, i, dat); |
| return transblob; |
| } |
| else return(mfv_transval_insert_at(transblob, dat, i)); |
| } |
| |
| PG_FUNCTION_INFO_V1(__mfvsketch_final); |
| /*! |
| * scalar function taking an mfv sketch, returning a histogram of |
| * its most frequent values |
| */ |
| Datum __mfvsketch_final(PG_FUNCTION_ARGS) |
| { |
| bytea * transblob = PG_GETARG_BYTEA_P(0); |
| mfvtransval *transval = NULL; |
| ArrayType * retval; |
| uint32 i; |
| int dims[2], lbs[2]; |
| /* Oid typInput, typIOParam; */ |
| Oid outFuncOid; |
| bool typIsVarlena; |
| int16 typlen; |
| bool typbyval; |
| char typalign; |
| char typdelim; |
| Oid typioparam; |
| Oid typiofunc; |
| |
| |
| if (PG_ARGISNULL(0)) PG_RETURN_NULL(); |
| if (VARSIZE(transblob) < MFV_TRANSVAL_SZ(0)) PG_RETURN_NULL(); |
| |
| check_mfvtransval(transblob); |
| transval = (mfvtransval *)VARDATA(transblob); |
| /* |
| * We only declare the variable-length array histo here after some sanity |
| * checking. We risk a stack overflow otherwise. In particular, we need to |
| * make sure that transval->max_mfvs is initialized. It might not be if the |
| * (strict) transition function is never called. (MADLIB-254) |
| */ |
| Datum histo[transval->max_mfvs][2]; |
| |
| qsort(transval->mfvs, transval->next_mfv, sizeof(offsetcnt), cnt_cmp_desc); |
| getTypeOutputInfo(INT8OID, |
| &outFuncOid, |
| &typIsVarlena); |
| |
| for (i = 0; i < transval->next_mfv; i++) { |
| void *tmpp = mfv_transval_getval(transblob,i); |
| Datum curval = PointerExtractDatum(tmpp, transval->typByVal); |
| char *countbuf = |
| OidOutputFunctionCall(outFuncOid, |
| Int64GetDatum(transval->mfvs[i].cnt)); |
| char *valbuf = OidOutputFunctionCall(transval->outFuncOid, curval); |
| |
| histo[i][0] = PointerGetDatum(cstring_to_text(valbuf)); |
| histo[i][1] = PointerGetDatum(cstring_to_text(countbuf)); |
| pfree(countbuf); |
| pfree(valbuf); |
| } |
| |
| /* |
| * Get info about element type |
| */ |
| get_type_io_data(TEXTOID, IOFunc_output, |
| &typlen, &typbyval, |
| &typalign, &typdelim, |
| &typioparam, &typiofunc); |
| |
| dims[0] = i; |
| dims[1] = 2; |
| lbs[0] = lbs[1] = 0; |
| retval = construct_md_array((Datum *)histo, |
| NULL, |
| 2, |
| dims, |
| lbs, |
| TEXTOID, |
| -1, |
| 0, |
| 'i'); |
| PG_RETURN_ARRAYTYPE_P(retval); |
| } |
| |
| |
| |
| /*! |
| * support function to sort by count |
| * \param i an offsetcnt object cast to a (void *) |
| * \param j an offsetcnt object cast to a (void *) |
| */ |
| int cnt_cmp_desc(const void *i, const void *j) |
| { |
| offsetcnt *o = (offsetcnt *)i; |
| offsetcnt *p = (offsetcnt *)j; |
| |
| return (p->cnt - o->cnt); |
| } |
| |
| |
| /*! |
| * Greenplum "prefunc" to combine sketches from multiple machines. |
| * See notes at top of file regarding the heuristic nature of this approach. |
| */ |
| PG_FUNCTION_INFO_V1(__mfvsketch_merge); |
| Datum __mfvsketch_merge(PG_FUNCTION_ARGS) |
| { |
| bytea * transblob1 = (bytea *)PG_GETARG_BYTEA_P(0); |
| bytea * transblob2 = (bytea *)PG_GETARG_BYTEA_P(1); |
| |
| PG_RETURN_DATUM(PointerGetDatum(mfvsketch_merge_c(transblob1, transblob2))); |
| } |
| |
| /*! |
| * implementation of the merge of two mfv sketches. we |
| * first merge the embedded countmin sketches to get the |
| * sums of the counts, and then use those sums to pick the |
| * top values for the resulting histogram. We overwrite |
| * the first argument and return it. |
| * \param transblob1 an mfv transval stored inside a bytea |
| * \param transblob2 another mfv transval in a bytea |
| */ |
| bytea *mfvsketch_merge_c(bytea *transblob1, bytea *transblob2) |
| { |
| mfvtransval *transval1 = (mfvtransval *)VARDATA(transblob1); |
| mfvtransval *transval2 = (mfvtransval *)VARDATA(transblob2); |
| void *newblob; |
| mfvtransval *newval; |
| uint32 i, j, cnt; |
| |
| /* handle uninitialized args */ |
| if (VARSIZE(transblob1) <= sizeof(MFV_TRANSVAL_SZ(0)) |
| && VARSIZE(transblob2) <= sizeof(MFV_TRANSVAL_SZ(0))) |
| return(transblob1); |
| else if (VARSIZE(transblob1) <= sizeof(MFV_TRANSVAL_SZ(0))) { |
| transblob1 = mfv_init_transval(transval2->max_mfvs, transval2->typOid); |
| transval1 = (mfvtransval *)VARDATA(transblob1); |
| } |
| else if (VARSIZE(transblob2) <= sizeof(MFV_TRANSVAL_SZ(0))) { |
| transblob2 = mfv_init_transval(transval1->max_mfvs, transval1->typOid); |
| transval2 = (mfvtransval *)VARDATA(transblob2); |
| } |
| check_mfvtransval(transblob1); |
| check_mfvtransval(transblob2); |
| |
| if ( transval1->typOid != transval2->typOid ) { |
| elog(ERROR, "cannot merge two transition state with different element type"); |
| } |
| |
| /* initialize output */ |
| newblob = mfv_init_transval(transval1->max_mfvs, transval1->typOid); |
| newval = (mfvtransval *)VARDATA(newblob); |
| |
| /* combine sketches */ |
| for (i = 0; i < DEPTH; i++) |
| for (j = 0; j < NUMCOUNTERS; j++) |
| newval->sketch[i][j] = transval1->sketch[i][j] |
| + transval2->sketch[i][j]; |
| |
| /* recompute the counts using the merged sketch */ |
| for (i = 0; i < transval1->next_mfv; i++) { |
| void *tmpp = mfv_transval_getval(transblob1,i); |
| Datum dat = PointerExtractDatum(tmpp, transval1->typByVal); |
| |
| transval1->mfvs[i].cnt = cmsketch_count_c(newval->sketch, |
| dat, |
| newval->outFuncOid, |
| newval->typOid); |
| } |
| for (i = 0; i < transval2->next_mfv; i++) { |
| void *tmpp = mfv_transval_getval(transblob2,i); |
| Datum dat = PointerExtractDatum(tmpp, transval2->typByVal); |
| |
| transval2->mfvs[i].cnt = cmsketch_count_c(newval->sketch, |
| dat, |
| newval->outFuncOid, |
| newval->typOid); |
| } |
| |
| /* now take maxes on mfvs in a sort-merge style, copying into transval1 */ |
| qsort(transval1->mfvs, transval1->next_mfv, sizeof(offsetcnt), cnt_cmp_desc); |
| qsort(transval2->mfvs, transval2->next_mfv, sizeof(offsetcnt), cnt_cmp_desc); |
| |
| Datum *values_seen = palloc0((2 * newval->max_mfvs) * sizeof(Datum)); |
| unsigned int k; |
| |
| /* choose top k from transval1 and transval2 */ |
| for (i = j = cnt = 0; |
| cnt < newval->max_mfvs |
| && (j < transval2->next_mfv || i < transval1->next_mfv);) { |
| Datum iDatum, jDatum; |
| |
| if (i < transval1->next_mfv && |
| (j == transval2->next_mfv |
| || transval1->mfvs[i].cnt >= transval2->mfvs[j].cnt)) { |
| /* next item comes from transval1 */ |
| iDatum = PointerExtractDatum(mfv_transval_getval(transblob1, i), |
| transval1->typByVal); |
| for (k = 0; k < cnt; k++) { |
| if (datumIsEqual(iDatum, values_seen[k], transval1->typByVal, transval1->typLen)) { |
| break; |
| } |
| } |
| |
| if (k == cnt) { |
| newblob = mfv_transval_append(newblob, iDatum); |
| newval = (mfvtransval *)VARDATA(newblob); |
| newval->mfvs[cnt].cnt = transval1->mfvs[i].cnt; |
| if (cnt < 2 * transval1->max_mfvs) { |
| values_seen[cnt] = datumCopy(iDatum, transval1->typByVal, |
| transval1->typLen); |
| cnt++; |
| } else { |
| elog(WARNING, "Merge received more than %d values from transition functions", 2*transval1->max_mfvs); |
| } |
| } |
| i++; |
| } |
| else if (j < transval2->next_mfv && |
| (i == transval1->next_mfv |
| || transval1->mfvs[i].cnt < transval2->mfvs[j].cnt)) { |
| /* next item comes from transval2 */ |
| jDatum = PointerExtractDatum(mfv_transval_getval(transblob2, j), |
| transval2->typByVal); |
| |
| for (k = 0; k < cnt; k++) { |
| if (datumIsEqual(jDatum, values_seen[k], transval2->typByVal, |
| transval2->typLen)) { |
| break; |
| } |
| } |
| |
| if (k == cnt) { |
| newblob = mfv_transval_append(newblob, jDatum); |
| newval = (mfvtransval *)VARDATA(newblob); |
| newval->mfvs[cnt].cnt = transval2->mfvs[j].cnt; |
| if (cnt < 2*transval1->max_mfvs) { |
| values_seen[cnt] = datumCopy(jDatum, transval2->typByVal, transval2->typLen); |
| cnt++; |
| } else { |
| elog(WARNING, "Merge received more than %d values from transition functions", 2*transval2->max_mfvs); |
| } |
| } |
| j++; |
| } |
| } |
| |
| if (!transval1->typByVal){ |
| for (k = 0; k < cnt; k++) { |
| Pointer s = DatumGetPointer(values_seen[k]); |
| pfree(s); |
| } |
| } |
| pfree(values_seen); |
| |
| return(newblob); |
| } |
