On Sat, 3 Sept 2022 at 00:37, Ranier Vilela <ranier...@gmail.com> wrote:
>> But +1 to fix this and other issues even if they would never crash.
Yeah, I don't think any of this coding would lead to a crash, but it's
pretty weird coding and we should fix it.
> 1. Once that ranges->nranges is invariant, avoid the loop if ranges->nranges
> <= 0.
> This matches the current behavior.
>
> 2. Once that ranges->nsorted is invariant, avoid the loop if ranges->nsorted
> <= 0.
> This matches the current behavior.
>
> 3. Remove the invariant cxt from ranges->nsorted loop.
>
> 4. Avoid possible overflows when using int to store length strings.
>
> 5. Avoid possible out-of-bounds when ranges->nranges == 0.
>
> 6. Avoid overhead when using unnecessary StringInfoData to convert Datum a to
> Text b.
I've ripped out #4 and #6 for now. I think we should do #6 in master
only, probably as part of a wider cleanup of StringInfo misusages.
I also spent some time trying to ensure I understand this code
correctly. I was unable to work out what the nsorted field was from
just the comments. I needed to look at the code to figure it out, so I
think the comments for that field need to be improved. A few of the
others were not that clear either. I hope I've improved them in the
attached.
I was also a bit confused at various other comments. e.g:
/*
* Is the value greater than the maxval? If yes, we'll recurse to the
* right side of range array.
*/
I don't see any sort of recursion going on here. All I see are
skipping of values that are out of bounds of the lower bound of the
lowest range, and above the upper bound of the highest range.
I propose to backpatch the attached into v14 tomorrow morning (about
12 hours from now).
The only other weird coding I found was in brin_range_deserialize:
for (i = 0; (i < nvalues) && (!typbyval); i++)
I imagine most compilers would optimize that so that the typbyval
check is done before the first loop and not done on every loop, but I
don't think that coding practice helps the human readers out much. I
left that one alone, for now.
David
diff --git a/src/backend/access/brin/brin_minmax_multi.c
b/src/backend/access/brin/brin_minmax_multi.c
index a581659fe2..99ef97206a 100644
--- a/src/backend/access/brin/brin_minmax_multi.c
+++ b/src/backend/access/brin/brin_minmax_multi.c
@@ -149,13 +149,16 @@ typedef struct MinMaxMultiOptions
* single-point ranges. That is, we have (2*nranges + nvalues) boundary
* values in the array.
*
- * +---------------------------------+-------------------------------+
- * | ranges (sorted pairs of values) | sorted values (single points) |
- * +---------------------------------+-------------------------------+
+ * +-------------------------+----------------------------------+
+ * | ranges (2 * nranges of) | single point values (nvalues of) |
+ * +-------------------------+----------------------------------+
*
* This allows us to quickly add new values, and store outliers without
* making the other ranges very wide.
*
+ * 'nsorted' denotes how many of 'nvalues' in the values[] array are sorted.
+ * When nsorted == nvalues, all single point values are sorted.
+ *
* We never store more than maxvalues values (as set by values_per_range
* reloption). If needed we merge some of the ranges.
*
@@ -173,10 +176,10 @@ typedef struct Ranges
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) */
+ int nranges; /* number of ranges in
the values[] array */
+ int nsorted; /* number of nvalues
which are sorted */
+ int nvalues; /* number of point
values in values[] */
+ int maxvalues; /* number of elements
in the values[] array */
/*
* We simply add the values into a large buffer, without any expensive
@@ -318,102 +321,99 @@ AssertCheckRanges(Ranges *ranges, FmgrInfo *cmpFn, Oid
colloid)
* Check that none of the values are not covered by ranges (both sorted
* and unsorted)
*/
- for (i = 0; i < ranges->nvalues; i++)
+ if (ranges->nranges > 0)
{
- Datum compar;
- int start,
- end;
- Datum minvalue,
- maxvalue;
-
- Datum value = ranges->values[2 * ranges->nranges + i];
-
- 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 maxval? 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)
+ for (i = 0; i < ranges->nvalues; i++)
{
- int midpoint = (start + end) / 2;
-
- /* this means we ran out of ranges in the last step */
- if (start > end)
- break;
+ Datum compar;
+ int start,
+ end;
+ Datum minvalue = ranges->values[0];
+ Datum maxvalue = ranges->values[2 *
ranges->nranges - 1];
+ Datum value = ranges->values[2 *
ranges->nranges + i];
- /* copy the min/max values from the ranges */
- minvalue = ranges->values[2 * midpoint];
- maxvalue = ranges->values[2 * midpoint + 1];
+ compar = FunctionCall2Coll(cmpFn, colloid, value,
minvalue);
/*
- * Is the value smaller than the minval? If yes, we'll
recurse to
- * the left side of range array.
+ * If the value is smaller than the lower bound in the
first range
+ * then it cannot possibly be in any of the ranges.
*/
- compar = FunctionCall2Coll(cmpFn, colloid, value,
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(cmpFn, colloid, maxvalue,
value);
- /* larger than the largest value in this range */
+ /*
+ * Likewise, if the value is larger than the upper
bound of the
+ * final range, then it cannot possibly be inside any
of the
+ * ranges.
+ */
if (DatumGetBool(compar))
- {
- start = (midpoint + 1);
continue;
- }
- /* hey, we found a matching range */
- Assert(false);
+ /* bsearch the ranges to see if 'value' fits within any
of them */
+ 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)
+ break;
+
+ /* 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(cmpFn, colloid,
value, 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(cmpFn, colloid,
maxvalue, value);
+
+ /* larger than the largest value in this range
*/
+ if (DatumGetBool(compar))
+ {
+ start = (midpoint + 1);
+ continue;
+ }
+
+ /* 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++)
+ /* and values in the unsorted part must not be in the sorted part */
+ if (ranges->nsorted > 0)
{
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);
+ for (i = ranges->nsorted; i < ranges->nvalues; i++)
+ {
+ Datum value = ranges->values[2 *
ranges->nranges + i];
+
+ Assert(bsearch_arg(&value, &ranges->values[2 *
ranges->nranges],
+ ranges->nsorted,
sizeof(Datum),
+ compare_values, (void
*) &cxt) == NULL);
+ }
}
#endif
}
@@ -923,8 +923,8 @@ has_matching_range(BrinDesc *bdesc, Oid colloid, Ranges
*ranges,
{
Datum compar;
- Datum minvalue = ranges->values[0];
- Datum maxvalue = ranges->values[2 * ranges->nranges - 1];
+ Datum minvalue;
+ Datum maxvalue;
FmgrInfo *cmpLessFn;
FmgrInfo *cmpGreaterFn;
@@ -936,6 +936,9 @@ has_matching_range(BrinDesc *bdesc, Oid colloid, Ranges
*ranges,
if (ranges->nranges == 0)
return false;
+ minvalue = ranges->values[0];
+ 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
