public inbox for [email protected]help / color / mirror / Atom feed
[PATCH 7/9] Remove the special batch mode, use a larger buffer always 3+ messages / 3 participants [nested] [flat]
* [PATCH 7/9] Remove the special batch mode, use a larger buffer always @ 2021-02-02 00:57 Tomas Vondra <[email protected]> 0 siblings, 0 replies; 3+ messages in thread From: Tomas Vondra @ 2021-02-02 00:57 UTC (permalink / raw) Instead of using a batch mode (with a larger input buffer) only for new ranges, which introduces "special cases" in various places, use it as the standard approach. Also, instead of sizing the buffer to cover the whole range, limit it to some reasonable limit (10x the user-specified size). That should give us most of the benefits without consuming a lot of memory. --- src/backend/access/brin/brin_minmax_multi.c | 851 ++++++++++++-------- 1 file changed, 525 insertions(+), 326 deletions(-) diff --git a/src/backend/access/brin/brin_minmax_multi.c b/src/backend/access/brin/brin_minmax_multi.c index 69a72da337..08d0d55b06 100644 --- a/src/backend/access/brin/brin_minmax_multi.c +++ b/src/backend/access/brin/brin_minmax_multi.c @@ -92,7 +92,15 @@ */ #define PROCNUM_BASE 11 -#define MINMAX_LOAD_FACTOR 0.75 +/* + * Sizing the insert buffer - we use 10x the number of values specified + * in the reloption, but we cap it to 8192 not to get too large. When + * the buffer gets full, we reduce the number of values by half. + */ +#define MINMAX_BUFFER_FACTOR 10 +#define MINMAX_BUFFER_MIN 256 +#define MINMAX_BUFFER_MAX 8192 +#define MINMAX_BUFFER_LOAD_FACTOR 0.5 typedef struct MinmaxMultiOpaque { @@ -155,23 +163,24 @@ typedef struct Ranges Oid typid; Oid colloid; AttrNumber attno; + FmgrInfo *cmp; /* (2*nranges + nvalues) <= maxvalues */ int nranges; /* number of ranges in the array (stored) */ + int nsorted; /* number of sorted values (ranges + points) */ int nvalues; /* number of values in the data array (all) */ int maxvalues; /* maximum number of values (reloption) */ /* - * In batch mode, we simply add the values into a buffer, without any - * expensive steps (sorting, deduplication, ...). The buffer is sized - * to be larger than the target number of values per range, which - * reduces the number of compactions - operating on larger buffers is - * significantly more efficient, in most cases. We keep the actual - * target and compact to the requested number of values at the very - * end, before serializing to on-disk representation. + * We simply add the values into a large buffer, without any expensive + * steps (sorting, deduplication, ...). The buffer is a multiple of + * the target number of values, so the compaction happen less often, + * amortizing the costs. We keep the actual target and compact to + * the requested number of values at the very end, before serializing + * to on-disk representation. */ - bool batch_mode; - int target_maxvalues; /* requested number of values */ + /* requested number of values */ + int target_maxvalues; /* values stored for this range - either raw values, or ranges */ Datum values[FLEXIBLE_ARRAY_MEMBER]; @@ -203,7 +212,7 @@ typedef struct SerializedRanges static SerializedRanges *range_serialize(Ranges *range); -static Ranges *range_deserialize(SerializedRanges *range); +static Ranges *range_deserialize(int maxvalues, SerializedRanges *range); /* Cache for support and strategy procesures. */ @@ -213,6 +222,14 @@ static FmgrInfo *minmax_multi_get_procinfo(BrinDesc *bdesc, uint16 attno, static FmgrInfo *minmax_multi_get_strategy_procinfo(BrinDesc *bdesc, uint16 attno, Oid subtype, uint16 strategynum); +typedef struct compare_context +{ + FmgrInfo *cmpFn; + Oid colloid; +} compare_context; + +static int compare_values(const void *a, const void *b, void *arg); + /* * minmax_multi_init @@ -240,6 +257,57 @@ minmax_multi_init(int maxvalues) return ranges; } +static void +AssertCheckRanges(Ranges *ranges, FmgrInfo *cmpFn, Oid colloid); + + +static void +range_deduplicate_values(Ranges *range) +{ + int i, n; + int start; + compare_context cxt; + + /* + * If there are no unsorted values, we're done (this probably can't + * happen, as we're adding values to unsorted part). + */ + if (range->nsorted == range->nvalues) + return; + + /* sort the values */ + cxt.colloid = range->colloid; + cxt.cmpFn = range->cmp; + + /* how many values to sort? */ + start = 2 * range->nranges; + + qsort_arg(&range->values[start], + range->nvalues, sizeof(Datum), + compare_values, (void *) &cxt); + + n = 1; + for (i = 1; i < range->nvalues; i++) + { + /* same as preceding value, so store it */ + if (compare_values(&range->values[start + i - 1], + &range->values[start + i], + (void *) &cxt) == 0) + continue; + + range->values[start + n] = range->values[start + i]; + + n++; + } + + /* now all the values are sorted */ + range->nvalues = n; + range->nsorted = n; + + AssertCheckRanges(range, range->cmp, range->colloid); +} + + /* * range_serialize * Serialize the in-memory representation into a compact varlena value. @@ -262,14 +330,25 @@ range_serialize(Ranges *range) /* simple sanity checks */ Assert(range->nranges >= 0); + Assert(range->nsorted >= 0); Assert(range->nvalues >= 0); Assert(range->maxvalues > 0); + Assert(range->target_maxvalues > 0); + + /* at this point the range should be compacted to the target size */ + Assert(2*range->nranges + range->nvalues <= range->target_maxvalues); + + Assert(range->target_maxvalues <= range->maxvalues); + + /* range boundaries are always sorted */ + Assert(range->nvalues >= range->nsorted); + + /* sort and deduplicate values, if there's unsorted part */ + range_deduplicate_values(range); /* see how many Datum values we actually have */ nvalues = 2*range->nranges + range->nvalues; - Assert(2*range->nranges + range->nvalues <= range->maxvalues); - typid = range->typid; typbyval = get_typbyval(typid); typlen = get_typlen(typid); @@ -316,7 +395,7 @@ range_serialize(Ranges *range) serialized->typid = typid; serialized->nranges = range->nranges; serialized->nvalues = range->nvalues; - serialized->maxvalues = range->maxvalues; + serialized->maxvalues = range->target_maxvalues; /* * And now copy also the boundary values (like the length calculation @@ -367,7 +446,7 @@ range_serialize(Ranges *range) * in the in-memory value array. */ static Ranges * -range_deserialize(SerializedRanges *serialized) +range_deserialize(int maxvalues, SerializedRanges *serialized) { int i, nvalues; @@ -384,15 +463,18 @@ range_deserialize(SerializedRanges *serialized) nvalues = 2*serialized->nranges + serialized->nvalues; Assert(nvalues <= serialized->maxvalues); + Assert(serialized->maxvalues <= maxvalues); - range = minmax_multi_init(serialized->maxvalues); + range = minmax_multi_init(maxvalues); /* copy the header info */ range->nranges = serialized->nranges; range->nvalues = serialized->nvalues; - range->maxvalues = serialized->maxvalues; + range->nsorted = serialized->nvalues; + range->maxvalues = maxvalues; + range->target_maxvalues = serialized->maxvalues; + range->typid = serialized->typid; - range->batch_mode = false; typbyval = get_typbyval(serialized->typid); typlen = get_typlen(serialized->typid); @@ -439,12 +521,6 @@ range_deserialize(SerializedRanges *serialized) return range; } -typedef struct compare_context -{ - FmgrInfo *cmpFn; - Oid colloid; -} compare_context; - /* * Used to represent ranges expanded during merging and combining (to * reduce number of boundary values to store). @@ -528,6 +604,115 @@ compare_values(const void *a, const void *b, void *arg) return 0; } +void *bsearch_arg(const void *key, const void *base, + size_t nmemb, size_t size, + int (*compar) (const void *, const void *, void *), + void *arg); + +static bool +has_matching_range(BrinDesc *bdesc, Oid colloid, Ranges *ranges, + Datum newval, AttrNumber attno, Oid typid) +{ + Datum compar; + + Datum minvalue = ranges->values[0]; + Datum maxvalue = ranges->values[2*ranges->nranges - 1]; + + FmgrInfo *cmpLessFn; + FmgrInfo *cmpGreaterFn; + + /* binary search on ranges */ + int start, + end; + + if (ranges->nranges == 0) + return false; + + /* + * Otherwise, need to compare the new value with boundaries of all + * the ranges. First check if it's less than the absolute minimum, + * which is the first value in the array. + */ + cmpLessFn = minmax_multi_get_strategy_procinfo(bdesc, attno, typid, + BTLessStrategyNumber); + compar = FunctionCall2Coll(cmpLessFn, colloid, newval, minvalue); + + /* smaller than the smallest value in the range list */ + if (DatumGetBool(compar)) + return false; + + /* + * And now compare it to the existing maximum (last value in the + * data array). But only if we haven't already ruled out a possible + * match in the minvalue check. + */ + cmpGreaterFn = minmax_multi_get_strategy_procinfo(bdesc, attno, typid, + BTGreaterStrategyNumber); + compar = FunctionCall2Coll(cmpGreaterFn, colloid, newval, maxvalue); + + if (DatumGetBool(compar)) + return false; + + /* + * So we know it's in the general min/max, the question is whether it + * falls in one of the ranges or gaps. We'll use a binary search on + * the ranges. + * + * it's in the general range, but is it actually covered by any + * of the ranges? Repeat the check for each range. + * + * XXX We simply walk the ranges sequentially, but maybe we could + * further leverage the ordering and non-overlap and use bsearch to + * speed this up a bit. + */ + start = 0; /* first range */ + end = ranges->nranges - 1; /* last range */ + while (true) + { + int midpoint = (start + end) / 2; + + /* this means we ran out of ranges in the last step */ + if (start > end) + return false; + + /* copy the min/max values from the ranges */ + minvalue = ranges->values[2 * midpoint]; + maxvalue = ranges->values[2 * midpoint + 1]; + + /* + * Is the value smaller than the minval? If yes, we'll recurse + * to the left side of range array. + */ + compar = FunctionCall2Coll(cmpLessFn, colloid, newval, minvalue); + + /* smaller than the smallest value in this range */ + if (DatumGetBool(compar)) + { + end = (midpoint - 1); + continue; + } + + /* + * Is the value greater than the minval? If yes, we'll recurse + * to the right side of range array. + */ + compar = FunctionCall2Coll(cmpGreaterFn, colloid, newval, maxvalue); + + /* larger than the largest value in this range */ + if (DatumGetBool(compar)) + { + start = (midpoint + 1); + continue; + } + + /* hey, we found a matching range */ + return true; + } + + return false; +} + + /* * range_contains_value * See if the new value is already contained in the range list. @@ -552,8 +737,6 @@ range_contains_value(BrinDesc *bdesc, Oid colloid, Ranges *ranges, Datum newval) { int i; - FmgrInfo *cmpLessFn; - FmgrInfo *cmpGreaterFn; FmgrInfo *cmpEqualFn; Oid typid = attr->atttypid; @@ -562,77 +745,8 @@ range_contains_value(BrinDesc *bdesc, Oid colloid, * range, and only when there's still a chance of getting a match we * inspect the individual ranges. */ - if (ranges->nranges > 0) - { - Datum compar; - bool match = true; - - Datum minvalue = ranges->values[0]; - Datum maxvalue = ranges->values[2*ranges->nranges - 1]; - - /* - * Otherwise, need to compare the new value with boundaries of all - * the ranges. First check if it's less than the absolute minimum, - * which is the first value in the array. - */ - cmpLessFn = minmax_multi_get_strategy_procinfo(bdesc, attno, typid, - BTLessStrategyNumber); - compar = FunctionCall2Coll(cmpLessFn, colloid, newval, minvalue); - - /* smaller than the smallest value in the range list */ - if (DatumGetBool(compar)) - match = false; - - /* - * And now compare it to the existing maximum (last value in the - * data array). But only if we haven't already ruled out a possible - * match in the minvalue check. - */ - if (match) - { - cmpGreaterFn = minmax_multi_get_strategy_procinfo(bdesc, attno, typid, - BTGreaterStrategyNumber); - compar = FunctionCall2Coll(cmpGreaterFn, colloid, newval, maxvalue); - - if (DatumGetBool(compar)) - match = false; - } - - /* - * So it's in the general range, but is it actually covered by any - * of the ranges? Repeat the check for each range. - * - * XXX We simply walk the ranges sequentially, but maybe we could - * further leverage the ordering and non-overlap and use bsearch to - * speed this up a bit. - */ - for (i = 0; i < ranges->nranges && match; i++) - { - /* copy the min/max values from the ranges */ - minvalue = ranges->values[2*i]; - maxvalue = ranges->values[2*i+1]; - - /* - * Otherwise, need to compare the new value with boundaries of all - * the ranges. First check if it's less than the absolute minimum, - * which is the first value in the array. - */ - compar = FunctionCall2Coll(cmpLessFn, colloid, newval, minvalue); - - /* smaller than the smallest value in this range */ - if (DatumGetBool(compar)) - continue; - - compar = FunctionCall2Coll(cmpGreaterFn, colloid, newval, maxvalue); - - /* larger than the largest value in this range */ - if (DatumGetBool(compar)) - continue; - - /* hey, we found a matching row */ - return true; - } - } + if (has_matching_range(bdesc, colloid, ranges, newval, attno, typid)) + return true; cmpEqualFn = minmax_multi_get_strategy_procinfo(bdesc, attno, typid, BTEqualStrategyNumber); @@ -640,92 +754,42 @@ range_contains_value(BrinDesc *bdesc, Oid colloid, /* * We're done with the ranges, now let's inspect the exact values. * - * XXX Again, we do sequentially search the values - consider leveraging - * the ordering of values to improve performance. + * XXX We do sequential search for small number of values, and bsearch + * once we have more than 16 values. + * + * XXX We only inspect the sorted part - that's OK. For building it may + * produce false negatives, but only after we already added some values + * to the unsorted part, so we've modified the value. And when querying + * the index, there should be no unsorted values. */ - for (i = 2*ranges->nranges; i < 2*ranges->nranges + ranges->nvalues; i++) + if (ranges->nsorted >= 16) { - Datum compar; + compare_context cxt; - compar = FunctionCall2Coll(cmpEqualFn, colloid, newval, ranges->values[i]); + cxt.colloid = ranges->colloid; + cxt.cmpFn = ranges->cmp; - /* found an exact match */ - if (DatumGetBool(compar)) + if (bsearch_arg(&newval, &ranges->values[2*ranges->nranges], + ranges->nsorted, sizeof(Datum), + compare_values, (void *) &cxt) != NULL) return true; } - - /* the value is not covered by this BRIN tuple */ - return false; -} - -/* - * insert_value - * Adds a new value into the single-point part, while maintaining ordering. - * - * The function inserts the new value to the right place in the single-point - * part of the range. It assumes there's enough free space, and then does - * essentially an insert-sort. - * - * XXX Assumes the 'values' array has space for (nvalues+1) entries, and that - * only the first nvalues are used. - */ -static void -insert_value(FmgrInfo *cmp, Oid colloid, Datum *values, int nvalues, - Datum newvalue) -{ - int i; - Datum lt; - - /* If there are no values yet, store the new one and we're done. */ - if (!nvalues) + else { - values[0] = newvalue; - return; - } - - /* - * A common case is that the new value is entirely out of the existing - * range, i.e. it's either smaller or larger than all previous values. - * So we check and handle this case first - first we check the larger - * case, because in that case we can just append the value to the end - * of the array and we're done. - */ + for (i = 2*ranges->nranges; i < 2*ranges->nranges + ranges->nsorted; i++) + { + Datum compar; - /* Is it greater than all existing values in the array? */ - lt = FunctionCall2Coll(cmp, colloid, values[nvalues-1], newvalue); - if (DatumGetBool(lt)) - { - /* just copy it in-place and we're done */ - values[nvalues] = newvalue; - return; - } + compar = FunctionCall2Coll(cmpEqualFn, colloid, newval, ranges->values[i]); - /* - * OK, I lied a bit - we won't check the smaller case explicitly, but - * we'll just compare the value to all existing values in the array. - * But we happen to start with the smallest value, so we're actually - * doing the check anyway. - * - * XXX We do walk the values sequentially. Perhaps we could/should be - * smarter and do some sort of bisection, to improve performance? - */ - for (i = 0; i < nvalues; i++) - { - lt = FunctionCall2Coll(cmp, colloid, newvalue, values[i]); - if (DatumGetBool(lt)) - { - /* - * Move values to make space for the new entry, which should go - * to index 'i'. Entries 0 ... (i-1) should stay where they are. - */ - memmove(&values[i+1], &values[i], (nvalues-i) * sizeof(Datum)); - values[i] = newvalue; - return; + /* found an exact match */ + if (DatumGetBool(compar)) + return true; } } - /* We should never really get here. */ - Assert(false); + /* the value is not covered by this BRIN tuple */ + return false; } #ifdef USE_ASSERT_CHECKING @@ -754,11 +818,12 @@ static void AssertCheckRanges(Ranges *ranges, FmgrInfo *cmpFn, Oid colloid) { #ifdef USE_ASSERT_CHECKING - int i, j; + int i; /* some basic sanity checks */ Assert(ranges->nranges >= 0); - Assert(ranges->nvalues >= 0); + Assert(ranges->nsorted >= 0); + Assert(ranges->nvalues >= ranges->nsorted); Assert(ranges->maxvalues >= 2 * ranges->nranges + ranges->nvalues); Assert(ranges->typid != InvalidOid); @@ -770,32 +835,111 @@ AssertCheckRanges(Ranges *ranges, FmgrInfo *cmpFn, Oid colloid) */ AssertArrayOrder(cmpFn, colloid, ranges->values, 2*ranges->nranges); - /* finally check that none of the values are not covered by ranges */ + /* then the single-point ranges (with nvalues boundar values ) */ + AssertArrayOrder(cmpFn, colloid, &ranges->values[2*ranges->nranges], + ranges->nsorted); + + /* + * Check that none of the values are not covered by ranges (both + * sorted and unsorted) + */ for (i = 0; i < ranges->nvalues; i++) { + Datum compar; + int start, + end; + Datum minvalue, + maxvalue; + Datum value = ranges->values[2 * ranges->nranges + i]; - for (j = 0; j < ranges->nranges; j++) + if (ranges->nranges == 0) + break; + + minvalue = ranges->values[0]; + maxvalue = ranges->values[2*ranges->nranges - 1]; + + /* + * Is the value smaller than the minval? If yes, we'll recurse + * to the left side of range array. + */ + compar = FunctionCall2Coll(cmpFn, colloid, value, minvalue); + + /* smaller than the smallest value in the first range */ + if (DatumGetBool(compar)) + continue; + + /* + * Is the value greater than the minval? If yes, we'll recurse + * to the right side of range array. + */ + compar = FunctionCall2Coll(cmpFn, colloid, maxvalue, value); + + /* larger than the largest value in the last range */ + if (DatumGetBool(compar)) + continue; + + start = 0; /* first range */ + end = ranges->nranges - 1; /* last range */ + while (true) { - Datum r; + int midpoint = (start + end) / 2; + + /* this means we ran out of ranges in the last step */ + if (start > end) + break; + + /* copy the min/max values from the ranges */ + minvalue = ranges->values[2 * midpoint]; + maxvalue = ranges->values[2 * midpoint + 1]; - Datum minval = ranges->values[2 * j]; - Datum maxval = ranges->values[2 * j + 1]; + /* + * Is the value smaller than the minval? If yes, we'll recurse + * to the left side of range array. + */ + compar = FunctionCall2Coll(cmpFn, colloid, value, minvalue); - /* if value is smaller than range minimum, that's OK */ - r = FunctionCall2Coll(cmpFn, colloid, value, minval); - if (DatumGetBool(r)) + /* smaller than the smallest value in this range */ + if (DatumGetBool(compar)) + { + end = (midpoint - 1); continue; + } + + /* + * Is the value greater than the minval? If yes, we'll recurse + * to the right side of range array. + */ + compar = FunctionCall2Coll(cmpFn, colloid, maxvalue, value); - /* if value is greater than range maximum, that's OK */ - r = FunctionCall2Coll(cmpFn, colloid, maxval, value); - if (DatumGetBool(r)) + /* larger than the largest value in this range */ + if (DatumGetBool(compar)) + { + start = (midpoint + 1); continue; + } - /* value is between [min,max], which is wrong */ + /* hey, we found a matching range */ Assert(false); } } + + /* and values in the unsorted part must not be in sorted part */ + for (i = ranges->nsorted; i < ranges->nvalues; i++) + { + compare_context cxt; + Datum value = ranges->values[2 * ranges->nranges + i]; + + if (ranges->nsorted == 0) + break; + + cxt.colloid = ranges->colloid; + cxt.cmpFn = ranges->cmp; + + Assert(bsearch_arg(&value, &ranges->values[2*ranges->nranges], + ranges->nsorted, sizeof(Datum), + compare_values, (void *) &cxt) == NULL); + } #endif } @@ -1106,8 +1250,7 @@ build_distances(FmgrInfo *distanceFn, Oid colloid, */ static CombineRange * build_combine_ranges(FmgrInfo *cmp, Oid colloid, Ranges *ranges, - bool addvalue, Datum newvalue, int *nranges, - bool deduplicate) + int *nranges) { int ncranges; CombineRange *cranges; @@ -1115,28 +1258,15 @@ build_combine_ranges(FmgrInfo *cmp, Oid colloid, Ranges *ranges, /* now do the actual merge sort */ ncranges = ranges->nranges + ranges->nvalues; - /* should we add an extra value? */ - if (addvalue) - ncranges += 1; - cranges = (CombineRange *) palloc0(ncranges * sizeof(CombineRange)); - /* put the new value at the beginning */ - if (addvalue) - { - cranges[0].minval = newvalue; - cranges[0].maxval = newvalue; - cranges[0].collapsed = true; - - /* then the regular and collapsed ranges */ - fill_combine_ranges(&cranges[1], ncranges-1, ranges); - } - else - fill_combine_ranges(cranges, ncranges, ranges); + /* fll the combine ranges */ + fill_combine_ranges(cranges, ncranges, ranges); /* and sort the ranges */ - ncranges = sort_combine_ranges(cmp, colloid, cranges, ncranges, - deduplicate); + ncranges = sort_combine_ranges(cmp, colloid, + cranges, ncranges, + true); /* deduplicate */ /* remember how many cranges we built */ *nranges = ncranges; @@ -1321,19 +1451,28 @@ store_combine_ranges(Ranges *ranges, CombineRange *cranges, int ncranges) } } + /* all the values are sorted */ + ranges->nsorted = ranges->nvalues; + Assert(count_values(cranges, ncranges) == 2*ranges->nranges + ranges->nvalues); Assert(2*ranges->nranges + ranges->nvalues <= ranges->maxvalues); } + + /* - * range_add_value - * Add the new value to the multi-minmax range. + * Consider freeing space in the ranges. + * + * Returns true if the value was actually modified. */ static bool -range_add_value(BrinDesc *bdesc, Oid colloid, - AttrNumber attno, Form_pg_attribute attr, - Ranges *ranges, Datum newval) +ensure_free_space_in_buffer(BrinDesc *bdesc, Oid colloid, + AttrNumber attno, Form_pg_attribute attr, + Ranges *range) { + MemoryContext ctx; + MemoryContext oldctx; + FmgrInfo *cmpFn, *distanceFn; @@ -1342,109 +1481,44 @@ range_add_value(BrinDesc *bdesc, Oid colloid, int ncranges; DistanceValue *distances; - MemoryContext ctx; - MemoryContext oldctx; - - /* we'll certainly need the comparator, so just look it up now */ - cmpFn = minmax_multi_get_strategy_procinfo(bdesc, attno, attr->atttypid, - BTLessStrategyNumber); - - /* comprehensive checks of the input ranges */ - AssertCheckRanges(ranges, cmpFn, colloid); - - Assert((ranges->nranges >= 0) && (ranges->nvalues >= 0) && (ranges->maxvalues >= 0)); - /* - * When batch-building, there should be no ranges. So either the - * number of ranges is 0 or we're not in batching mode. + * If there is free space in the buffer, we're done without having + * to modify anything. */ - Assert(!ranges->batch_mode || (ranges->nranges == 0)); - - /* When batch-building, just add it and we're done. */ - if (ranges->batch_mode) - { - /* there has to be free space, if we've sized the struct */ - Assert(ranges->nvalues < ranges->maxvalues); - - /* Make a copy of the value, if needed. */ - ranges->values[ranges->nvalues++] - = datumCopy(newval, attr->attbyval, attr->attlen);; - - return true; - } - - /* - * Bail out if the value already is covered by the range. - * - * We could also add values until we hit values_per_range, and then - * do the deduplication in a batch, hoping for better efficiency. But - * that would mean we actually modify the range every time, which means - * having to serialize the value, which does palloc, walks the values, - * copies them, etc. Not exactly cheap. - * - * So instead we do the check, which should be fairly cheap - assuming - * the comparator function is not very expensive. - * - * This also implies means the values array can't contain duplicities. - */ - if (range_contains_value(bdesc, colloid, attno, attr, ranges, newval)) + if (2*range->nranges + range->nvalues < range->maxvalues) return false; - /* Make a copy of the value, if needed. */ - newval = datumCopy(newval, attr->attbyval, attr->attlen); - - /* - * If there's space in the values array, copy it in and we're done. - * - * We do want to keep the values sorted (to speed up searches), so we - * do a simple insertion sort. We could do something more elaborate, - * e.g. by sorting the values only now and then, but for small counts - * (e.g. when maxvalues is 64) this should be fine. - */ - if (2*ranges->nranges + ranges->nvalues < ranges->maxvalues) - { - Datum *values; - - /* beginning of the 'single value' part (for convenience) */ - values = &ranges->values[2*ranges->nranges]; - - insert_value(cmpFn, colloid, values, ranges->nvalues, newval); - - ranges->nvalues++; - - /* - * Check we haven't broken the ordering of boundary values (checks - * both parts, but that doesn't hurt). - */ - AssertCheckRanges(ranges, cmpFn, colloid); + /* we'll certainly need the comparator, so just look it up now */ + cmpFn = minmax_multi_get_strategy_procinfo(bdesc, attno, attr->atttypid, + BTLessStrategyNumber); - /* Also check the range contains the value we just added. */ - // FIXME Assert(ranges, cmpFn, colloid); + /* Try deduplicating values in the unsorted part */ + range_deduplicate_values(range); - /* yep, we've modified the range */ + /* did we reduce enough free space by just the deduplication? */ + if (2*range->nranges + range->nvalues <= range->maxvalues * MINMAX_BUFFER_LOAD_FACTOR) return true; - } /* - * Damn - the new value is not in the range yet, but we don't have space - * to just insert it. So we need to combine some of the existing ranges, - * to reduce the number of values we need to store (joining two intervals - * reduces the number of boundaries to store by 2). + * we need to combine some of the existing ranges, to reduce the number + * of values we need to store (joining intervals reduces the number of + * boundary values). * - * To do that we first construct an array of CombineRange items - each - * combine range tracks if it's a regular range or collapsed range, where - * "collapsed" means "single point." + * We first construct an array of CombineRange items - each combine range + * tracks if it's a regular range or a collapsed range, where "collapsed" + * means "single point." This makes the processing easier, as it allows + * handling ranges and points the same way. * - * Existing ranges (we have ranges->nranges of them) map to combine ranges - * directly, while single points (ranges->nvalues of them) have to be - * expanded. We neet the combine ranges to be sorted, and we do that by - * performing a merge sort of ranges, values and new value. + * Then we sort the combine ranges - this is necessary, because although + * ranges and points were sorted on their own, the new array is not. We + * do that by performing a merge sort of the two parts. * * The distanceFn calls (which may internally call e.g. numeric_le) may - * allocate quite a bit of memory, and we must not leak it. Otherwise - * we'd have problems e.g. when building indexes. So we create a local - * memory context and make sure we free the memory before leaving this - * function (not after every call). + * allocate quite a bit of memory, and we must not leak it (we might have + * to do this repeatedly, even for a single BRIN page range). Otherwise + * we'd have problems e.g. when building new indexes. So we use a memory + * context and make sure we free the memory at the end (so if we call + * the distance function many times, it might be an issue, but meh). */ ctx = AllocSetContextCreate(CurrentMemoryContext, "minmax-multi context", @@ -1453,9 +1527,7 @@ range_add_value(BrinDesc *bdesc, Oid colloid, oldctx = MemoryContextSwitchTo(ctx); /* OK build the combine ranges */ - cranges = build_combine_ranges(cmpFn, colloid, ranges, - true, newval, &ncranges, - false); + cranges = build_combine_ranges(cmpFn, colloid, range, &ncranges); /* and we'll also need the 'distance' procedure */ distanceFn = minmax_multi_get_procinfo(bdesc, attno, PROCNUM_DISTANCE); @@ -1469,21 +1541,104 @@ range_add_value(BrinDesc *bdesc, Oid colloid, * use too low or high value. */ ncranges = reduce_combine_ranges(cranges, ncranges, distances, - ranges->maxvalues * MINMAX_LOAD_FACTOR, + range->maxvalues * MINMAX_BUFFER_LOAD_FACTOR, cmpFn, colloid); - Assert(count_values(cranges, ncranges) <= ranges->maxvalues * MINMAX_LOAD_FACTOR); + Assert(count_values(cranges, ncranges) <= range->maxvalues * MINMAX_BUFFER_LOAD_FACTOR); /* decompose the combine ranges into regular ranges and single values */ - store_combine_ranges(ranges, cranges, ncranges); + store_combine_ranges(range, cranges, ncranges); MemoryContextSwitchTo(oldctx); MemoryContextDelete(ctx); /* Did we break the ranges somehow? */ + AssertCheckRanges(range, cmpFn, colloid); + + return true; +} + +/* + * range_add_value + * Add the new value to the multi-minmax range. + */ +static bool +range_add_value(BrinDesc *bdesc, Oid colloid, + AttrNumber attno, Form_pg_attribute attr, + Ranges *ranges, Datum newval) +{ + FmgrInfo *cmpFn; + bool modified = false; + + /* we'll certainly need the comparator, so just look it up now */ + cmpFn = minmax_multi_get_strategy_procinfo(bdesc, attno, attr->atttypid, + BTLessStrategyNumber); + + /* comprehensive checks of the input ranges */ AssertCheckRanges(ranges, cmpFn, colloid); + + /* + * Make sure there's enough free space in the buffer. We only trigger + * this when the buffer is full, which means it had to be modified as + * we size it to be larger than what is stored on disk. + * + * XXX This needs to happen before we check if the value is contained + * in the range, because the value might be in the unsorted part, and + * we don't check that in range_contains_value. The deduplication would + * then move it to the sorted part, and we'd add the value too, which + * violates the rule that we never have duplicates with the ranges + * or sorted values. + * + * XXX At the moment this only does the deduplication. + * + * XXX We might also deduplicate and recheck if the value is contained, + * but that seems like an overkill. We'd need to deduplicate anyway, + * so why not do it now. + */ + modified = ensure_free_space_in_buffer(bdesc, colloid, + attno, attr, ranges); + + /* + * Bail out if the value already is covered by the range. + * + * We could also add values until we hit values_per_range, and then + * do the deduplication in a batch, hoping for better efficiency. But + * that would mean we actually modify the range every time, which means + * having to serialize the value, which does palloc, walks the values, + * copies them, etc. Not exactly cheap. + * + * So instead we do the check, which should be fairly cheap - assuming + * the comparator function is not very expensive. + * + * This also implies means the values array can't contain duplicities. + */ + if (range_contains_value(bdesc, colloid, attno, attr, ranges, newval)) + return modified; + + /* Make a copy of the value, if needed. */ + newval = datumCopy(newval, attr->attbyval, attr->attlen); + + /* + * If there's space in the values array, copy it in and we're done. + * + * We do want to keep the values sorted (to speed up searches), so we + * do a simple insertion sort. We could do something more elaborate, + * e.g. by sorting the values only now and then, but for small counts + * (e.g. when maxvalues is 64) this should be fine. + */ + ranges->values[2*ranges->nranges + ranges->nvalues] = newval; + ranges->nvalues++; + + /* + * Check we haven't broken the ordering of boundary values (checks + * both parts, but that doesn't hurt). + */ + AssertCheckRanges(ranges, cmpFn, colloid); + + /* Also check the range contains the value we just added. */ // FIXME Assert(ranges, cmpFn, colloid); + /* yep, we've modified the range */ return true; } @@ -1506,12 +1661,6 @@ compactify_ranges(BrinDesc *bdesc, Ranges *ranges, int max_values) MemoryContext ctx; MemoryContext oldctx; - /* - * This should only be used in batch mode, and there should be no - * ranges, just individual values. - */ - Assert((ranges->batch_mode) && (ranges->nranges == 0)); - /* we'll certainly need the comparator, so just look it up now */ cmpFn = minmax_multi_get_strategy_procinfo(bdesc, ranges->attno, ranges->typid, BTLessStrategyNumber); @@ -1534,8 +1683,7 @@ compactify_ranges(BrinDesc *bdesc, Ranges *ranges, int max_values) /* OK build the combine ranges */ cranges = build_combine_ranges(cmpFn, ranges->colloid, ranges, - false, (Datum) 0, &ncranges, - true); /* deduplicate */ + &ncranges); /* deduplicate */ if (ncranges > 1) { @@ -1548,7 +1696,7 @@ compactify_ranges(BrinDesc *bdesc, Ranges *ranges, int max_values) * don't expect more tuples to be inserted soon. */ ncranges = reduce_combine_ranges(cranges, ncranges, distances, - max_values, cmpFn, ranges->colloid); + max_values, cmpFn, ranges->colloid); Assert(count_values(cranges, ncranges) <= max_values); } @@ -2052,8 +2200,7 @@ brin_minmax_multi_serialize(BrinDesc *bdesc, Datum src, Datum *dst) * In batch mode, we need to compress the accumulated values to the * actually requested number of values/ranges. */ - if (ranges->batch_mode) - compactify_ranges(bdesc, ranges, ranges->target_maxvalues); + compactify_ranges(bdesc, ranges, ranges->target_maxvalues); s = range_serialize(ranges); dst[0] = PointerGetDatum(s); @@ -2114,15 +2261,39 @@ brin_minmax_multi_add_value(PG_FUNCTION_ARGS) { MemoryContext oldctx; + int target_maxvalues; + int maxvalues; BlockNumber pagesPerRange = BrinGetPagesPerRange(bdesc->bd_index); + /* what was specified as a reloption? */ + target_maxvalues = brin_minmax_multi_get_values(bdesc, opts); + + /* + * Determine the insert buffer size - we use 10x the target, capped + * to the maximum number of values in the heap range. This is more + * than enough, considering the actual number of rows per page is + * likely much lower, but meh. + */ + maxvalues = Min(target_maxvalues * MINMAX_BUFFER_FACTOR, + MaxHeapTuplesPerPage * pagesPerRange); + + /* but always at least the original value */ + maxvalues = Max(maxvalues, target_maxvalues); + + /* always cap by MIN/MAX */ + maxvalues = Max(maxvalues, MINMAX_BUFFER_MIN); + maxvalues = Min(maxvalues, MINMAX_BUFFER_MAX); + oldctx = MemoryContextSwitchTo(column->bv_context); - ranges = minmax_multi_init(MaxHeapTuplesPerPage * pagesPerRange); + ranges = minmax_multi_init(maxvalues); ranges->attno = attno; ranges->colloid = colloid; ranges->typid = attr->atttypid; - ranges->batch_mode = true; - ranges->target_maxvalues = brin_minmax_multi_get_values(bdesc, opts); + ranges->target_maxvalues = target_maxvalues; + + /* we'll certainly need the comparator, so just look it up now */ + ranges->cmp = minmax_multi_get_strategy_procinfo(bdesc, attno, attr->atttypid, + BTLessStrategyNumber); MemoryContextSwitchTo(oldctx); @@ -2136,10 +2307,38 @@ brin_minmax_multi_add_value(PG_FUNCTION_ARGS) { MemoryContext oldctx; + int maxvalues; + BlockNumber pagesPerRange = BrinGetPagesPerRange(bdesc->bd_index); + oldctx = MemoryContextSwitchTo(column->bv_context); serialized = (SerializedRanges *) PG_DETOAST_DATUM(column->bv_values[0]); - ranges = range_deserialize(serialized); + + /* + * Determine the insert buffer size - we use 10x the target, capped + * to the maximum number of values in the heap range. This is more + * than enough, considering the actual number of rows per page is + * likely much lower, but meh. + */ + maxvalues = Min(serialized->maxvalues * MINMAX_BUFFER_FACTOR, + MaxHeapTuplesPerPage * pagesPerRange); + + /* but always at least the original value */ + maxvalues = Max(maxvalues, serialized->maxvalues); + + /* always cap by MIN/MAX */ + maxvalues = Max(maxvalues, MINMAX_BUFFER_MIN); + maxvalues = Min(maxvalues, MINMAX_BUFFER_MAX); + + ranges = range_deserialize(maxvalues, serialized); + + ranges->attno = attno; + ranges->colloid = colloid; + ranges->typid = attr->atttypid; + + /* we'll certainly need the comparator, so just look it up now */ + ranges->cmp = minmax_multi_get_strategy_procinfo(bdesc, attno, attr->atttypid, + BTLessStrategyNumber); column->bv_mem_value = PointerGetDatum(ranges); column->bv_serialize = brin_minmax_multi_serialize; @@ -2184,7 +2383,7 @@ brin_minmax_multi_consistent(PG_FUNCTION_ARGS) attno = column->bv_attno; serialized = (SerializedRanges *) PG_DETOAST_DATUM(column->bv_values[0]); - ranges = range_deserialize(serialized); + ranges = range_deserialize(serialized->maxvalues, serialized); /* inspect the ranges, and for each one evaluate the scan keys */ for (rangeno = 0; rangeno < ranges->nranges; rangeno++) @@ -2371,8 +2570,8 @@ brin_minmax_multi_union(PG_FUNCTION_ARGS) serialized_a = (SerializedRanges *) PG_DETOAST_DATUM(col_a->bv_values[0]); serialized_b = (SerializedRanges *) PG_DETOAST_DATUM(col_b->bv_values[0]); - ranges_a = range_deserialize(serialized_a); - ranges_b = range_deserialize(serialized_b); + ranges_a = range_deserialize(serialized_a->maxvalues, serialized_a); + ranges_b = range_deserialize(serialized_b->maxvalues, serialized_b); /* make sure neither of the ranges is NULL */ Assert(ranges_a && ranges_b); @@ -2408,7 +2607,7 @@ brin_minmax_multi_union(PG_FUNCTION_ARGS) cmpFn = minmax_multi_get_strategy_procinfo(bdesc, attno, attr->atttypid, BTLessStrategyNumber); - /* sort the combine ranges (don't deduplicate) */ + /* sort the combine ranges (no need to deduplicate) */ sort_combine_ranges(cmpFn, colloid, cranges, ncranges, false); /* @@ -2637,7 +2836,7 @@ brin_minmax_multi_summary_out(PG_FUNCTION_ARGS) fmgr_info(outfunc, &fmgrinfo); /* deserialize the range info easy-to-process pieces */ - ranges_deserialized = range_deserialize(ranges); + ranges_deserialized = range_deserialize(ranges->maxvalues, ranges); appendStringInfo(&str, "nranges: %u nvalues: %u maxvalues: %u", ranges_deserialized->nranges, -- 2.26.2 --------------0E72B707603BED22B4040825 Content-Type: text/x-patch; charset=UTF-8; name="0006-Batch-mode-when-building-new-BRIN-multi-min-20210211.patch" Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename*0="0006-Batch-mode-when-building-new-BRIN-multi-min-20210211.pa"; filename*1="tch" ^ permalink raw reply [nested|flat] 3+ messages in thread
* Add Pipelining support in psql @ 2024-11-27 09:50 Anthonin Bonnefoy <[email protected]> 0 siblings, 1 reply; 3+ messages in thread From: Anthonin Bonnefoy @ 2024-11-27 09:50 UTC (permalink / raw) To: pgsql-hackers Hi, With \bind, \parse, \bind_named and \close, it is possible to issue queries from psql using the extended protocol. However, it wasn't possible to send those queries using pipelining and the only way to test pipelined queries was through pgbench's tap tests. The attached patch adds pipelining support to psql with 3 new meta-commands, mirroring what's already done in pgbench: - \startpipeline starts a new pipeline. All extended queries will be queued until the end of the pipeline is reached. - \endpipeline ends an ongoing pipeline. All queued commands will be sent to the server and all responses will be processed by the psql. - \syncpipeline queue a synchronisation point without flushing the commands to the server Those meta-commands will allow testing pipelined query behaviour using psql regression tests. Regards, Anthonin Attachments: [application/octet-stream] v01-0001-Add-pipelining-support-in-psql.patch (30.8K, ../../CAO6_XqroE7JuMEm1sWz55rp9fAYX2JwmcP_3m_v51vnOFdsLiQ@mail.gmail.com/2-v01-0001-Add-pipelining-support-in-psql.patch) download | inline diff: From d972efc1e629edc50495cbe023e90143e3f9f181 Mon Sep 17 00:00:00 2001 From: Anthonin Bonnefoy <[email protected]> Date: Tue, 5 Nov 2024 10:26:54 +0100 Subject: Add pipelining support in psql With \bind, \parse, \bind_named and \close, it is possible to issue queries from psql using the extended protocol. However, it wasn't possible to send those queries using pipelining and the only way to test pipelined queries was through pgbench's tap tests. This patch adds additional psql meta-commands to support pipelining: \startpipeline, \endpipeline and \syncpipeline, mirroring the existing meta-commands in pgbench. \startpipeline starts a new pipeline. All extended queries will be queued until the end of the pipeline is reached. \endpipeline ends an ongoing pipeline. All queued commands will be sent to the server and all responses will be processed by the psql. \syncpipeline queue a synchronisation point without flushing the commands to the server Those meta-commands will allow to test pipeline behaviour using psql regression tests. --- doc/src/sgml/ref/psql-ref.sgml | 59 +++++ src/bin/psql/command.c | 77 +++++++ src/bin/psql/common.c | 98 ++++++++- src/bin/psql/help.c | 3 + src/bin/psql/settings.h | 4 + src/bin/psql/tab-complete.in.c | 4 +- src/test/regress/expected/psql.out | 335 +++++++++++++++++++++++++++++ src/test/regress/sql/psql.sql | 209 ++++++++++++++++++ 8 files changed, 785 insertions(+), 4 deletions(-) diff --git a/doc/src/sgml/ref/psql-ref.sgml b/doc/src/sgml/ref/psql-ref.sgml index e42073ed748..ed166f757a6 100644 --- a/doc/src/sgml/ref/psql-ref.sgml +++ b/doc/src/sgml/ref/psql-ref.sgml @@ -3562,6 +3562,65 @@ testdb=> <userinput>\setenv LESS -imx4F</userinput> </listitem> </varlistentry> + <varlistentry id="app-psql-meta-command-pipeline"> + <term><literal>\startpipeline</literal></term> + <term><literal>\syncpipeline</literal></term> + <term><literal>\endpipeline</literal></term> + + <listitem> + <para> + This group of commands implements pipelining of SQL statements. + A pipeline must begin with a <command>\startpipeline</command> + and end with an <command>\endpipeline</command>. In between there + may be any number of <command>\syncpipeline</command> commands, + which sends a <link linkend="protocol-flow-ext-query">sync message</link> + without ending the ongoing pipeline and flushing the send buffer. + In pipeline mode, statements are sent to the server without waiting + for the results of previous statements. See + <xref linkend="libpq-pipeline-mode"/> for more details. + </para> + + <para> + Pipeline mode requires the use of extended query protocol. All queries need + to be sent using the meta-commands <literal>\bind</literal>, + <literal>\bind_named</literal>, <literal>\close</literal> or + <literal>\parse</literal>. While a pipeline is ongoing, + <literal>\g</literal> will append the current query buffer to the pipeline and + other meta-commands like <literal>\gx</literal> or <literal>\gdesc</literal> + are not allowed in pipeline mode. + </para> + + <para> + </para> + + <para> + Example: +<programlisting> +\startpipeline +-- Pipe a Parse, Bind and Execute of "SELECT 1" +SELECT 1 \bind \g +-- Pipe a Parse of "SELECT $1", storing the result in the prepared statement 'stmt1' +SELECT $1 \parse stmt1 +-- Pipe a Bind and Execute of the prepared statement 'stmt1' with parameter 1 +\bind_named stmt1 1 \g +-- Pipe a Parse, Bind and Execute of "SELECT pg_current_xact_id()" +SELECT pg_current_xact_id() \bind \g +-- Pipe a synchronisation point. This will commit the implicit transaction +-- block started by the pipeline +\syncpipeline +-- Pipe a Parse, Bind and Execute of SELECT pg_current_xact_id() +-- This will display a different xid as a new transaction block +-- was started after the synchronisation point +SELECT pg_current_xact_id() \bind \g +-- Pipe a Close of the prepared statement 'stmt1' +\close stmt1 +-- End the pipeline, sending all piped commands to the server and process the results +\endpipeline +</programlisting></para> + + </listitem> + </varlistentry> + <varlistentry id="app-psql-meta-command-t-lc"> <term><literal>\t</literal></term> diff --git a/src/bin/psql/command.c b/src/bin/psql/command.c index 1f3cbb11f7c..067329d241a 100644 --- a/src/bin/psql/command.c +++ b/src/bin/psql/command.c @@ -90,6 +90,7 @@ static backslashResult exec_command_else(PsqlScanState scan_state, ConditionalSt PQExpBuffer query_buf); static backslashResult exec_command_endif(PsqlScanState scan_state, ConditionalStack cstack, PQExpBuffer query_buf); +static backslashResult exec_command_endpipeline(PsqlScanState scan_state, bool active_branch); static backslashResult exec_command_encoding(PsqlScanState scan_state, bool active_branch); static backslashResult exec_command_errverbose(PsqlScanState scan_state, bool active_branch); static backslashResult exec_command_f(PsqlScanState scan_state, bool active_branch); @@ -132,6 +133,8 @@ static backslashResult exec_command_setenv(PsqlScanState scan_state, bool active const char *cmd); static backslashResult exec_command_sf_sv(PsqlScanState scan_state, bool active_branch, const char *cmd, bool is_func); +static backslashResult exec_command_startpipeline(PsqlScanState scan_state, bool active_branch); +static backslashResult exec_command_syncpipeline(PsqlScanState scan_state, bool active_branch); static backslashResult exec_command_t(PsqlScanState scan_state, bool active_branch); static backslashResult exec_command_T(PsqlScanState scan_state, bool active_branch); static backslashResult exec_command_timing(PsqlScanState scan_state, bool active_branch); @@ -351,6 +354,8 @@ exec_command(const char *cmd, status = exec_command_else(scan_state, cstack, query_buf); else if (strcmp(cmd, "endif") == 0) status = exec_command_endif(scan_state, cstack, query_buf); + else if (strcmp(cmd, "endpipeline") == 0) + status = exec_command_endpipeline(scan_state, active_branch); else if (strcmp(cmd, "encoding") == 0) status = exec_command_encoding(scan_state, active_branch); else if (strcmp(cmd, "errverbose") == 0) @@ -408,6 +413,10 @@ exec_command(const char *cmd, status = exec_command_sf_sv(scan_state, active_branch, cmd, true); else if (strcmp(cmd, "sv") == 0 || strcmp(cmd, "sv+") == 0) status = exec_command_sf_sv(scan_state, active_branch, cmd, false); + else if (strcmp(cmd, "startpipeline") == 0) + status = exec_command_startpipeline(scan_state, active_branch); + else if (strcmp(cmd, "syncpipeline") == 0) + status = exec_command_syncpipeline(scan_state, active_branch); else if (strcmp(cmd, "t") == 0) status = exec_command_t(scan_state, active_branch); else if (strcmp(cmd, "T") == 0) @@ -1526,6 +1535,13 @@ exec_command_g(PsqlScanState scan_state, bool active_branch, const char *cmd) if (status == PSQL_CMD_SKIP_LINE && active_branch) { + if (strcmp(cmd, "gx") == 0 && PQpipelineStatus(pset.db) == PQ_PIPELINE_ON) + { + pg_log_error("\\gx not allowed in pipeline mode"); + clean_extended_state(); + return PSQL_CMD_ERROR; + } + if (!fname) pset.gfname = NULL; else @@ -1689,6 +1705,12 @@ exec_command_gexec(PsqlScanState scan_state, bool active_branch) if (active_branch) { + if (PQpipelineStatus(pset.db) == PQ_PIPELINE_ON) + { + pg_log_error("\\gexec not allowed in pipeline mode"); + clean_extended_state(); + return PSQL_CMD_ERROR; + } pset.gexec_flag = true; status = PSQL_CMD_SEND; } @@ -1709,6 +1731,13 @@ exec_command_gset(PsqlScanState scan_state, bool active_branch) char *prefix = psql_scan_slash_option(scan_state, OT_NORMAL, NULL, false); + if (PQpipelineStatus(pset.db) == PQ_PIPELINE_ON) + { + pg_log_error("\\gset not allowed in pipeline mode"); + clean_extended_state(); + return PSQL_CMD_ERROR; + } + if (prefix) pset.gset_prefix = prefix; else @@ -2672,6 +2701,54 @@ exec_command_sf_sv(PsqlScanState scan_state, bool active_branch, return status; } +/* + * \startpipeline -- enter pipeline mode + */ +static backslashResult +exec_command_startpipeline(PsqlScanState scan_state, bool active_branch) +{ + if (active_branch) + { + pset.send_mode = PSQL_START_PIPELINE_MODE; + } + else + ignore_slash_options(scan_state); + + return PSQL_CMD_SEND; +} + +/* + * \syncpipeline -- send a sync message to an active pipeline + */ +static backslashResult +exec_command_syncpipeline(PsqlScanState scan_state, bool active_branch) +{ + if (active_branch) + { + pset.send_mode = PSQL_SEND_PIPELINE_SYNC; + } + else + ignore_slash_options(scan_state); + + return PSQL_CMD_SEND; +} + +/* + * \endpipeline -- end pipeline mode + */ +static backslashResult +exec_command_endpipeline(PsqlScanState scan_state, bool active_branch) +{ + if (active_branch) + { + pset.send_mode = PSQL_END_PIPELINE_MODE; + } + else + ignore_slash_options(scan_state); + + return PSQL_CMD_SEND; +} + /* * \t -- turn off table headers and row count */ diff --git a/src/bin/psql/common.c b/src/bin/psql/common.c index 8a9211db41a..3c0ad4dd9ce 100644 --- a/src/bin/psql/common.c +++ b/src/bin/psql/common.c @@ -415,6 +415,8 @@ AcceptResult(const PGresult *result, bool show_error) case PGRES_EMPTY_QUERY: case PGRES_COPY_IN: case PGRES_COPY_OUT: + case PGRES_PIPELINE_SYNC: + case PGRES_PIPELINE_ABORTED: /* Fine, do nothing */ OK = true; break; @@ -1418,6 +1420,28 @@ DescribeQuery(const char *query, double *elapsed_msec) return OK; } +/* + * Read and discard all results until a synchronisation point is found. + */ +static PGresult * +discardUntilSync(void) +{ + for (;;) + { + PGresult *res = PQgetResult(pset.db); + ExecStatusType result_status = PQresultStatus(res); + + if (result_status == PGRES_PIPELINE_SYNC) + return res; + + /* + * An aborted pipeline will have either NULL results or results in an + * PGRES_PIPELINE_ABORTED status + */ + Assert(res == NULL || result_status == PGRES_PIPELINE_ABORTED); + PQclear(res); + } +} /* * ExecQueryAndProcessResults: utility function for use by SendQuery() @@ -1451,6 +1475,7 @@ ExecQueryAndProcessResults(const char *query, bool timing = pset.timing; bool success = false; bool return_early = false; + bool process_pipeline = false; instr_time before, after; PGresult *result; @@ -1484,6 +1509,21 @@ ExecQueryAndProcessResults(const char *query, (const char *const *) pset.bind_params, NULL, NULL, 0); break; + case PSQL_START_PIPELINE_MODE: + success = PQenterPipelineMode(pset.db); + break; + case PSQL_END_PIPELINE_MODE: + success = PQpipelineSync(pset.db); + /* End of the pipeline, all queued commands need to be processed */ + process_pipeline = true; + if (success) + pset.num_syncs++; + break; + case PSQL_SEND_PIPELINE_SYNC: + success = PQsendPipelineSync(pset.db); + if (success) + pset.num_syncs++; + break; case PSQL_SEND_QUERY: success = PQsendQuery(pset.db, query); break; @@ -1501,6 +1541,15 @@ ExecQueryAndProcessResults(const char *query, return -1; } + if (!process_pipeline && PQpipelineStatus(pset.db) == PQ_PIPELINE_ON) + { + /* + * We're in a pipeline and haven't received a pipeline end so there's + * no result to process yet. + */ + return 0; + } + /* * Fetch the result in chunks if FETCH_COUNT is set, except when: * @@ -1585,6 +1634,15 @@ ExecQueryAndProcessResults(const char *query, * ignore manually. */ result = NULL; + else if (process_pipeline) + { + /* + * We have an error within a pipeline. All commands are + * aborted until the next synchronisation point. We need to + * consume all results until this synchronisation point. + */ + result = discardUntilSync(); + } else result = PQgetResult(pset.db); @@ -1771,12 +1829,38 @@ ExecQueryAndProcessResults(const char *query, } } + if (result_status == PGRES_PIPELINE_SYNC) + { + /* We have a sync response, decrease the sync counter */ + pset.num_syncs--; + + /* + * After a synchronisation point, reset success state to print + * possible successful results + */ + success = true; + /* If all syncs were processed, exit pipeline mode */ + if (pset.num_syncs <= 0) + success &= PQexitPipelineMode(pset.db); + } + /* * Check PQgetResult() again. In the typical case of a single-command * string, it will return NULL. Otherwise, we'll have other results * to process. We need to do that to check whether this is the last. */ next_result = PQgetResult(pset.db); + if (process_pipeline && result_status != PGRES_PIPELINE_SYNC) + { + /* + * In pipeline mode, a NULL result indicates the end of the + * current query being processed. We need to call PQgetResult a + * second time to move to the next response. + */ + Assert(next_result == NULL); + next_result = PQgetResult(pset.db); + } + last = (next_result == NULL); /* @@ -1798,8 +1882,12 @@ ExecQueryAndProcessResults(const char *query, *elapsed_msec = INSTR_TIME_GET_MILLISEC(after); } - /* this may or may not print something depending on settings */ - if (result != NULL) + /* + * This may or may not print something depending on settings. A + * pipeline sync will have a non null result but doesn't have anything + * to print, thus we ignore them + */ + if (result != NULL && result_status != PGRES_PIPELINE_SYNC) { /* * If results need to be printed into the file specified by \g, @@ -1837,6 +1925,9 @@ ExecQueryAndProcessResults(const char *query, /* close \g file if we opened it */ CloseGOutput(gfile_fout, gfile_is_pipe); + /* After query process, pipeline num_syncs should be 0 */ + Assert(pset.num_syncs == 0); + /* may need this to recover from conn loss during COPY */ if (!CheckConnection()) return -1; @@ -2296,6 +2387,9 @@ clean_extended_state(void) free(pset.stmtName); pset.bind_params = NULL; break; + case PSQL_START_PIPELINE_MODE: /* \startpipeline */ + case PSQL_END_PIPELINE_MODE: /* \endpipeline */ + case PSQL_SEND_PIPELINE_SYNC: /* \syncpipeline */ case PSQL_SEND_QUERY: break; } diff --git a/src/bin/psql/help.c b/src/bin/psql/help.c index 3f4afc2d141..e550a29297c 100644 --- a/src/bin/psql/help.c +++ b/src/bin/psql/help.c @@ -167,6 +167,7 @@ slashUsage(unsigned short int pager) HELP0(" \\close STMT_NAME close an existing prepared statement\n"); HELP0(" \\copyright show PostgreSQL usage and distribution terms\n"); HELP0(" \\crosstabview [COLUMNS] execute query and display result in crosstab\n"); + HELP0(" \\endpipeline exit pipeline mode\n"); HELP0(" \\errverbose show most recent error message at maximum verbosity\n"); HELP0(" \\g [(OPTIONS)] [FILE] execute query (and send result to file or |pipe);\n" " \\g with no arguments is equivalent to a semicolon\n"); @@ -176,6 +177,8 @@ slashUsage(unsigned short int pager) HELP0(" \\gx [(OPTIONS)] [FILE] as \\g, but forces expanded output mode\n"); HELP0(" \\parse STMT_NAME create a prepared statement\n"); HELP0(" \\q quit psql\n"); + HELP0(" \\startpipeline enter pipeline mode\n"); + HELP0(" \\syncpipeline add a synchronisation point to an ongoing pipeline\n"); HELP0(" \\watch [[i=]SEC] [c=N] [m=MIN]\n" " execute query every SEC seconds, up to N times,\n" " stop if less than MIN rows are returned\n"); diff --git a/src/bin/psql/settings.h b/src/bin/psql/settings.h index a22de8ef78e..55247c4dc33 100644 --- a/src/bin/psql/settings.h +++ b/src/bin/psql/settings.h @@ -69,6 +69,9 @@ typedef enum PSQL_SEND_EXTENDED_PARSE, PSQL_SEND_EXTENDED_QUERY_PARAMS, PSQL_SEND_EXTENDED_QUERY_PREPARED, + PSQL_SEND_PIPELINE_SYNC, + PSQL_START_PIPELINE_MODE, + PSQL_END_PIPELINE_MODE, } PSQL_SEND_MODE; typedef enum @@ -108,6 +111,7 @@ typedef struct _psqlSettings PSQL_SEND_MODE send_mode; /* one-shot request to send query with normal * or extended query protocol */ int bind_nparams; /* number of parameters */ + int num_syncs; /* number of ongoing syncs */ char **bind_params; /* parameters for extended query protocol call */ char *stmtName; /* prepared statement name used for extended * query protocol commands */ diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c index fad2277991d..742d2627afc 100644 --- a/src/bin/psql/tab-complete.in.c +++ b/src/bin/psql/tab-complete.in.c @@ -1867,7 +1867,7 @@ psql_completion(const char *text, int start, int end) "\\drds", "\\drg", "\\dRs", "\\dRp", "\\ds", "\\dt", "\\dT", "\\dv", "\\du", "\\dx", "\\dX", "\\dy", "\\echo", "\\edit", "\\ef", "\\elif", "\\else", "\\encoding", - "\\endif", "\\errverbose", "\\ev", + "\\endif", "\\endpipeline", "\\errverbose", "\\ev", "\\f", "\\g", "\\gdesc", "\\getenv", "\\gexec", "\\gset", "\\gx", "\\help", "\\html", @@ -1877,7 +1877,7 @@ psql_completion(const char *text, int start, int end) "\\parse", "\\password", "\\print", "\\prompt", "\\pset", "\\qecho", "\\quit", "\\reset", - "\\s", "\\set", "\\setenv", "\\sf", "\\sv", + "\\s", "\\set", "\\setenv", "\\sf", "\\startpipeline", "\\sv", "\\syncpipeline", "\\t", "\\T", "\\timing", "\\unset", "\\x", diff --git a/src/test/regress/expected/psql.out b/src/test/regress/expected/psql.out index 36dc31c16c4..6e05f908484 100644 --- a/src/test/regress/expected/psql.out +++ b/src/test/regress/expected/psql.out @@ -6828,3 +6828,338 @@ CREATE TABLE defprivs (a int); \pset null '' DROP TABLE defprivs; +-- pipelining +CREATE TABLE psql_pipeline(a INTEGER PRIMARY KEY, s TEXT); +-- single query +\startpipeline +SELECT $1 \bind 'val1' \g +\endpipeline + ?column? +---------- + val1 +(1 row) + +-- multiple queries +\startpipeline +SELECT $1 \bind 'val1' \g +SELECT $1, $2 \bind 'val2' 'val3' \g +SELECT $1, $2 \bind 'val2' 'val3' \g +\endpipeline + ?column? +---------- + val1 +(1 row) + + ?column? | ?column? +----------+---------- + val2 | val3 +(1 row) + + ?column? | ?column? +----------+---------- + val2 | val3 +(1 row) + +-- send multiple syncs +\startpipeline +SELECT $1 \bind 'val1' \g +\syncpipeline +\syncpipeline +SELECT $1, $2 \bind 'val2' 'val3' \g +\syncpipeline +SELECT $1, $2 \bind 'val4' 'val5' \g +\endpipeline + ?column? +---------- + val1 +(1 row) + + ?column? | ?column? +----------+---------- + val2 | val3 +(1 row) + + ?column? | ?column? +----------+---------- + val4 | val5 +(1 row) + +-- startpipeline shouldn't have any effect if already in a pipeline +\startpipeline +\startpipeline +SELECT $1 \bind 'val1' \g +\endpipeline + ?column? +---------- + val1 +(1 row) + +-- Convert an implicit tx block to an explicit tx block +\startpipeline +INSERT INTO psql_pipeline VALUES ($1) \bind 1 \g +BEGIN \bind \g +INSERT INTO psql_pipeline VALUES ($1) \bind 2 \g +ROLLBACK \bind \g +\endpipeline +-- Multiple explicit transactions +\startpipeline +BEGIN \bind \g +INSERT INTO psql_pipeline VALUES ($1) \bind 1 \g +ROLLBACK \bind \g +BEGIN \bind \g +INSERT INTO psql_pipeline VALUES ($1) \bind 1 \g +COMMIT \bind \g +\endpipeline +-- COPY FROM STDIN +\startpipeline +SELECT $1 \bind 'val1' \g +COPY psql_pipeline FROM STDIN \bind \g +\endpipeline + ?column? +---------- + val1 +(1 row) + +-- COPY TO STDOUT +\startpipeline +SELECT $1 \bind 'val1' \g +copy psql_pipeline TO STDOUT \bind \g +SELECT $1 \bind 'val2' \g +\endpipeline + ?column? +---------- + val1 +(1 row) + +1 \N +2 test2 +3 test3 + ?column? +---------- + val2 +(1 row) + +-- Use \parse and \bind_named +\startpipeline +SELECT $1 \parse '' +SELECT $1, $2 \parse '' +SELECT $2 \parse pipeline_1 +\bind_named '' 1 2 \g +\bind_named pipeline_1 2 \g +\endpipeline +ERROR: could not determine data type of parameter $1 +-- pipelining errors +-- endpipeline outside of pipeline should fail +\endpipeline +cannot send pipeline when not in pipeline mode +-- Query using simple protocol should not be sent and should leave the pipeline usable +\startpipeline +SELECT 1; +PQsendQuery not allowed in pipeline mode +SELECT $1 \bind 'val1' \g +\endpipeline + ?column? +---------- + val1 +(1 row) + +-- After an aborted pipeline, commands after a sync should be displayed +\startpipeline +SELECT $1 \bind \g +\syncpipeline +SELECT $1 \bind 1 \g +\endpipeline +ERROR: bind message supplies 0 parameters, but prepared statement "" requires 1 + ?column? +---------- + 1 +(1 row) + +-- Incorrect number of parameters, the pipeline will be aborted and following queries won't be executed +\startpipeline +SELECT \bind 'val1' \g +SELECT $1 \bind 'val1' \g +\endpipeline +ERROR: bind message supplies 1 parameters, but prepared statement "" requires 0 +-- An explicit transaction with an error needs to be rollbacked after the pipeline +\startpipeline +BEGIN \bind \g +INSERT INTO psql_pipeline VALUES ($1) \bind 1 \g +ROLLBACK \bind \g +\endpipeline +ERROR: duplicate key value violates unique constraint "psql_pipeline_pkey" +DETAIL: Key (a)=(1) already exists. +ROLLBACK; +-- \watch sends a simple query which won't be allowed within a pipeline +\startpipeline +SELECT \bind \g +\watch 1 +PQsendQuery not allowed in pipeline mode + +\endpipeline +-- +(1 row) + +-- \gdesc should fail as synchronous commands are not allowed in pipeline, pipeline should still be usable +\startpipeline +SELECT $1 \bind 1 \gdesc +synchronous command execution functions are not allowed in pipeline mode +SELECT $1 \bind 1 \g +\endpipeline + ?column? +---------- + 1 +(1 row) + +-- \gset is not allowed, pipeline should still be usable +\startpipeline +SELECT $1 as i, $2 as j \parse '' +SELECT $1 as k, $2 as l \parse 'second' +\bind_named '' 1 2 \gset +\gset not allowed in pipeline mode +\bind_named second 1 2 \gset pref02_ \echo :pref02_i :pref02_j +\gset not allowed in pipeline mode +\bind_named '' 1 2 \g +\endpipeline + i | j +---+--- + 1 | 2 +(1 row) + +-- \gx is not allowed, pipeline should still be usable +\startpipeline +SELECT $1 \bind 1 \gx +\gx not allowed in pipeline mode +\reset +SELECT $1 \bind 1 \g +\endpipeline + ?column? +---------- + 1 +(1 row) + +-- \gexec is not allowed, pipeline should still be usable +\startpipeline +SELECT 'INSERT INTO psql_pipeline(a) SELECT generate_series(1, 10)' \parse 'insert_stmt' +\bind_named insert_stmt \gexec +\gexec not allowed in pipeline mode +\bind_named insert_stmt \g +SELECT COUNT(*) FROM psql_pipeline \bind \g +\endpipeline + ?column? +------------------------------------------------------------ + INSERT INTO psql_pipeline(a) SELECT generate_series(1, 10) +(1 row) + + count +------- + 3 +(1 row) + +-- pipelining and transaction block behaviour +-- set local will issue a warning when modifying a GUC outside of a transaction block +-- The change will still be valid as a pipeline runs within an implicit transaction block +-- Sending a sync will commit the implicit transaction block. The first command after a sync +-- won't be seen as belonging to a pipeline. +\startpipeline +SET LOCAL statement_timeout='1h' \bind \g +SHOW statement_timeout \bind \g +\syncpipeline +SHOW statement_timeout \bind \g +SET LOCAL statement_timeout='2h' \bind \g +SHOW statement_timeout \bind \g +\endpipeline +WARNING: SET LOCAL can only be used in transaction blocks + statement_timeout +------------------- + 1h +(1 row) + + statement_timeout +------------------- + 0 +(1 row) + + statement_timeout +------------------- + 2h +(1 row) + +-- Reindex concurrently is forbidden in the middle of a pipeline +\startpipeline +SELECT $1 \bind 1 \g +REINDEX TABLE CONCURRENTLY psql_pipeline \bind \g +SELECT $1 \bind 2 \g +\endpipeline + ?column? +---------- + 1 +(1 row) + +ERROR: REINDEX CONCURRENTLY cannot run inside a transaction block +-- Reindex concurrently will work if it's the first command of a pipeline +\startpipeline +REINDEX TABLE CONCURRENTLY psql_pipeline \bind \g +SELECT $1 \bind 2 \g +\endpipeline + ?column? +---------- + 2 +(1 row) + +-- subtransactions are not allowed in pipeline mode +\startpipeline +SAVEPOINT a \bind \g +SELECT $1 \bind 1 \g +ROLLBACK TO SAVEPOINT a \bind \g +SELECT $1 \bind 2 \g +\endpipeline +ERROR: SAVEPOINT can only be used in transaction blocks +-- Lock command will fail as first pipeline command is not seen as a transaction block +\startpipeline +LOCK psql_pipeline \bind \g +SELECT $1 \bind 2 \g +\endpipeline +ERROR: LOCK TABLE can only be used in transaction blocks +-- Lock command will succeed after the first command as pipeline will be seen as an implicit transaction block +\startpipeline +SELECT $1 \bind 1 \g +LOCK psql_pipeline \bind \g +SELECT $1 \bind 2 \g +\endpipeline + ?column? +---------- + 1 +(1 row) + + ?column? +---------- + 2 +(1 row) + +-- Vacuum command will work as the first command +\startpipeline +VACUUM psql_pipeline \bind \g +\endpipeline +-- Vacuum command will fail within pipeline implicit transaction +\startpipeline +SELECT 1 \bind \g +VACUUM psql_pipeline \bind \g +\endpipeline + ?column? +---------- + 1 +(1 row) + +ERROR: VACUUM cannot run inside a transaction block +-- Vacuum command will work after a sync +\startpipeline +SELECT 1 \bind \g +\syncpipeline +VACUUM psql_pipeline \bind \g +\endpipeline + ?column? +---------- + 1 +(1 row) + diff --git a/src/test/regress/sql/psql.sql b/src/test/regress/sql/psql.sql index c5021fc0b13..e9bd8972996 100644 --- a/src/test/regress/sql/psql.sql +++ b/src/test/regress/sql/psql.sql @@ -1929,3 +1929,212 @@ CREATE TABLE defprivs (a int); \z defprivs \pset null '' DROP TABLE defprivs; + +-- pipelining +CREATE TABLE psql_pipeline(a INTEGER PRIMARY KEY, s TEXT); + +-- single query +\startpipeline +SELECT $1 \bind 'val1' \g +\endpipeline + +-- multiple queries +\startpipeline +SELECT $1 \bind 'val1' \g +SELECT $1, $2 \bind 'val2' 'val3' \g +SELECT $1, $2 \bind 'val2' 'val3' \g +\endpipeline + +-- send multiple syncs +\startpipeline +SELECT $1 \bind 'val1' \g +\syncpipeline +\syncpipeline +SELECT $1, $2 \bind 'val2' 'val3' \g +\syncpipeline +SELECT $1, $2 \bind 'val4' 'val5' \g +\endpipeline + +-- startpipeline shouldn't have any effect if already in a pipeline +\startpipeline +\startpipeline +SELECT $1 \bind 'val1' \g +\endpipeline + +-- Convert an implicit tx block to an explicit tx block +\startpipeline +INSERT INTO psql_pipeline VALUES ($1) \bind 1 \g +BEGIN \bind \g +INSERT INTO psql_pipeline VALUES ($1) \bind 2 \g +ROLLBACK \bind \g +\endpipeline + +-- Multiple explicit transactions +\startpipeline +BEGIN \bind \g +INSERT INTO psql_pipeline VALUES ($1) \bind 1 \g +ROLLBACK \bind \g +BEGIN \bind \g +INSERT INTO psql_pipeline VALUES ($1) \bind 1 \g +COMMIT \bind \g +\endpipeline + +-- COPY FROM STDIN +\startpipeline +SELECT $1 \bind 'val1' \g +COPY psql_pipeline FROM STDIN \bind \g +\endpipeline +2 test2 +3 test3 +\. + +-- COPY TO STDOUT +\startpipeline +SELECT $1 \bind 'val1' \g +copy psql_pipeline TO STDOUT \bind \g +SELECT $1 \bind 'val2' \g +\endpipeline + +-- Use \parse and \bind_named +\startpipeline +SELECT $1 \parse '' +SELECT $1, $2 \parse '' +SELECT $2 \parse pipeline_1 +\bind_named '' 1 2 \g +\bind_named pipeline_1 2 \g +\endpipeline + +-- pipelining errors + +-- endpipeline outside of pipeline should fail +\endpipeline + +-- Query using simple protocol should not be sent and should leave the pipeline usable +\startpipeline +SELECT 1; +SELECT $1 \bind 'val1' \g +\endpipeline + +-- After an aborted pipeline, commands after a sync should be displayed +\startpipeline +SELECT $1 \bind \g +\syncpipeline +SELECT $1 \bind 1 \g +\endpipeline + +-- Incorrect number of parameters, the pipeline will be aborted and following queries won't be executed +\startpipeline +SELECT \bind 'val1' \g +SELECT $1 \bind 'val1' \g +\endpipeline + +-- An explicit transaction with an error needs to be rollbacked after the pipeline +\startpipeline +BEGIN \bind \g +INSERT INTO psql_pipeline VALUES ($1) \bind 1 \g +ROLLBACK \bind \g +\endpipeline +ROLLBACK; + +-- \watch sends a simple query which won't be allowed within a pipeline +\startpipeline +SELECT \bind \g +\watch 1 +\endpipeline + +-- \gdesc should fail as synchronous commands are not allowed in pipeline, pipeline should still be usable +\startpipeline +SELECT $1 \bind 1 \gdesc +SELECT $1 \bind 1 \g +\endpipeline + +-- \gset is not allowed, pipeline should still be usable +\startpipeline +SELECT $1 as i, $2 as j \parse '' +SELECT $1 as k, $2 as l \parse 'second' +\bind_named '' 1 2 \gset +\bind_named second 1 2 \gset pref02_ \echo :pref02_i :pref02_j +\bind_named '' 1 2 \g +\endpipeline + +-- \gx is not allowed, pipeline should still be usable +\startpipeline +SELECT $1 \bind 1 \gx +\reset +SELECT $1 \bind 1 \g +\endpipeline + +-- \gexec is not allowed, pipeline should still be usable +\startpipeline +SELECT 'INSERT INTO psql_pipeline(a) SELECT generate_series(1, 10)' \parse 'insert_stmt' +\bind_named insert_stmt \gexec +\bind_named insert_stmt \g +SELECT COUNT(*) FROM psql_pipeline \bind \g +\endpipeline + +-- pipelining and transaction block behaviour + +-- set local will issue a warning when modifying a GUC outside of a transaction block +-- The change will still be valid as a pipeline runs within an implicit transaction block +-- Sending a sync will commit the implicit transaction block. The first command after a sync +-- won't be seen as belonging to a pipeline. +\startpipeline +SET LOCAL statement_timeout='1h' \bind \g +SHOW statement_timeout \bind \g +\syncpipeline +SHOW statement_timeout \bind \g +SET LOCAL statement_timeout='2h' \bind \g +SHOW statement_timeout \bind \g +\endpipeline + +-- Reindex concurrently is forbidden in the middle of a pipeline +\startpipeline +SELECT $1 \bind 1 \g +REINDEX TABLE CONCURRENTLY psql_pipeline \bind \g +SELECT $1 \bind 2 \g +\endpipeline + +-- Reindex concurrently will work if it's the first command of a pipeline +\startpipeline +REINDEX TABLE CONCURRENTLY psql_pipeline \bind \g +SELECT $1 \bind 2 \g +\endpipeline + +-- subtransactions are not allowed in pipeline mode +\startpipeline +SAVEPOINT a \bind \g +SELECT $1 \bind 1 \g +ROLLBACK TO SAVEPOINT a \bind \g +SELECT $1 \bind 2 \g +\endpipeline + +-- Lock command will fail as first pipeline command is not seen as a transaction block +\startpipeline +LOCK psql_pipeline \bind \g +SELECT $1 \bind 2 \g +\endpipeline + +-- Lock command will succeed after the first command as pipeline will be seen as an implicit transaction block +\startpipeline +SELECT $1 \bind 1 \g +LOCK psql_pipeline \bind \g +SELECT $1 \bind 2 \g +\endpipeline + +-- Vacuum command will work as the first command +\startpipeline +VACUUM psql_pipeline \bind \g +\endpipeline + +-- Vacuum command will fail within pipeline implicit transaction +\startpipeline +SELECT 1 \bind \g +VACUUM psql_pipeline \bind \g +\endpipeline + +-- Vacuum command will work after a sync +\startpipeline +SELECT 1 \bind \g +\syncpipeline +VACUUM psql_pipeline \bind \g +\endpipeline -- 2.39.5 (Apple Git-154) ^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Add Pipelining support in psql @ 2024-11-27 10:46 Kirill Reshke <[email protected]> parent: Anthonin Bonnefoy <[email protected]> 0 siblings, 0 replies; 3+ messages in thread From: Kirill Reshke @ 2024-11-27 10:46 UTC (permalink / raw) To: Anthonin Bonnefoy <[email protected]>; +Cc: pgsql-hackers On Wed, 27 Nov 2024 at 14:50, Anthonin Bonnefoy <[email protected]> wrote: > > Hi, > > With \bind, \parse, \bind_named and \close, it is possible to issue > queries from psql using the extended protocol. However, it wasn't > possible to send those queries using pipelining and the only way to > test pipelined queries was through pgbench's tap tests. Hello, good concept. Our connection pooler testing will greatly benefit from this feature. At the moment, our tests are golang-based and build pipelines in extended proto and verify the outcome. Regression tests based on SQL will be far more thorough and organic. > The attached patch adds pipelining support to psql with 3 new > meta-commands, mirroring what's already done in pgbench: > - \startpipeline starts a new pipeline. All extended queries will be > queued until the end of the pipeline is reached. > - \endpipeline ends an ongoing pipeline. All queued commands will be > sent to the server and all responses will be processed by the psql. > - \syncpipeline queue a synchronisation point without flushing the > commands to the server I'm very doubtful about the \syncpipeline . Maybe we should instead support \sync meta-command in psql? This will be a useful contribution itself. > Those meta-commands will allow testing pipelined query behaviour using > psql regression tests. > > Regards, > Anthonin I haven't looked into the patch in detail yet. -- Best regards, Kirill Reshke ^ permalink raw reply [nested|flat] 3+ messages in thread
end of thread, other threads:[~2024-11-27 10:46 UTC | newest] Thread overview: 3+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2021-02-02 00:57 [PATCH 7/9] Remove the special batch mode, use a larger buffer always Tomas Vondra <[email protected]> 2024-11-27 09:50 Add Pipelining support in psql Anthonin Bonnefoy <[email protected]> 2024-11-27 10:46 ` Re: Add Pipelining support in psql Kirill Reshke <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox