public inbox for [email protected]help / color / mirror / Atom feed
[PATCH 2/9] Move IS [NOT] NULL handling from BRIN support functions 4+ messages / 4 participants [nested] [flat]
* [PATCH 2/9] Move IS [NOT] NULL handling from BRIN support functions @ 2020-09-17 15:26 Tomas Vondra <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Tomas Vondra @ 2020-09-17 15:26 UTC (permalink / raw) The handling of IS [NOT] NULL clauses is independent of an opclass, and most of the code was exactly the same in both minmax and inclusion. So instead move the code from support procedures to the AM methods etc. This simplifies the code quite a bit - especially the support procedures quite a bit, as they don't need to care about NULL values and flags at all. It also means the IS [NOT] NULL clauses can be evaluated without invoking the support procedure at all. Author: Tomas Vondra <[email protected]> Author: Nikita Glukhov <[email protected]> Reviewed-by: Alvaro Herrera <[email protected]> Reviewed-by: Mark Dilger <[email protected]> Reviewed-by: Alexander Korotkov <[email protected]> Reviewed-by: Masahiko Sawada <[email protected]> Reviewed-by: John Naylor <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/backend/access/brin/brin.c | 293 +++++++++++++++++------ src/backend/access/brin/brin_inclusion.c | 106 +------- src/backend/access/brin/brin_minmax.c | 103 +------- src/include/access/brin_internal.h | 3 + 4 files changed, 239 insertions(+), 266 deletions(-) diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index dc187153aa..14da9ed17f 100644 --- a/src/backend/access/brin/brin.c +++ b/src/backend/access/brin/brin.c @@ -35,6 +35,7 @@ #include "storage/freespace.h" #include "utils/acl.h" #include "utils/builtins.h" +#include "utils/datum.h" #include "utils/index_selfuncs.h" #include "utils/memutils.h" #include "utils/rel.h" @@ -77,7 +78,9 @@ static void form_and_insert_tuple(BrinBuildState *state); static void union_tuples(BrinDesc *bdesc, BrinMemTuple *a, BrinTuple *b); static void brin_vacuum_scan(Relation idxrel, BufferAccessStrategy strategy); - +static bool add_values_to_range(Relation idxRel, BrinDesc *bdesc, + BrinMemTuple *dtup, Datum *values, bool *nulls); +static bool check_null_keys(BrinValues *bval, ScanKey *nullkeys, int nnullkeys); /* * BRIN handler function: return IndexAmRoutine with access method parameters @@ -179,7 +182,6 @@ brininsert(Relation idxRel, Datum *values, bool *nulls, OffsetNumber off; BrinTuple *brtup; BrinMemTuple *dtup; - int keyno; CHECK_FOR_INTERRUPTS(); @@ -243,31 +245,7 @@ brininsert(Relation idxRel, Datum *values, bool *nulls, dtup = brin_deform_tuple(bdesc, brtup, NULL); - /* - * Compare the key values of the new tuple to the stored index values; - * our deformed tuple will get updated if the new tuple doesn't fit - * the original range (note this means we can't break out of the loop - * early). Make a note of whether this happens, so that we know to - * insert the modified tuple later. - */ - for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++) - { - Datum result; - BrinValues *bval; - FmgrInfo *addValue; - - bval = &dtup->bt_columns[keyno]; - addValue = index_getprocinfo(idxRel, keyno + 1, - BRIN_PROCNUM_ADDVALUE); - result = FunctionCall4Coll(addValue, - idxRel->rd_indcollation[keyno], - PointerGetDatum(bdesc), - PointerGetDatum(bval), - values[keyno], - nulls[keyno]); - /* if that returned true, we need to insert the updated tuple */ - need_insert |= DatumGetBool(result); - } + need_insert = add_values_to_range(idxRel, bdesc, dtup, values, nulls); if (!need_insert) { @@ -390,8 +368,10 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm) BrinMemTuple *dtup; BrinTuple *btup = NULL; Size btupsz = 0; - ScanKey **keys; - int *nkeys; + ScanKey **keys, + **nullkeys; + int *nkeys, + *nnullkeys; int keyno; opaque = (BrinOpaque *) scan->opaque; @@ -416,10 +396,13 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm) /* * Make room for per-attribute lists of scan keys that we'll pass to the - * consistent support procedure. + * consistent support procedure. We keep null and regular keys separate, + * so that we can easily pass regular keys to the consistent function. */ keys = palloc0(sizeof(ScanKey *) * bdesc->bd_tupdesc->natts); + nullkeys = palloc0(sizeof(ScanKey *) * bdesc->bd_tupdesc->natts); nkeys = palloc0(sizeof(int) * bdesc->bd_tupdesc->natts); + nnullkeys = palloc0(sizeof(int) * bdesc->bd_tupdesc->natts); /* * Preprocess the scan keys - split them into per-attribute arrays. @@ -440,23 +423,24 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm) TupleDescAttr(bdesc->bd_tupdesc, keyattno - 1)->attcollation)); - /* First time we see this index attribute, so init as needed. */ - if (!keys[keyattno-1]) + /* + * First time we see this index attribute, so init as needed. + * + * This is a bit of an overkill - we don't know how many scan + * keys are there for a given attribute, so we simply allocate + * the largest number possible (as if all scan keys belonged to + * the same attribute). This may waste a bit of memory, but we + * only expect small number of scan keys in general, so this + * should be negligible, and it's probably cheaper than having + * to repalloc repeatedly. + */ + if (consistentFn[keyattno - 1].fn_oid == InvalidOid) { FmgrInfo *tmp; - /* - * This is a bit of an overkill - we don't know how many - * scan keys are there for this attribute, so we simply - * allocate the largest number possible. This may waste - * a bit of memory, but we only expect small number of - * scan keys in general, so this should be negligible, - * and it's cheaper than having to repalloc repeatedly. - */ - keys[keyattno - 1] = palloc0(sizeof(ScanKey) * scan->numberOfKeys); - - /* First time this column, so look up consistent function */ - Assert(consistentFn[keyattno - 1].fn_oid == InvalidOid); + /* No key/null arrays for this attribute. */ + Assert((keys[keyattno - 1] == NULL) && (nkeys[keyattno - 1] == 0)); + Assert((nullkeys[keyattno - 1] == NULL) && (nnullkeys[keyattno - 1] == 0)); tmp = index_getprocinfo(idxRel, keyattno, BRIN_PROCNUM_CONSISTENT); @@ -464,9 +448,23 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm) CurrentMemoryContext); } - /* Add key to the per-attribute array. */ - keys[keyattno - 1][nkeys[keyattno - 1]] = key; - nkeys[keyattno - 1]++; + /* Add key to the proper per-attribute array. */ + if (key->sk_flags & SK_ISNULL) + { + if (!nullkeys[keyattno - 1]) + nullkeys[keyattno - 1] = palloc0(sizeof(ScanKey) * scan->numberOfKeys); + + nullkeys[keyattno - 1][nnullkeys[keyattno - 1]] = key; + nnullkeys[keyattno - 1]++; + } + else + { + if (!keys[keyattno - 1]) + keys[keyattno - 1] = palloc0(sizeof(ScanKey) * scan->numberOfKeys); + + keys[keyattno - 1][nkeys[keyattno - 1]] = key; + nkeys[keyattno - 1]++; + } } /* allocate an initial in-memory tuple, out of the per-range memcxt */ @@ -544,15 +542,57 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm) BrinValues *bval; Datum add; - /* skip attributes without any san keys */ - if (nkeys[attno - 1] == 0) + /* + * skip attributes without any scan keys (both regular + * and IS [NOT] NULL) + */ + if (nkeys[attno - 1] == 0 && nnullkeys[attno - 1] == 0) continue; bval = &dtup->bt_columns[attno - 1]; + /* + * First check if there are any IS [NOT] NULL scan keys, + * and if we're violating them. In that case we can + * terminate early, without invoking the support function. + * + * As there may be more keys, we can only detemine mismatch + * within this loop. + */ + if (bdesc->bd_info[attno - 1]->oi_regular_nulls && + !check_null_keys(bval, nullkeys[attno - 1], + nnullkeys[attno - 1])) + { + /* + * If any of the IS [NOT] NULL keys failed, the page + * range as a whole can't pass. So terminate the loop. + */ + addrange = false; + break; + } + + /* + * So either there are no IS [NOT] NULL keys, or all passed. + * If there are no regular scan keys, we're done - the page + * range matches. If there are regular keys, but the page + * range is marked as 'all nulls' it can't possibly pass + * (we're assuming the operators are strict). + */ + + /* No regular scan keys - page range as a whole passes. */ + if (!nkeys[attno - 1]) + continue; + Assert((nkeys[attno - 1] > 0) && (nkeys[attno - 1] <= scan->numberOfKeys)); + /* If it is all nulls, it cannot possibly be consistent. */ + if (bval->bv_allnulls) + { + addrange = false; + break; + } + /* * Check whether the scan key is consistent with the page * range values; if so, have the pages in the range added @@ -695,7 +735,6 @@ brinbuildCallback(Relation index, { BrinBuildState *state = (BrinBuildState *) brstate; BlockNumber thisblock; - int i; thisblock = ItemPointerGetBlockNumber(tid); @@ -724,25 +763,8 @@ brinbuildCallback(Relation index, } /* Accumulate the current tuple into the running state */ - for (i = 0; i < state->bs_bdesc->bd_tupdesc->natts; i++) - { - FmgrInfo *addValue; - BrinValues *col; - Form_pg_attribute attr = TupleDescAttr(state->bs_bdesc->bd_tupdesc, i); - - col = &state->bs_dtuple->bt_columns[i]; - addValue = index_getprocinfo(index, i + 1, - BRIN_PROCNUM_ADDVALUE); - - /* - * Update dtuple state, if and as necessary. - */ - FunctionCall4Coll(addValue, - attr->attcollation, - PointerGetDatum(state->bs_bdesc), - PointerGetDatum(col), - values[i], isnull[i]); - } + (void) add_values_to_range(index, state->bs_bdesc, state->bs_dtuple, + values, isnull); } /* @@ -1521,6 +1543,39 @@ union_tuples(BrinDesc *bdesc, BrinMemTuple *a, BrinTuple *b) FmgrInfo *unionFn; BrinValues *col_a = &a->bt_columns[keyno]; BrinValues *col_b = &db->bt_columns[keyno]; + BrinOpcInfo *opcinfo = bdesc->bd_info[keyno]; + + if (opcinfo->oi_regular_nulls) + { + /* Adjust "hasnulls". */ + if (!col_a->bv_hasnulls && col_b->bv_hasnulls) + col_a->bv_hasnulls = true; + + /* If there are no values in B, there's nothing left to do. */ + if (col_b->bv_allnulls) + continue; + + /* + * Adjust "allnulls". If A doesn't have values, just copy the + * values from B into A, and we're done. We cannot run the + * operators in this case, because values in A might contain + * garbage. Note we already established that B contains values. + */ + if (col_a->bv_allnulls) + { + int i; + + col_a->bv_allnulls = false; + + for (i = 0; i < opcinfo->oi_nstored; i++) + col_a->bv_values[i] = + datumCopy(col_b->bv_values[i], + opcinfo->oi_typcache[i]->typbyval, + opcinfo->oi_typcache[i]->typlen); + + continue; + } + } unionFn = index_getprocinfo(bdesc->bd_index, keyno + 1, BRIN_PROCNUM_UNION); @@ -1574,3 +1629,103 @@ brin_vacuum_scan(Relation idxrel, BufferAccessStrategy strategy) */ FreeSpaceMapVacuum(idxrel); } + +static bool +add_values_to_range(Relation idxRel, BrinDesc *bdesc, BrinMemTuple *dtup, + Datum *values, bool *nulls) +{ + int keyno; + bool modified = false; + + /* + * Compare the key values of the new tuple to the stored index values; + * our deformed tuple will get updated if the new tuple doesn't fit + * the original range (note this means we can't break out of the loop + * early). Make a note of whether this happens, so that we know to + * insert the modified tuple later. + */ + for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++) + { + Datum result; + BrinValues *bval; + FmgrInfo *addValue; + + bval = &dtup->bt_columns[keyno]; + + if (bdesc->bd_info[keyno]->oi_regular_nulls && nulls[keyno]) + { + /* + * If the new value is null, we record that we saw it if it's + * the first one; otherwise, there's nothing to do. + */ + if (!bval->bv_hasnulls) + { + bval->bv_hasnulls = true; + modified = true; + } + + continue; + } + + addValue = index_getprocinfo(idxRel, keyno + 1, + BRIN_PROCNUM_ADDVALUE); + result = FunctionCall4Coll(addValue, + idxRel->rd_indcollation[keyno], + PointerGetDatum(bdesc), + PointerGetDatum(bval), + values[keyno], + nulls[keyno]); + /* if that returned true, we need to insert the updated tuple */ + modified |= DatumGetBool(result); + } + + return modified; +} + +static bool +check_null_keys(BrinValues *bval, ScanKey *nullkeys, int nnullkeys) +{ + int keyno; + + /* + * First check if there are any IS [NOT] NULL scan keys, and if we're + * violating them. + */ + for (keyno = 0; keyno < nnullkeys; keyno++) + { + ScanKey key = nullkeys[keyno]; + + Assert(key->sk_attno == bval->bv_attno); + + /* Handle only IS NULL/IS NOT NULL tests */ + if (!(key->sk_flags & SK_ISNULL)) + continue; + + if (key->sk_flags & SK_SEARCHNULL) + { + /* IS NULL scan key, but range has no NULLs */ + if (!bval->bv_allnulls && !bval->bv_hasnulls) + return false; + } + else if (key->sk_flags & SK_SEARCHNOTNULL) + { + /* + * For IS NOT NULL, we can only skip ranges that are known to + * have only nulls. + */ + if (bval->bv_allnulls) + return false; + } + else + { + /* + * Neither IS NULL nor IS NOT NULL was used; assume all indexable + * operators are strict and thus return false with NULL value in + * the scan key. + */ + return false; + } + } + + return true; +} diff --git a/src/backend/access/brin/brin_inclusion.c b/src/backend/access/brin/brin_inclusion.c index 215bc794d3..f4730be3b9 100644 --- a/src/backend/access/brin/brin_inclusion.c +++ b/src/backend/access/brin/brin_inclusion.c @@ -110,6 +110,7 @@ brin_inclusion_opcinfo(PG_FUNCTION_ARGS) */ result = palloc0(MAXALIGN(SizeofBrinOpcInfo(3)) + sizeof(InclusionOpaque)); result->oi_nstored = 3; + result->oi_regular_nulls = true; result->oi_opaque = (InclusionOpaque *) MAXALIGN((char *) result + SizeofBrinOpcInfo(3)); @@ -141,7 +142,7 @@ brin_inclusion_add_value(PG_FUNCTION_ARGS) BrinDesc *bdesc = (BrinDesc *) PG_GETARG_POINTER(0); BrinValues *column = (BrinValues *) PG_GETARG_POINTER(1); Datum newval = PG_GETARG_DATUM(2); - bool isnull = PG_GETARG_BOOL(3); + bool isnull PG_USED_FOR_ASSERTS_ONLY = PG_GETARG_BOOL(3); Oid colloid = PG_GET_COLLATION(); FmgrInfo *finfo; Datum result; @@ -149,18 +150,7 @@ brin_inclusion_add_value(PG_FUNCTION_ARGS) AttrNumber attno; Form_pg_attribute attr; - /* - * If the new value is null, we record that we saw it if it's the first - * one; otherwise, there's nothing to do. - */ - if (isnull) - { - if (column->bv_hasnulls) - PG_RETURN_BOOL(false); - - column->bv_hasnulls = true; - PG_RETURN_BOOL(true); - } + Assert(!isnull); attno = column->bv_attno; attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1); @@ -264,63 +254,6 @@ brin_inclusion_consistent(PG_FUNCTION_ARGS) int nkeys = PG_GETARG_INT32(3); Oid colloid = PG_GET_COLLATION(); int keyno; - bool regular_keys = false; - - /* - * First check if there are any IS NULL scan keys, and if we're - * violating them. In that case we can terminate early, without - * inspecting the ranges. - */ - for (keyno = 0; keyno < nkeys; keyno++) - { - ScanKey key = keys[keyno]; - - Assert(key->sk_attno == column->bv_attno); - - /* handle IS NULL/IS NOT NULL tests */ - if (key->sk_flags & SK_ISNULL) - { - if (key->sk_flags & SK_SEARCHNULL) - { - if (column->bv_allnulls || column->bv_hasnulls) - continue; /* this key is fine, continue */ - - PG_RETURN_BOOL(false); - } - - /* - * For IS NOT NULL, we can only skip ranges that are known to have - * only nulls. - */ - if (key->sk_flags & SK_SEARCHNOTNULL) - { - if (column->bv_allnulls) - PG_RETURN_BOOL(false); - - continue; - } - - /* - * Neither IS NULL nor IS NOT NULL was used; assume all indexable - * operators are strict and return false. - */ - PG_RETURN_BOOL(false); - } - else - /* note we have regular (non-NULL) scan keys */ - regular_keys = true; - } - - /* - * If the page range is all nulls, it cannot possibly be consistent if - * there are some regular scan keys. - */ - if (column->bv_allnulls && regular_keys) - PG_RETURN_BOOL(false); - - /* If there are no regular keys, the page range is considered consistent. */ - if (!regular_keys) - PG_RETURN_BOOL(true); /* It has to be checked, if it contains elements that are not mergeable. */ if (DatumGetBool(column->bv_values[INCLUSION_UNMERGEABLE])) @@ -331,9 +264,8 @@ brin_inclusion_consistent(PG_FUNCTION_ARGS) { ScanKey key = keys[keyno]; - /* ignore IS NULL/IS NOT NULL tests handled above */ - if (key->sk_flags & SK_ISNULL) - continue; + /* NULL keys are handled and filtered-out in bringetbitmap */ + Assert(!(key->sk_flags & SK_ISNULL)); /* * When there are multiple scan keys, failure to meet the @@ -572,37 +504,11 @@ brin_inclusion_union(PG_FUNCTION_ARGS) Datum result; Assert(col_a->bv_attno == col_b->bv_attno); - - /* Adjust "hasnulls". */ - if (!col_a->bv_hasnulls && col_b->bv_hasnulls) - col_a->bv_hasnulls = true; - - /* If there are no values in B, there's nothing left to do. */ - if (col_b->bv_allnulls) - PG_RETURN_VOID(); + Assert(!col_a->bv_allnulls && !col_b->bv_allnulls); attno = col_a->bv_attno; attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1); - /* - * Adjust "allnulls". If A doesn't have values, just copy the values from - * B into A, and we're done. We cannot run the operators in this case, - * because values in A might contain garbage. Note we already established - * that B contains values. - */ - if (col_a->bv_allnulls) - { - col_a->bv_allnulls = false; - col_a->bv_values[INCLUSION_UNION] = - datumCopy(col_b->bv_values[INCLUSION_UNION], - attr->attbyval, attr->attlen); - col_a->bv_values[INCLUSION_UNMERGEABLE] = - col_b->bv_values[INCLUSION_UNMERGEABLE]; - col_a->bv_values[INCLUSION_CONTAINS_EMPTY] = - col_b->bv_values[INCLUSION_CONTAINS_EMPTY]; - PG_RETURN_VOID(); - } - /* If B includes empty elements, mark A similarly, if needed. */ if (!DatumGetBool(col_a->bv_values[INCLUSION_CONTAINS_EMPTY]) && DatumGetBool(col_b->bv_values[INCLUSION_CONTAINS_EMPTY])) diff --git a/src/backend/access/brin/brin_minmax.c b/src/backend/access/brin/brin_minmax.c index 12878ff3a0..6c8852d404 100644 --- a/src/backend/access/brin/brin_minmax.c +++ b/src/backend/access/brin/brin_minmax.c @@ -48,6 +48,7 @@ brin_minmax_opcinfo(PG_FUNCTION_ARGS) result = palloc0(MAXALIGN(SizeofBrinOpcInfo(2)) + sizeof(MinmaxOpaque)); result->oi_nstored = 2; + result->oi_regular_nulls = true; result->oi_opaque = (MinmaxOpaque *) MAXALIGN((char *) result + SizeofBrinOpcInfo(2)); result->oi_typcache[0] = result->oi_typcache[1] = @@ -69,7 +70,7 @@ brin_minmax_add_value(PG_FUNCTION_ARGS) BrinDesc *bdesc = (BrinDesc *) PG_GETARG_POINTER(0); BrinValues *column = (BrinValues *) PG_GETARG_POINTER(1); Datum newval = PG_GETARG_DATUM(2); - bool isnull = PG_GETARG_DATUM(3); + bool isnull PG_USED_FOR_ASSERTS_ONLY = PG_GETARG_DATUM(3); Oid colloid = PG_GET_COLLATION(); FmgrInfo *cmpFn; Datum compar; @@ -77,18 +78,7 @@ brin_minmax_add_value(PG_FUNCTION_ARGS) Form_pg_attribute attr; AttrNumber attno; - /* - * If the new value is null, we record that we saw it if it's the first - * one; otherwise, there's nothing to do. - */ - if (isnull) - { - if (column->bv_hasnulls) - PG_RETURN_BOOL(false); - - column->bv_hasnulls = true; - PG_RETURN_BOOL(true); - } + Assert(!isnull); attno = column->bv_attno; attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1); @@ -152,72 +142,14 @@ brin_minmax_consistent(PG_FUNCTION_ARGS) int nkeys = PG_GETARG_INT32(3); Oid colloid = PG_GET_COLLATION(); int keyno; - bool regular_keys = false; - - /* - * First check if there are any IS NULL scan keys, and if we're - * violating them. In that case we can terminate early, without - * inspecting the ranges. - */ - for (keyno = 0; keyno < nkeys; keyno++) - { - ScanKey key = keys[keyno]; - - Assert(key->sk_attno == column->bv_attno); - - /* handle IS NULL/IS NOT NULL tests */ - if (key->sk_flags & SK_ISNULL) - { - if (key->sk_flags & SK_SEARCHNULL) - { - if (column->bv_allnulls || column->bv_hasnulls) - continue; /* this key is fine, continue */ - - PG_RETURN_BOOL(false); - } - - /* - * For IS NOT NULL, we can only skip ranges that are known to have - * only nulls. - */ - if (key->sk_flags & SK_SEARCHNOTNULL) - { - if (column->bv_allnulls) - PG_RETURN_BOOL(false); - - continue; - } - - /* - * Neither IS NULL nor IS NOT NULL was used; assume all indexable - * operators are strict and return false. - */ - PG_RETURN_BOOL(false); - } - else - /* note we have regular (non-NULL) scan keys */ - regular_keys = true; - } - - /* - * If the page range is all nulls, it cannot possibly be consistent if - * there are some regular scan keys. - */ - if (column->bv_allnulls && regular_keys) - PG_RETURN_BOOL(false); - - /* If there are no regular keys, the page range is considered consistent. */ - if (!regular_keys) - PG_RETURN_BOOL(true); /* Check that the range is consistent with all scan keys. */ for (keyno = 0; keyno < nkeys; keyno++) { ScanKey key = keys[keyno]; - /* ignore IS NULL/IS NOT NULL tests handled above */ - if (key->sk_flags & SK_ISNULL) - continue; + /* NULL keys are handled and filtered-out in bringetbitmap */ + Assert(!(key->sk_flags & SK_ISNULL)); /* * When there are multiple scan keys, failure to meet the @@ -308,34 +240,11 @@ brin_minmax_union(PG_FUNCTION_ARGS) bool needsadj; Assert(col_a->bv_attno == col_b->bv_attno); - - /* Adjust "hasnulls" */ - if (!col_a->bv_hasnulls && col_b->bv_hasnulls) - col_a->bv_hasnulls = true; - - /* If there are no values in B, there's nothing left to do */ - if (col_b->bv_allnulls) - PG_RETURN_VOID(); + Assert(!col_a->bv_allnulls && !col_b->bv_allnulls); attno = col_a->bv_attno; attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1); - /* - * Adjust "allnulls". If A doesn't have values, just copy the values from - * B into A, and we're done. We cannot run the operators in this case, - * because values in A might contain garbage. Note we already established - * that B contains values. - */ - if (col_a->bv_allnulls) - { - col_a->bv_allnulls = false; - col_a->bv_values[0] = datumCopy(col_b->bv_values[0], - attr->attbyval, attr->attlen); - col_a->bv_values[1] = datumCopy(col_b->bv_values[1], - attr->attbyval, attr->attlen); - PG_RETURN_VOID(); - } - /* Adjust minimum, if B's min is less than A's min */ finfo = minmax_get_strategy_procinfo(bdesc, attno, attr->atttypid, BTLessStrategyNumber); diff --git a/src/include/access/brin_internal.h b/src/include/access/brin_internal.h index 78c89a6961..79440ebe7b 100644 --- a/src/include/access/brin_internal.h +++ b/src/include/access/brin_internal.h @@ -27,6 +27,9 @@ typedef struct BrinOpcInfo /* Number of columns stored in an index column of this opclass */ uint16 oi_nstored; + /* Regular processing of NULLs in BrinValues? */ + bool oi_regular_nulls; + /* Opaque pointer for the opclass' private use */ void *oi_opaque; -- 2.26.2 --------------0E72B707603BED22B4040825 Content-Type: text/x-patch; charset=UTF-8; name="0001-Pass-all-scan-keys-to-BRIN-consistent-funct-20210211.patch" Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename*0="0001-Pass-all-scan-keys-to-BRIN-consistent-funct-20210211.pa"; filename*1="tch" ^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: refactoring basebackup.c @ 2022-01-18 14:42 Jeevan Ladhe <[email protected]> 0 siblings, 2 replies; 4+ messages in thread From: Jeevan Ladhe @ 2022-01-18 14:42 UTC (permalink / raw) To: tushar <[email protected]>; +Cc: Robert Haas <[email protected]>; Dmitry Dolgov <[email protected]>; Mark Dilger <[email protected]>; pgsql-hackers Hi, Similar to LZ4 server-side compression, I have also tried to add a ZSTD server-side compression in the attached patch. I have done some initial testing and things seem to be working. Example run: pg_basebackup -t server:/tmp/data_zstd -Xnone --server-compression=zstd The patch surely needs some grooming, but I am expecting some initial review, specially in the area where we are trying to close the zstd stream in bbsink_zstd_end_archive(). We need to tell the zstd library to end the compression by calling ZSTD_compressStream2() thereby sending a ZSTD_e_end flag. But, this also needs some input string, which per example[1] line # 686, I have taken as an empty ZSTD_inBuffer. Thanks, Tushar for testing the LZ4 patch. I have added the LZ4 option in the pg_basebackup help now. Note: Before applying these patches please apply Robert's v10 version of patches 0002, 0003, and 0004. [1] https://fuchsia.googlesource.com/fuchsia/+/refs/heads/main/zircon/tools/zbi/zbi.cc Regards, Jeevan Ladhe On Wed, Jan 5, 2022 at 10:24 PM tushar <[email protected]> wrote: > > > On Tue, Dec 28, 2021 at 1:12 PM Jeevan Ladhe < > [email protected]> wrote: > >> Hi Tushar, >> >> You need to apply Robert's v10 version patches 0002, 0003 and 0004, >> before applying the lz4 patch(v8 version). >> Please let me know if you still face any issues. >> > > Thanks, Jeevan. > I tested —server-compression option using different other options of > pg_basebackup, also checked -t/—server-compression from pg_basebackup of > v15 will > throw an error if the server version is v14 or below. Things are looking > good to me. > Two open issues - > 1)lz4 value is missing for --server-compression in pg_basebackup --help > 2)Error messages need to improve if using -t server with -z/-Z > > regards, > Attachments: [application/octet-stream] v9-0001-Add-a-LZ4-compression-method-for-server-side-compres.patch (15.2K, ../../CAOgcT0NBurXSm+ZeF=PiBcAm9mywHFo6Jc9G+F7JFU8O9sndBQ@mail.gmail.com/3-v9-0001-Add-a-LZ4-compression-method-for-server-side-compres.patch) download | inline diff: From 80aa8cb9ecbeb3303562129ab13a772aa29dd1b4 Mon Sep 17 00:00:00 2001 From: Jeevan Ladhe <[email protected]> Date: Tue, 18 Jan 2022 19:46:36 +0530 Subject: [PATCH 1/2] Add a LZ4 compression method for server side compression. Adds LZ4 server side compression option --server-compression=lz4 Add documentation for LZ4. Add pg_basebackup help for ZSTD option Example: pg_basebackup -t server:/tmp/data_lz4 -Xnone --server-compression=lz4 --- doc/src/sgml/ref/pg_basebackup.sgml | 49 +++- src/backend/replication/Makefile | 1 + src/backend/replication/basebackup.c | 7 +- src/backend/replication/basebackup_lz4.c | 285 ++++++++++++++++++++++ src/bin/pg_basebackup/pg_basebackup.c | 2 +- src/include/replication/basebackup_sink.h | 1 + 6 files changed, 335 insertions(+), 10 deletions(-) create mode 100644 src/backend/replication/basebackup_lz4.c diff --git a/doc/src/sgml/ref/pg_basebackup.sgml b/doc/src/sgml/ref/pg_basebackup.sgml index 9ce8b8d89d..44395a749b 100644 --- a/doc/src/sgml/ref/pg_basebackup.sgml +++ b/doc/src/sgml/ref/pg_basebackup.sgml @@ -230,14 +230,7 @@ PostgreSQL documentation <para> Allows the tar files generated for each tablespace to be compressed - on the server, before they are sent to the client. The default value - is <literal>none</literal>, which performs no compression. If set - to <literal>gzip</literal>, compression is performed using gzip and - the suffix <filename>.gz</filename> will automatically be added to - compressed files. A numeric digit between 1 and 9 can be added to - specify the compression level; for instance, <literal>gzip9</literal> - will provide the maximum compression that the <literal>gzip</literal> - algorithm can provide. + on the server, before they are sent to the client. </para> <para> Since the write-ahead logs are fetched via a separate client @@ -245,7 +238,47 @@ PostgreSQL documentation the <literal>--gzip</literal> and <literal>--compress</literal> options. </para> + <para> + The following <replaceable>target</replaceable> algorithms for + server-compression are supported: + + <variablelist> + <varlistentry> + <term><literal>none</literal></term> + <listitem> + <para> + Perform no compression. This is the default value. + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term><literal>gzip</literal></term> + <listitem> + <para> + Compression is performed using <literal>gzip</literal> and the + suffix <filename>.gz </filename> will automatically be added to + compressed files. A numeric digit between 1 and 9 can be added to + specify the compression level; for instance, <literal>gzip9 + </literal> will provide the maximum compression that the + <literal>gzip</literal> algorithm can provide. + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term><literal>lz4</literal></term> + <listitem> + <para> + Compression is performed using <literal>lz4</literal> and the + suffix <filename>.lz4</filename> will automatically be added to + compressed files. + </para> + </listitem> + </varlistentry> + </variablelist> + </para> </listitem> </varlistentry> diff --git a/src/backend/replication/Makefile b/src/backend/replication/Makefile index 8ec60ded76..74043ff331 100644 --- a/src/backend/replication/Makefile +++ b/src/backend/replication/Makefile @@ -19,6 +19,7 @@ OBJS = \ basebackup.o \ basebackup_copy.o \ basebackup_gzip.o \ + basebackup_lz4.o \ basebackup_progress.o \ basebackup_server.o \ basebackup_sink.o \ diff --git a/src/backend/replication/basebackup.c b/src/backend/replication/basebackup.c index 4bed0f18b7..9dea1c9bcc 100644 --- a/src/backend/replication/basebackup.c +++ b/src/backend/replication/basebackup.c @@ -64,7 +64,8 @@ typedef enum typedef enum { BACKUP_COMPRESSION_NONE, - BACKUP_COMPRESSION_GZIP + BACKUP_COMPRESSION_GZIP, + BACKUP_COMPRESSION_LZ4 } basebackup_compression_type; typedef struct @@ -909,6 +910,8 @@ parse_basebackup_options(List *options, basebackup_options *opt) opt->compression = BACKUP_COMPRESSION_GZIP; opt->compression_level = optval[4] - '0'; } + else if (strcmp(optval, "lz4") == 0) + opt->compression = BACKUP_COMPRESSION_LZ4; else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -1013,6 +1016,8 @@ SendBaseBackup(BaseBackupCmd *cmd) /* Set up server-side compression, if client requested it */ if (opt.compression == BACKUP_COMPRESSION_GZIP) sink = bbsink_gzip_new(sink, opt.compression_level); + else if (opt.compression == BACKUP_COMPRESSION_LZ4) + sink = bbsink_lz4_new(sink); /* Set up progress reporting. */ sink = bbsink_progress_new(sink, opt.progress); diff --git a/src/backend/replication/basebackup_lz4.c b/src/backend/replication/basebackup_lz4.c new file mode 100644 index 0000000000..0f49def813 --- /dev/null +++ b/src/backend/replication/basebackup_lz4.c @@ -0,0 +1,285 @@ +/*------------------------------------------------------------------------- + * + * basebackup_lz4.c + * Basebackup sink implementing lz4 compression. + * + * Portions Copyright (c) 2010-2020, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/backend/replication/basebackup_lz4.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#ifdef HAVE_LIBLZ4 +#include <lz4frame.h> +#endif +#include <unistd.h> + +#include "replication/basebackup_sink.h" + +#ifdef HAVE_LIBLZ4 + +typedef struct bbsink_lz4 +{ + /* Common information for all types of sink. */ + bbsink base; + + LZ4F_compressionContext_t ctx; + LZ4F_preferences_t prefs; + + /* Number of bytes staged in output buffer. */ + size_t bytes_written; +} bbsink_lz4; + +static void bbsink_lz4_begin_backup(bbsink *sink); +static void bbsink_lz4_begin_archive(bbsink *sink, const char *archive_name); +static void bbsink_lz4_archive_contents(bbsink *sink, size_t avail_in); +static void bbsink_lz4_manifest_contents(bbsink *sink, size_t len); +static void bbsink_lz4_end_archive(bbsink *sink); +static void bbsink_lz4_cleanup(bbsink *sink); + +const bbsink_ops bbsink_lz4_ops = { + .begin_backup = bbsink_lz4_begin_backup, + .begin_archive = bbsink_lz4_begin_archive, + .archive_contents = bbsink_lz4_archive_contents, + .end_archive = bbsink_lz4_end_archive, + .begin_manifest = bbsink_forward_begin_manifest, + .manifest_contents = bbsink_lz4_manifest_contents, + .end_manifest = bbsink_forward_end_manifest, + .end_backup = bbsink_forward_end_backup, + .cleanup = bbsink_lz4_cleanup +}; +#endif + +/* Create a new basebackup sink that performs lz4 compression. */ +bbsink * +bbsink_lz4_new(bbsink *next) +{ +#ifndef HAVE_LIBLZ4 + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("lz4 compression is not supported by this build"))); +#else + bbsink_lz4 *sink; + + Assert(next != NULL); + + sink = palloc0(sizeof(bbsink_lz4)); + *((const bbsink_ops **) &sink->base.bbs_ops) = &bbsink_lz4_ops; + sink->base.bbs_next = next; + + return &sink->base; +#endif +} + +#ifdef HAVE_LIBLZ4 + +/* + * Begin backup. + */ +static void +bbsink_lz4_begin_backup(bbsink *sink) +{ + bbsink_lz4 *mysink = (bbsink_lz4 *) sink; + size_t output_buffer_bound; + LZ4F_preferences_t *prefs = &mysink->prefs; + + /* Initialize compressor object. */ + memset(prefs, 0, sizeof(LZ4F_preferences_t)); + prefs->frameInfo.blockSizeID = LZ4F_max256KB; + + /* + * We need our own buffer, because we're going to pass different data to + * the next sink than what gets passed to us. + */ + mysink->base.bbs_buffer = palloc(mysink->base.bbs_buffer_length); + + /* + * Since LZ4F_compressUpdate() requires the output buffer of size equal or + * greater than that of LZ4F_compressBound(), make sure we have the next + * sink's bbs_buffer of length that can accommodate the compressed input + * buffer. + */ + output_buffer_bound = LZ4F_compressBound(mysink->base.bbs_buffer_length, + &mysink->prefs); + + /* + * The buffer length is expected to be a multiple of BLCKSZ, so round up. + */ + output_buffer_bound = output_buffer_bound + BLCKSZ - + (output_buffer_bound % BLCKSZ); + + bbsink_begin_backup(sink->bbs_next, sink->bbs_state, output_buffer_bound); +} + +/* + * Prepare to compress the next archive. + */ +static void +bbsink_lz4_begin_archive(bbsink *sink, const char *archive_name) +{ + bbsink_lz4 *mysink = (bbsink_lz4 *) sink; + char *lz4_archive_name; + LZ4F_errorCode_t ctxError; + size_t headerSize; + + ctxError = LZ4F_createCompressionContext(&mysink->ctx, LZ4F_VERSION); + if (LZ4F_isError(ctxError)) + elog(ERROR, "could not create lz4 compression context: %s", + LZ4F_getErrorName(ctxError)); + + /* First of all write the frame header to destination buffer. */ + headerSize = LZ4F_compressBegin(mysink->ctx, + mysink->base.bbs_next->bbs_buffer, + mysink->base.bbs_next->bbs_buffer_length, + &mysink->prefs); + + if (LZ4F_isError(headerSize)) + elog(ERROR, "could not write lz4 header: %s", + LZ4F_getErrorName(headerSize)); + + /* + * We need to write the compressed data after the header in the output + * buffer. So, make sure to update the notion of bytes written to output + * buffer. + */ + mysink->bytes_written += headerSize; + + /* Add ".lz4" to the archive name. */ + lz4_archive_name = psprintf("%s.lz4", archive_name); + Assert(sink->bbs_next != NULL); + bbsink_begin_archive(sink->bbs_next, lz4_archive_name); + pfree(lz4_archive_name); +} + +/* + * Compress the input data to the output buffer until we run out of input + * data. Each time the output buffer falls below the compression bound for + * the input buffer, invoke the archive_contents() method for then next sink. + * + * Note that since we're compressing the input, it may very commonly happen + * that we consume all the input data without filling the output buffer. In + * that case, the compressed representation of the current input data won't + * actually be sent to the next bbsink until a later call to this function, + * or perhaps even not until bbsink_lz4_end_archive() is invoked. + */ +static void +bbsink_lz4_archive_contents(bbsink *sink, size_t avail_in) +{ + bbsink_lz4 *mysink = (bbsink_lz4 *) sink; + size_t compressedSize; + size_t avail_in_bound; + + avail_in_bound = LZ4F_compressBound(avail_in, &mysink->prefs); + + /* + * If the number of available bytes has fallen below the value computed by + * LZ4F_compressBound(), ask the next sink to process the data so that we + * can empty the buffer. + */ + if ((mysink->base.bbs_next->bbs_buffer_length - mysink->bytes_written) <= + avail_in_bound) + { + bbsink_archive_contents(sink->bbs_next, mysink->bytes_written); + mysink->bytes_written = 0; + } + + /* + * Compress the input buffer and write it into the output buffer. + */ + compressedSize = LZ4F_compressUpdate(mysink->ctx, + mysink->base.bbs_next->bbs_buffer + mysink->bytes_written, + mysink->base.bbs_next->bbs_buffer_length - mysink->bytes_written, + (uint8 *) mysink->base.bbs_buffer, + avail_in, + NULL); + + if (LZ4F_isError(compressedSize)) + elog(ERROR, "could not compress data: %s", + LZ4F_getErrorName(compressedSize)); + + /* + * Update our notion of how many bytes we've written into output buffer. + */ + mysink->bytes_written += compressedSize; +} + +/* + * There might be some data inside lz4's internal buffers; we need to get + * that flushed out and also finalize the lz4 frame and then get that forwarded + * to the successor sink as archive content. + * + * Then we can end processing for this archive. + */ +static void +bbsink_lz4_end_archive(bbsink *sink) +{ + bbsink_lz4 *mysink = (bbsink_lz4 *) sink; + size_t compressedSize; + size_t lz4_footer_bound; + + lz4_footer_bound = LZ4F_compressBound(0, &mysink->prefs); + + Assert(mysink->base.bbs_next->bbs_buffer_length >= lz4_footer_bound); + + if ((mysink->base.bbs_next->bbs_buffer_length - mysink->bytes_written) <= + lz4_footer_bound) + { + bbsink_archive_contents(sink->bbs_next, mysink->bytes_written); + mysink->bytes_written = 0; + } + + compressedSize = LZ4F_compressEnd(mysink->ctx, + mysink->base.bbs_next->bbs_buffer + mysink->bytes_written, + mysink->base.bbs_next->bbs_buffer_length - mysink->bytes_written, + NULL); + + if (LZ4F_isError(compressedSize)) + elog(ERROR, "could not end lz4 compression: %s", + LZ4F_getErrorName(compressedSize)); + + /* Update our notion of how many bytes we've written. */ + mysink->bytes_written += compressedSize; + + /* Send whatever accumulated output bytes we have. */ + bbsink_archive_contents(sink->bbs_next, mysink->bytes_written); + mysink->bytes_written = 0; + + /* Release the resources. */ + LZ4F_freeCompressionContext(mysink->ctx); + mysink->ctx = NULL; + + /* Pass on the information that this archive has ended. */ + bbsink_forward_end_archive(sink); +} + +/* + * Manifest contents are not compressed, but we do need to copy them into + * the successor sink's buffer, because we have our own. + */ +static void +bbsink_lz4_manifest_contents(bbsink *sink, size_t len) +{ + memcpy(sink->bbs_next->bbs_buffer, sink->bbs_buffer, len); + bbsink_manifest_contents(sink->bbs_next, len); +} + +/* + * In case the backup fails, make sure we free the compression context by + * calling LZ4F_freeCompressionContext() if needed to avoid memory leak. + */ +static void +bbsink_lz4_cleanup(bbsink *sink) +{ + bbsink_lz4 *mysink = (bbsink_lz4 *) sink; + + if (mysink->ctx) + { + LZ4F_freeCompressionContext(mysink->ctx); + mysink->ctx = NULL; + } +} + +#endif diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c index 00fa55b982..d8da1cb2e9 100644 --- a/src/bin/pg_basebackup/pg_basebackup.c +++ b/src/bin/pg_basebackup/pg_basebackup.c @@ -374,7 +374,7 @@ usage(void) " (in kB/s, or use suffix \"k\" or \"M\")\n")); printf(_(" -R, --write-recovery-conf\n" " write configuration for replication\n")); - printf(_(" --server-compression=none|gzip|gzip[1-9]\n" + printf(_(" --server-compression=none|gzip|gzip[1-9]|lz4\n" " compress backup on server\n")); printf(_(" -T, --tablespace-mapping=OLDDIR=NEWDIR\n" " relocate tablespace in OLDDIR to NEWDIR\n")); diff --git a/src/include/replication/basebackup_sink.h b/src/include/replication/basebackup_sink.h index d3276b2487..964752ef5d 100644 --- a/src/include/replication/basebackup_sink.h +++ b/src/include/replication/basebackup_sink.h @@ -285,6 +285,7 @@ extern void bbsink_forward_cleanup(bbsink *sink); extern bbsink *bbsink_copystream_new(bool send_to_client); extern bbsink *bbsink_copytblspc_new(void); extern bbsink *bbsink_gzip_new(bbsink *next, int compresslevel); +extern bbsink *bbsink_lz4_new(bbsink *next); extern bbsink *bbsink_progress_new(bbsink *next, bool estimate_backup_size); extern bbsink *bbsink_server_new(bbsink *next, char *pathname); extern bbsink *bbsink_throttle_new(bbsink *next, uint32 maxrate); -- 2.25.1 [application/octet-stream] v9-0002-Add-a-ZSTD-compression-method-for-server-side-compre.patch (27.1K, ../../CAOgcT0NBurXSm+ZeF=PiBcAm9mywHFo6Jc9G+F7JFU8O9sndBQ@mail.gmail.com/4-v9-0002-Add-a-ZSTD-compression-method-for-server-side-compre.patch) download | inline diff: From 5b06f5b1039b51f0847e7c310c04a61308b3c7b9 Mon Sep 17 00:00:00 2001 From: Jeevan Ladhe <[email protected]> Date: Tue, 18 Jan 2022 19:48:33 +0530 Subject: [PATCH 2/2] Add a ZSTD compression method for server side compression. This patch introduces --server-compression=zstd option. Add config option --with-zstd. Add documentation for ZSTD option Add pg_basebackup help for ZSTD option Example: pg_basebackup -t server:/tmp/data_zstd -Xnone --server-compression=zstd --- configure | 240 ++++++++++++++++++- configure.ac | 32 +++ doc/src/sgml/ref/pg_basebackup.sgml | 11 + src/backend/replication/Makefile | 1 + src/backend/replication/basebackup.c | 7 +- src/backend/replication/basebackup_zstd.c | 267 ++++++++++++++++++++++ src/bin/pg_basebackup/pg_basebackup.c | 2 +- src/include/pg_config.h.in | 6 + src/include/replication/basebackup_sink.h | 1 + 9 files changed, 559 insertions(+), 8 deletions(-) create mode 100644 src/backend/replication/basebackup_zstd.c diff --git a/configure b/configure index 9c856cb1d5..a532e85e66 100755 --- a/configure +++ b/configure @@ -699,6 +699,9 @@ with_gnu_ld LD LDFLAGS_SL LDFLAGS_EX +ZSTD_LIBS +ZSTD_CFLAGS +with_zstd LZ4_LIBS LZ4_CFLAGS with_lz4 @@ -800,6 +803,7 @@ infodir docdir oldincludedir includedir +runstatedir localstatedir sharedstatedir sysconfdir @@ -868,6 +872,7 @@ with_libxslt with_system_tzdata with_zlib with_lz4 +with_zstd with_gnu_ld with_ssl with_openssl @@ -897,6 +902,8 @@ XML2_CFLAGS XML2_LIBS LZ4_CFLAGS LZ4_LIBS +ZSTD_CFLAGS +ZSTD_LIBS LDFLAGS_EX LDFLAGS_SL PERL @@ -941,6 +948,7 @@ datadir='${datarootdir}' sysconfdir='${prefix}/etc' sharedstatedir='${prefix}/com' localstatedir='${prefix}/var' +runstatedir='${localstatedir}/run' includedir='${prefix}/include' oldincludedir='/usr/include' docdir='${datarootdir}/doc/${PACKAGE_TARNAME}' @@ -1193,6 +1201,15 @@ do | -silent | --silent | --silen | --sile | --sil) silent=yes ;; + -runstatedir | --runstatedir | --runstatedi | --runstated \ + | --runstate | --runstat | --runsta | --runst | --runs \ + | --run | --ru | --r) + ac_prev=runstatedir ;; + -runstatedir=* | --runstatedir=* | --runstatedi=* | --runstated=* \ + | --runstate=* | --runstat=* | --runsta=* | --runst=* | --runs=* \ + | --run=* | --ru=* | --r=*) + runstatedir=$ac_optarg ;; + -sbindir | --sbindir | --sbindi | --sbind | --sbin | --sbi | --sb) ac_prev=sbindir ;; -sbindir=* | --sbindir=* | --sbindi=* | --sbind=* | --sbin=* \ @@ -1330,7 +1347,7 @@ fi for ac_var in exec_prefix prefix bindir sbindir libexecdir datarootdir \ datadir sysconfdir sharedstatedir localstatedir includedir \ oldincludedir docdir infodir htmldir dvidir pdfdir psdir \ - libdir localedir mandir + libdir localedir mandir runstatedir do eval ac_val=\$$ac_var # Remove trailing slashes. @@ -1483,6 +1500,7 @@ Fine tuning of the installation directories: --sysconfdir=DIR read-only single-machine data [PREFIX/etc] --sharedstatedir=DIR modifiable architecture-independent data [PREFIX/com] --localstatedir=DIR modifiable single-machine data [PREFIX/var] + --runstatedir=DIR modifiable per-process data [LOCALSTATEDIR/run] --libdir=DIR object code libraries [EPREFIX/lib] --includedir=DIR C header files [PREFIX/include] --oldincludedir=DIR C header files for non-gcc [/usr/include] @@ -1576,6 +1594,7 @@ Optional Packages: use system time zone data in DIR --without-zlib do not use Zlib --with-lz4 build with LZ4 support + --with-zstd build with ZSTD support --with-gnu-ld assume the C compiler uses GNU ld [default=no] --with-ssl=LIB use LIB for SSL/TLS support (openssl) --with-openssl obsolete spelling of --with-ssl=openssl @@ -1605,6 +1624,8 @@ Some influential environment variables: XML2_LIBS linker flags for XML2, overriding pkg-config LZ4_CFLAGS C compiler flags for LZ4, overriding pkg-config LZ4_LIBS linker flags for LZ4, overriding pkg-config + ZSTD_CFLAGS C compiler flags for ZSTD, overriding pkg-config + ZSTD_LIBS linker flags for ZSTD, overriding pkg-config LDFLAGS_EX extra linker flags for linking executables only LDFLAGS_SL extra linker flags for linking shared libraries only PERL Perl program @@ -9033,6 +9054,146 @@ fi done fi +# +# ZSTD +# +{ $as_echo "$as_me:${as_lineno-$LINENO}: checking whether to build with ZSTD support" >&5 +$as_echo_n "checking whether to build with ZSTD support... " >&6; } + + + +# Check whether --with-zstd was given. +if test "${with_zstd+set}" = set; then : + withval=$with_zstd; + case $withval in + yes) + +$as_echo "#define USE_ZSTD 1" >>confdefs.h + + ;; + no) + : + ;; + *) + as_fn_error $? "no argument expected for --with-zstd option" "$LINENO" 5 + ;; + esac + +else + with_zstd=no + +fi + + +{ $as_echo "$as_me:${as_lineno-$LINENO}: result: $with_zstd" >&5 +$as_echo "$with_zstd" >&6; } + + +if test "$with_zstd" = yes; then + +pkg_failed=no +{ $as_echo "$as_me:${as_lineno-$LINENO}: checking for libzstd" >&5 +$as_echo_n "checking for libzstd... " >&6; } + +if test -n "$ZSTD_CFLAGS"; then + pkg_cv_ZSTD_CFLAGS="$ZSTD_CFLAGS" + elif test -n "$PKG_CONFIG"; then + if test -n "$PKG_CONFIG" && \ + { { $as_echo "$as_me:${as_lineno-$LINENO}: \$PKG_CONFIG --exists --print-errors \"libzstd\""; } >&5 + ($PKG_CONFIG --exists --print-errors "libzstd") 2>&5 + ac_status=$? + $as_echo "$as_me:${as_lineno-$LINENO}: \$? = $ac_status" >&5 + test $ac_status = 0; }; then + pkg_cv_ZSTD_CFLAGS=`$PKG_CONFIG --cflags "libzstd" 2>/dev/null` + test "x$?" != "x0" && pkg_failed=yes +else + pkg_failed=yes +fi + else + pkg_failed=untried +fi +if test -n "$ZSTD_LIBS"; then + pkg_cv_ZSTD_LIBS="$ZSTD_LIBS" + elif test -n "$PKG_CONFIG"; then + if test -n "$PKG_CONFIG" && \ + { { $as_echo "$as_me:${as_lineno-$LINENO}: \$PKG_CONFIG --exists --print-errors \"libzstd\""; } >&5 + ($PKG_CONFIG --exists --print-errors "libzstd") 2>&5 + ac_status=$? + $as_echo "$as_me:${as_lineno-$LINENO}: \$? = $ac_status" >&5 + test $ac_status = 0; }; then + pkg_cv_ZSTD_LIBS=`$PKG_CONFIG --libs "libzstd" 2>/dev/null` + test "x$?" != "x0" && pkg_failed=yes +else + pkg_failed=yes +fi + else + pkg_failed=untried +fi + + + +if test $pkg_failed = yes; then + { $as_echo "$as_me:${as_lineno-$LINENO}: result: no" >&5 +$as_echo "no" >&6; } + +if $PKG_CONFIG --atleast-pkgconfig-version 0.20; then + _pkg_short_errors_supported=yes +else + _pkg_short_errors_supported=no +fi + if test $_pkg_short_errors_supported = yes; then + ZSTD_PKG_ERRORS=`$PKG_CONFIG --short-errors --print-errors --cflags --libs "libzstd" 2>&1` + else + ZSTD_PKG_ERRORS=`$PKG_CONFIG --print-errors --cflags --libs "libzstd" 2>&1` + fi + # Put the nasty error message in config.log where it belongs + echo "$ZSTD_PKG_ERRORS" >&5 + + as_fn_error $? "Package requirements (libzstd) were not met: + +$ZSTD_PKG_ERRORS + +Consider adjusting the PKG_CONFIG_PATH environment variable if you +installed software in a non-standard prefix. + +Alternatively, you may set the environment variables ZSTD_CFLAGS +and ZSTD_LIBS to avoid the need to call pkg-config. +See the pkg-config man page for more details." "$LINENO" 5 +elif test $pkg_failed = untried; then + { $as_echo "$as_me:${as_lineno-$LINENO}: result: no" >&5 +$as_echo "no" >&6; } + { { $as_echo "$as_me:${as_lineno-$LINENO}: error: in \`$ac_pwd':" >&5 +$as_echo "$as_me: error: in \`$ac_pwd':" >&2;} +as_fn_error $? "The pkg-config script could not be found or is too old. Make sure it +is in your PATH or set the PKG_CONFIG environment variable to the full +path to pkg-config. + +Alternatively, you may set the environment variables ZSTD_CFLAGS +and ZSTD_LIBS to avoid the need to call pkg-config. +See the pkg-config man page for more details. + +To get pkg-config, see <http://pkg-config.freedesktop.org/>. +See \`config.log' for more details" "$LINENO" 5; } +else + ZSTD_CFLAGS=$pkg_cv_ZSTD_CFLAGS + ZSTD_LIBS=$pkg_cv_ZSTD_LIBS + { $as_echo "$as_me:${as_lineno-$LINENO}: result: yes" >&5 +$as_echo "yes" >&6; } + +fi + # We only care about -I, -D, and -L switches; + # note that -lzstd will be added by AC_CHECK_LIB below. + for pgac_option in $ZSTD_CFLAGS; do + case $pgac_option in + -I*|-D*) CPPFLAGS="$CPPFLAGS $pgac_option";; + esac + done + for pgac_option in $ZSTD_LIBS; do + case $pgac_option in + -L*) LDFLAGS="$LDFLAGS $pgac_option";; + esac + done +fi # # Assignments # @@ -13136,6 +13297,56 @@ fi fi +if test "$with_zstd" = yes ; then + { $as_echo "$as_me:${as_lineno-$LINENO}: checking for ZSTD_compress in -lzstd" >&5 +$as_echo_n "checking for ZSTD_compress in -lzstd... " >&6; } +if ${ac_cv_lib_zstd_ZSTD_compress+:} false; then : + $as_echo_n "(cached) " >&6 +else + ac_check_lib_save_LIBS=$LIBS +LIBS="-lzstd $LIBS" +cat confdefs.h - <<_ACEOF >conftest.$ac_ext +/* end confdefs.h. */ + +/* Override any GCC internal prototype to avoid an error. + Use char because int might match the return type of a GCC + builtin and then its argument prototype would still apply. */ +#ifdef __cplusplus +extern "C" +#endif +char ZSTD_compress (); +int +main () +{ +return ZSTD_compress (); + ; + return 0; +} +_ACEOF +if ac_fn_c_try_link "$LINENO"; then : + ac_cv_lib_zstd_ZSTD_compress=yes +else + ac_cv_lib_zstd_ZSTD_compress=no +fi +rm -f core conftest.err conftest.$ac_objext \ + conftest$ac_exeext conftest.$ac_ext +LIBS=$ac_check_lib_save_LIBS +fi +{ $as_echo "$as_me:${as_lineno-$LINENO}: result: $ac_cv_lib_zstd_ZSTD_compress" >&5 +$as_echo "$ac_cv_lib_zstd_ZSTD_compress" >&6; } +if test "x$ac_cv_lib_zstd_ZSTD_compress" = xyes; then : + cat >>confdefs.h <<_ACEOF +#define HAVE_LIBZSTD 1 +_ACEOF + + LIBS="-lzstd $LIBS" + +else + as_fn_error $? "library 'zstd' is required for ZSTD support" "$LINENO" 5 +fi + +fi + # Note: We can test for libldap_r only after we know PTHREAD_LIBS; # also, on AIX, we may need to have openssl in LIBS for this step. if test "$with_ldap" = yes ; then @@ -13856,6 +14067,23 @@ done fi +if test "$with_zstd" = yes; then + for ac_header in zstd.h +do : + ac_fn_c_check_header_mongrel "$LINENO" "zstd.h" "ac_cv_header_zstd_h" "$ac_includes_default" +if test "x$ac_cv_header_zstd_h" = xyes; then : + cat >>confdefs.h <<_ACEOF +#define HAVE_ZSTD_H 1 +_ACEOF + +else + as_fn_error $? "zstd.h header file is required for ZSTD" "$LINENO" 5 +fi + +done + +fi + if test "$with_gssapi" = yes ; then for ac_header in gssapi/gssapi.h do : @@ -15259,7 +15487,7 @@ else We can't simply define LARGE_OFF_T to be 9223372036854775807, since some C++ compilers masquerading as C compilers incorrectly reject 9223372036854775807. */ -#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62)) +#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31)) int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721 && LARGE_OFF_T % 2147483647 == 1) ? 1 : -1]; @@ -15305,7 +15533,7 @@ else We can't simply define LARGE_OFF_T to be 9223372036854775807, since some C++ compilers masquerading as C compilers incorrectly reject 9223372036854775807. */ -#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62)) +#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31)) int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721 && LARGE_OFF_T % 2147483647 == 1) ? 1 : -1]; @@ -15329,7 +15557,7 @@ rm -f core conftest.err conftest.$ac_objext conftest.$ac_ext We can't simply define LARGE_OFF_T to be 9223372036854775807, since some C++ compilers masquerading as C compilers incorrectly reject 9223372036854775807. */ -#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62)) +#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31)) int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721 && LARGE_OFF_T % 2147483647 == 1) ? 1 : -1]; @@ -15374,7 +15602,7 @@ else We can't simply define LARGE_OFF_T to be 9223372036854775807, since some C++ compilers masquerading as C compilers incorrectly reject 9223372036854775807. */ -#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62)) +#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31)) int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721 && LARGE_OFF_T % 2147483647 == 1) ? 1 : -1]; @@ -15398,7 +15626,7 @@ rm -f core conftest.err conftest.$ac_objext conftest.$ac_ext We can't simply define LARGE_OFF_T to be 9223372036854775807, since some C++ compilers masquerading as C compilers incorrectly reject 9223372036854775807. */ -#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62)) +#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31)) int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721 && LARGE_OFF_T % 2147483647 == 1) ? 1 : -1]; diff --git a/configure.ac b/configure.ac index 95287705f6..85e15ff9f8 100644 --- a/configure.ac +++ b/configure.ac @@ -1056,6 +1056,30 @@ if test "$with_lz4" = yes; then done fi +# +# ZSTD +# +AC_MSG_CHECKING([whether to build with ZSTD support]) +PGAC_ARG_BOOL(with, zstd, no, [build with ZSTD support], + [AC_DEFINE([USE_ZSTD], 1, [Define to 1 to build with ZSTD support. (--with-zstd)])]) +AC_MSG_RESULT([$with_zstd]) +AC_SUBST(with_zstd) + +if test "$with_zstd" = yes; then + PKG_CHECK_MODULES(ZSTD, libzstd) + # We only care about -I, -D, and -L switches; + # note that -lzstd will be added by AC_CHECK_LIB below. + for pgac_option in $ZSTD_CFLAGS; do + case $pgac_option in + -I*|-D*) CPPFLAGS="$CPPFLAGS $pgac_option";; + esac + done + for pgac_option in $ZSTD_LIBS; do + case $pgac_option in + -L*) LDFLAGS="$LDFLAGS $pgac_option";; + esac + done +fi # # Assignments # @@ -1325,6 +1349,10 @@ if test "$with_lz4" = yes ; then AC_CHECK_LIB(lz4, LZ4_compress_default, [], [AC_MSG_ERROR([library 'lz4' is required for LZ4 support])]) fi +if test "$with_zstd" = yes ; then + AC_CHECK_LIB(zstd, ZSTD_compress, [], [AC_MSG_ERROR([library 'zstd' is required for ZSTD support])]) +fi + # Note: We can test for libldap_r only after we know PTHREAD_LIBS; # also, on AIX, we may need to have openssl in LIBS for this step. if test "$with_ldap" = yes ; then @@ -1488,6 +1516,10 @@ if test "$with_lz4" = yes; then AC_CHECK_HEADERS(lz4.h, [], [AC_MSG_ERROR([lz4.h header file is required for LZ4])]) fi +if test "$with_zstd" = yes; then + AC_CHECK_HEADERS(zstd.h, [], [AC_MSG_ERROR([zstd.h header file is required for ZSTD])]) +fi + if test "$with_gssapi" = yes ; then AC_CHECK_HEADERS(gssapi/gssapi.h, [], [AC_CHECK_HEADERS(gssapi.h, [], [AC_MSG_ERROR([gssapi.h header file is required for GSSAPI])])]) diff --git a/doc/src/sgml/ref/pg_basebackup.sgml b/doc/src/sgml/ref/pg_basebackup.sgml index 44395a749b..5cadadf16c 100644 --- a/doc/src/sgml/ref/pg_basebackup.sgml +++ b/doc/src/sgml/ref/pg_basebackup.sgml @@ -276,6 +276,17 @@ PostgreSQL documentation </para> </listitem> </varlistentry> + + <varlistentry> + <term><literal>zstd</literal></term> + <listitem> + <para> + Compression is performed using <literal>zstd</literal> and the + suffix <filename>.zst</filename> will automatically be added to + compressed files. + </para> + </listitem> + </varlistentry> </variablelist> </para> diff --git a/src/backend/replication/Makefile b/src/backend/replication/Makefile index 74043ff331..2e6de7007f 100644 --- a/src/backend/replication/Makefile +++ b/src/backend/replication/Makefile @@ -20,6 +20,7 @@ OBJS = \ basebackup_copy.o \ basebackup_gzip.o \ basebackup_lz4.o \ + basebackup_zstd.o \ basebackup_progress.o \ basebackup_server.o \ basebackup_sink.o \ diff --git a/src/backend/replication/basebackup.c b/src/backend/replication/basebackup.c index 9dea1c9bcc..12992b0a4d 100644 --- a/src/backend/replication/basebackup.c +++ b/src/backend/replication/basebackup.c @@ -65,7 +65,8 @@ typedef enum { BACKUP_COMPRESSION_NONE, BACKUP_COMPRESSION_GZIP, - BACKUP_COMPRESSION_LZ4 + BACKUP_COMPRESSION_LZ4, + BACKUP_COMPRESSION_ZSTD } basebackup_compression_type; typedef struct @@ -912,6 +913,8 @@ parse_basebackup_options(List *options, basebackup_options *opt) } else if (strcmp(optval, "lz4") == 0) opt->compression = BACKUP_COMPRESSION_LZ4; + else if (strcmp(optval, "zstd") == 0) + opt->compression = BACKUP_COMPRESSION_ZSTD; else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -1018,6 +1021,8 @@ SendBaseBackup(BaseBackupCmd *cmd) sink = bbsink_gzip_new(sink, opt.compression_level); else if (opt.compression == BACKUP_COMPRESSION_LZ4) sink = bbsink_lz4_new(sink); + else if (opt.compression == BACKUP_COMPRESSION_ZSTD) + sink = bbsink_zstd_new(sink); /* Set up progress reporting. */ sink = bbsink_progress_new(sink, opt.progress); diff --git a/src/backend/replication/basebackup_zstd.c b/src/backend/replication/basebackup_zstd.c new file mode 100644 index 0000000000..a4bba94e7e --- /dev/null +++ b/src/backend/replication/basebackup_zstd.c @@ -0,0 +1,267 @@ +/*------------------------------------------------------------------------- + * + * basebackup_zstd.c + * Basebackup sink implementing zstd compression. + * + * Portions Copyright (c) 2010-2020, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/backend/replication/basebackup_zstd.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#ifdef HAVE_LIBZSTD +#include <zstd.h> +#endif + +#include "replication/basebackup_sink.h" + +#ifdef HAVE_LIBZSTD + +typedef struct bbsink_zstd +{ + /* Common information for all types of sink. */ + bbsink base; + + ZSTD_CCtx *cctx; + ZSTD_outBuffer zstd_outBuf; +} bbsink_zstd; + +static void bbsink_zstd_begin_backup(bbsink *sink); +static void bbsink_zstd_begin_archive(bbsink *sink, const char *archive_name); +static void bbsink_zstd_archive_contents(bbsink *sink, size_t avail_in); +static void bbsink_zstd_manifest_contents(bbsink *sink, size_t len); +static void bbsink_zstd_end_archive(bbsink *sink); +static void bbsink_zstd_cleanup(bbsink *sink); +static void bbsink_zstd_end_backup(bbsink *sink, XLogRecPtr endptr, + TimeLineID endtli); + +const bbsink_ops bbsink_zstd_ops = { + .begin_backup = bbsink_zstd_begin_backup, + .begin_archive = bbsink_zstd_begin_archive, + .archive_contents = bbsink_zstd_archive_contents, + .end_archive = bbsink_zstd_end_archive, + .begin_manifest = bbsink_forward_begin_manifest, + .manifest_contents = bbsink_zstd_manifest_contents, + .end_manifest = bbsink_forward_end_manifest, + .end_backup = bbsink_zstd_end_backup, + .cleanup = bbsink_zstd_cleanup +}; +#endif + +/* Create a new basebackup sink that performs zstd compression. */ +bbsink * +bbsink_zstd_new(bbsink *next) +{ +#ifndef HAVE_LIBZSTD + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("zstd compression is not supported by this build"))); +#else + bbsink_zstd *sink; + + Assert(next != NULL); + + sink = palloc0(sizeof(bbsink_zstd)); + *((const bbsink_ops **) &sink->base.bbs_ops) = &bbsink_zstd_ops; + sink->base.bbs_next = next; + + return &sink->base; +#endif +} + +#ifdef HAVE_LIBZSTD + +/* + * Begin backup. + */ +static void +bbsink_zstd_begin_backup(bbsink *sink) +{ + bbsink_zstd *mysink = (bbsink_zstd *) sink; + size_t output_buffer_bound; + + mysink->cctx = ZSTD_createCCtx(); + if (!mysink->cctx) + elog(ERROR, "could not create zstd compression context"); + + /* + * We need our own buffer, because we're going to pass different data to + * the next sink than what gets passed to us. + */ + mysink->base.bbs_buffer = palloc(mysink->base.bbs_buffer_length); + + /* + * Make sure that the next sink's bbs_buffer is big enough to accommodate + * the compressed input buffer. + */ + output_buffer_bound = ZSTD_compressBound(mysink->base.bbs_buffer_length); + + /* + * The buffer length is expected to be a multiple of BLCKSZ, so round up. + */ + output_buffer_bound = output_buffer_bound + BLCKSZ - + (output_buffer_bound % BLCKSZ); + + bbsink_begin_backup(sink->bbs_next, sink->bbs_state, output_buffer_bound); +} + +/* + * Prepare to compress the next archive. + */ +static void +bbsink_zstd_begin_archive(bbsink *sink, const char *archive_name) +{ + bbsink_zstd *mysink = (bbsink_zstd *) sink; + char *zstd_archive_name; + + /* + * At the start of each archive we reset the state to start a new + * compression operation. The parameters are sticky and they would stick + * around as we are resetting with option ZSTD_reset_session_only. + */ + ZSTD_CCtx_reset(mysink->cctx, ZSTD_reset_session_only); + + mysink->zstd_outBuf.dst = mysink->base.bbs_next->bbs_buffer; + mysink->zstd_outBuf.size = mysink->base.bbs_next->bbs_buffer_length; + mysink->zstd_outBuf.pos = 0; + + /* Add ".zst" to the archive name. */ + zstd_archive_name = psprintf("%s.zst", archive_name); + Assert(sink->bbs_next != NULL); + bbsink_begin_archive(sink->bbs_next, zstd_archive_name); + pfree(zstd_archive_name); +} + +/* + * Compress the input data to the output buffer until we run out of input + * data. Each time the output buffer falls below the compression bound for + * the input buffer, invoke the archive_contents() method for then next sink. + * + * Note that since we're compressing the input, it may very commonly happen + * that we consume all the input data without filling the output buffer. In + * that case, the compressed representation of the current input data won't + * actually be sent to the next bbsink until a later call to this function, + * or perhaps even not until bbsink_zstd_end_archive() is invoked. + */ +static void +bbsink_zstd_archive_contents(bbsink *sink, size_t len) +{ + bbsink_zstd *mysink = (bbsink_zstd *) sink; + + ZSTD_inBuffer inBuf = { mysink->base.bbs_buffer, len, 0 }; + + while (inBuf.pos < inBuf.size) + { + size_t yet_to_flush; + size_t required_outBuf_bound = ZSTD_compressBound(inBuf.size - inBuf.pos); + + /* + * If the out buffer is not left with enough space, send the output + * buffer to the next sink, and reset it. + */ + if ((mysink->zstd_outBuf.size - mysink->zstd_outBuf.pos) <= + required_outBuf_bound) + { + bbsink_archive_contents(mysink->base.bbs_next, mysink->zstd_outBuf.pos); + mysink->zstd_outBuf.dst = mysink->base.bbs_next->bbs_buffer; + mysink->zstd_outBuf.size = mysink->base.bbs_next->bbs_buffer_length; + mysink->zstd_outBuf.pos = 0; + } + + yet_to_flush = ZSTD_compressStream2(mysink->cctx, &mysink->zstd_outBuf, + &inBuf, ZSTD_e_continue); + + if (ZSTD_isError(yet_to_flush)) + elog(ERROR, "could not compress data: %s", ZSTD_getErrorName(yet_to_flush)); + } +} + +/* + * There might be some data inside zstd's internal buffers; we need to get that + * flushed out, also end the zstd frame and then get that forwarded to the + * successor sink as archive content. + * + * Then we can end processing for this archive. + */ +static void +bbsink_zstd_end_archive(bbsink *sink) +{ + bbsink_zstd *mysink = (bbsink_zstd *) sink; + size_t yet_to_flush; + + do + { + ZSTD_inBuffer in = {}; + + bbsink_archive_contents(mysink->base.bbs_next, mysink->zstd_outBuf.pos); + mysink->zstd_outBuf.dst = mysink->base.bbs_next->bbs_buffer; + mysink->zstd_outBuf.size = mysink->base.bbs_next->bbs_buffer_length; + mysink->zstd_outBuf.pos = 0; + + yet_to_flush = ZSTD_compressStream2(mysink->cctx, + &mysink->zstd_outBuf, + &in, ZSTD_e_end); + + if (ZSTD_isError(yet_to_flush)) + elog(ERROR, "could not compress data: %s", + ZSTD_getErrorName(yet_to_flush)); + } while(yet_to_flush > 0); + + /* Make sure to pass the any remaining bytes to the next sink. */ + bbsink_archive_contents(mysink->base.bbs_next, mysink->zstd_outBuf.pos); + + /* Pass on the information that this archive has ended. */ + bbsink_forward_end_archive(sink); +} + +/* + * Free the resources and context. + */ +static void +bbsink_zstd_end_backup(bbsink *sink, XLogRecPtr endptr, + TimeLineID endtli) +{ + bbsink_zstd *mysink = (bbsink_zstd *) sink; + + /* Release the context. */ + if (mysink->cctx) + { + ZSTD_freeCCtx(mysink->cctx); + mysink->cctx = NULL; + } + + bbsink_forward_end_backup(sink, endptr, endtli); +} + +/* + * Manifest contents are not compressed, but we do need to copy them into + * the successor sink's buffer, because we have our own. + */ +static void +bbsink_zstd_manifest_contents(bbsink *sink, size_t len) +{ + memcpy(sink->bbs_next->bbs_buffer, sink->bbs_buffer, len); + bbsink_manifest_contents(sink->bbs_next, len); +} + +/* + * In case the backup fails, make sure we free the compression context by + * calling ZSTD_freeCCtx if needed to avoid memory leak. + */ +static void +bbsink_zstd_cleanup(bbsink *sink) +{ + bbsink_zstd *mysink = (bbsink_zstd *) sink; + + /* Release the context if not already released. */ + if (mysink->cctx) + { + ZSTD_freeCCtx(mysink->cctx); + mysink->cctx = NULL; + } +} + +#endif diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c index d8da1cb2e9..b0c4a0f5b2 100644 --- a/src/bin/pg_basebackup/pg_basebackup.c +++ b/src/bin/pg_basebackup/pg_basebackup.c @@ -374,7 +374,7 @@ usage(void) " (in kB/s, or use suffix \"k\" or \"M\")\n")); printf(_(" -R, --write-recovery-conf\n" " write configuration for replication\n")); - printf(_(" --server-compression=none|gzip|gzip[1-9]|lz4\n" + printf(_(" --server-compression=none|gzip|gzip[1-9]|lz4|zstd\n" " compress backup on server\n")); printf(_(" -T, --tablespace-mapping=OLDDIR=NEWDIR\n" " relocate tablespace in OLDDIR to NEWDIR\n")); diff --git a/src/include/pg_config.h.in b/src/include/pg_config.h.in index 9d9bd6b9ef..61b2220eeb 100644 --- a/src/include/pg_config.h.in +++ b/src/include/pg_config.h.in @@ -325,6 +325,9 @@ /* Define to 1 if you have the `lz4' library (-llz4). */ #undef HAVE_LIBLZ4 +/* Define to 1 if you have the `zstd' library (-lzstd). */ +#undef HAVE_LIBZSTD + /* Define to 1 if you have the `m' library (-lm). */ #undef HAVE_LIBM @@ -367,6 +370,9 @@ /* Define to 1 if you have the <lz4.h> header file. */ #undef HAVE_LZ4_H +/* Define to 1 if you have the <zstd.h> header file. */ +#undef HAVE_ZSTD_H + /* Define to 1 if you have the <mbarrier.h> header file. */ #undef HAVE_MBARRIER_H diff --git a/src/include/replication/basebackup_sink.h b/src/include/replication/basebackup_sink.h index 964752ef5d..8c18917a76 100644 --- a/src/include/replication/basebackup_sink.h +++ b/src/include/replication/basebackup_sink.h @@ -286,6 +286,7 @@ extern bbsink *bbsink_copystream_new(bool send_to_client); extern bbsink *bbsink_copytblspc_new(void); extern bbsink *bbsink_gzip_new(bbsink *next, int compresslevel); extern bbsink *bbsink_lz4_new(bbsink *next); +extern bbsink *bbsink_zstd_new(bbsink *next); extern bbsink *bbsink_progress_new(bbsink *next, bool estimate_backup_size); extern bbsink *bbsink_server_new(bbsink *next, char *pathname); extern bbsink *bbsink_throttle_new(bbsink *next, uint32 maxrate); -- 2.25.1 ^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: refactoring basebackup.c @ 2022-01-18 21:27 Robert Haas <[email protected]> parent: Jeevan Ladhe <[email protected]> 1 sibling, 0 replies; 4+ messages in thread From: Robert Haas @ 2022-01-18 21:27 UTC (permalink / raw) To: Jeevan Ladhe <[email protected]>; +Cc: tushar <[email protected]>; Dmitry Dolgov <[email protected]>; Mark Dilger <[email protected]>; pgsql-hackers On Tue, Jan 18, 2022 at 9:43 AM Jeevan Ladhe <[email protected]> wrote: > The patch surely needs some grooming, but I am expecting some initial > review, specially in the area where we are trying to close the zstd stream > in bbsink_zstd_end_archive(). We need to tell the zstd library to end the > compression by calling ZSTD_compressStream2() thereby sending a > ZSTD_e_end flag. But, this also needs some input string, which per > example[1] line # 686, I have taken as an empty ZSTD_inBuffer. As far as I can see, this is correct. I found https://zstd.docsforge.com/dev/api-documentation/#streaming-compression-howto which seems to endorse what you've done here. One (minor) thing that I notice is that, the way you've written the loop in bbsink_zstd_end_archive(), I think it will typically call bbsink_archive_contents() twice. It will flush whatever is already present in the next sink's buffer as a result of the previous calls to bbsink_zstd_archive_contents(), and then it will call ZSTD_compressStream2() which will partially refill the buffer you just emptied, and then there will be nothing left in the internal buffer, so it will call bbsink_archive_contents() again. But ... the initial flush may not have been necessary. It could be that there was enough space already in the output buffer for the ZSTD_compressStream2() call to succeed without a prior flush. So maybe: do { yet_to_flush = ZSTD_compressStream2(..., ZSTD_e_end); check ZSTD_isError here; if (mysink->zstd_outBuf.pos > 0) bbsink_archive_contents(); } while (yet_to_flush > 0); I believe this might be very slightly more efficient. -- Robert Haas EDB: http://www.enterprisedb.com ^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: refactoring basebackup.c @ 2022-01-19 16:30 tushar <[email protected]> parent: Jeevan Ladhe <[email protected]> 1 sibling, 0 replies; 4+ messages in thread From: tushar @ 2022-01-19 16:30 UTC (permalink / raw) To: Jeevan Ladhe <[email protected]>; +Cc: Robert Haas <[email protected]>; Dmitry Dolgov <[email protected]>; Mark Dilger <[email protected]>; pgsql-hackers On 1/18/22 8:12 PM, Jeevan Ladhe wrote: > Similar to LZ4 server-side compression, I have also tried to add a ZSTD > server-side compression in the attached patch. Thanks Jeevan. while testing found one scenario where the server is getting crash while performing pg_basebackup against server-compression=zstd for a huge data second time Steps to reproduce --PG sources ( apply v11-0001,v11-0001,v9-0001,v9-0002 , configure --with-lz4,--with-zstd, make/install, initdb, start server) --insert huge data (./pgbench -i -s 2000 postgres) --restart the server (./pg_ctl -D data restart) --pg_basebackup ( ./pg_basebackup -t server:/tmp/yc1 --server-compression=zstd -R -Xnone -n -N -l 'ccc' --no-estimate-size -v) --insert huge data (./pgbench -i -s 1000 postgres) --restart the server (./pg_ctl -D data restart) --run pg_basebackup again (./pg_basebackup -t server:/tmp/yc11 --server-compression=zstd -v -Xnone ) [edb@centos7tushar bin]$ ./pg_basebackup -t server:/tmp/yc11 --server-compression=zstd -v -Xnone pg_basebackup: initiating base backup, waiting for checkpoint to complete 2022-01-19 21:23:26.508 IST [30219] LOG: checkpoint starting: force wait 2022-01-19 21:23:26.608 IST [30219] LOG: checkpoint complete: wrote 0 buffers (0.0%); 0 WAL file(s) added, 1 removed, 0 recycled; write=0.001 s, sync=0.001 s, total=0.101 s; sync files=0, longest=0.000 s, average=0.000 s; distance=16369 kB, estimate=16369 kB pg_basebackup: checkpoint completed TRAP: FailedAssertion("len > 0 && len <= sink->bbs_buffer_length", File: "../../../src/include/replication/basebackup_sink.h", Line: 208, PID: 30226) postgres: walsender edb [local] sending backup "pg_basebackup base backup"(ExceptionalCondition+0x7a)[0x94ceca] postgres: walsender edb [local] sending backup "pg_basebackup base backup"[0x7b9a08] postgres: walsender edb [local] sending backup "pg_basebackup base backup"[0x7b9be2] postgres: walsender edb [local] sending backup "pg_basebackup base backup"[0x7b5b30] postgres: walsender edb [local] sending backup "pg_basebackup base backup"(SendBaseBackup+0x563)[0x7b7053] postgres: walsender edb [local] sending backup "pg_basebackup base backup"(exec_replication_command+0x961)[0x7c9a41] postgres: walsender edb [local] sending backup "pg_basebackup base backup"(PostgresMain+0x92f)[0x81ca3f] postgres: walsender edb [local] sending backup "pg_basebackup base backup"[0x48e430] postgres: walsender edb [local] sending backup "pg_basebackup base backup"(PostmasterMain+0xfd2)[0x785702] postgres: walsender edb [local] sending backup "pg_basebackup base backup"(main+0x1c6)[0x48fb96] /lib64/libc.so.6(__libc_start_main+0xf5)[0x7f63642c8555] postgres: walsender edb [local] sending backup "pg_basebackup base backup"[0x48feb5] pg_basebackup: error: could not read COPY data: server closed the connection unexpectedly This probably means the server terminated abnormally before or while processing the request. 2022-01-19 21:25:34.485 IST [30205] LOG: server process (PID 30226) was terminated by signal 6: Aborted 2022-01-19 21:25:34.485 IST [30205] DETAIL: Failed process was running: BASE_BACKUP ( LABEL 'pg_basebackup base backup', PROGRESS, MANIFEST 'yes', TABLESPACE_MAP, TARGET 'server', TARGET_DETAIL '/tmp/yc11', COMPRESSION 'zstd') 2022-01-19 21:25:34.485 IST [30205] LOG: terminating any other active server processes [edb@centos7tushar bin]$ 2022-01-19 21:25:34.489 IST [30205] LOG: all server processes terminated; reinitializing 2022-01-19 21:25:34.536 IST [30228] LOG: database system was interrupted; last known up at 2022-01-19 21:23:26 IST 2022-01-19 21:25:34.669 IST [30228] LOG: database system was not properly shut down; automatic recovery in progress 2022-01-19 21:25:34.671 IST [30228] LOG: redo starts at 9/7000028 2022-01-19 21:25:34.671 IST [30228] LOG: invalid record length at 9/7000148: wanted 24, got 0 2022-01-19 21:25:34.671 IST [30228] LOG: redo done at 9/7000110 system usage: CPU: user: 0.00 s, system: 0.00 s, elapsed: 0.00 s 2022-01-19 21:25:34.673 IST [30229] LOG: checkpoint starting: end-of-recovery immediate wait 2022-01-19 21:25:34.713 IST [30229] LOG: checkpoint complete: wrote 3 buffers (0.0%); 0 WAL file(s) added, 0 removed, 0 recycled; write=0.003 s, sync=0.001 s, total=0.041 s; sync files=2, longest=0.001 s, average=0.001 s; distance=0 kB, estimate=0 kB 2022-01-19 21:25:34.718 IST [30205] LOG: database system is ready to accept connections Observation - if we change server-compression method to lz4 from zstd then it is NOT happening. [edb@centos7tushar bin]$ ./pg_basebackup -t server:/tmp/ycc1 --server-compression=lz4 -v -Xnone pg_basebackup: initiating base backup, waiting for checkpoint to complete 2022-01-19 21:27:51.642 IST [30229] LOG: checkpoint starting: force wait 2022-01-19 21:27:51.687 IST [30229] LOG: checkpoint complete: wrote 0 buffers (0.0%); 0 WAL file(s) added, 1 removed, 0 recycled; write=0.001 s, sync=0.001 s, total=0.046 s; sync files=0, longest=0.000 s, average=0.000 s; distance=16383 kB, estimate=16383 kB pg_basebackup: checkpoint completed NOTICE: WAL archiving is not enabled; you must ensure that all required WAL segments are copied through other means to complete the backup pg_basebackup: base backup completed [edb@centos7tushar bin]$ -- regards,tushar EnterpriseDB https://www.enterprisedb.com/ The Enterprise PostgreSQL Company ^ permalink raw reply [nested|flat] 4+ messages in thread
end of thread, other threads:[~2022-01-19 16:30 UTC | newest] Thread overview: 4+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2020-09-17 15:26 [PATCH 2/9] Move IS [NOT] NULL handling from BRIN support functions Tomas Vondra <[email protected]> 2022-01-18 14:42 Re: refactoring basebackup.c Jeevan Ladhe <[email protected]> 2022-01-18 21:27 ` Re: refactoring basebackup.c Robert Haas <[email protected]> 2022-01-19 16:30 ` Re: refactoring basebackup.c tushar <[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