optimise-mul_var.patch

text/x-patch

Filename: optimise-mul_var.patch
Type: text/x-patch
Part: 0
Message: Re: Optimize numeric multiplication for one and two base-NBASE digit multiplicands.

Patch

Format: unified
File+
src/backend/utils/adt/numeric.c 43 23
diff --git a/src/backend/utils/adt/numeric.c b/src/backend/utils/adt/numeric.c
new file mode 100644
index f6e20cf..2a2e22d
--- a/src/backend/utils/adt/numeric.c
+++ b/src/backend/utils/adt/numeric.c
@@ -8690,18 +8690,21 @@ mul_var(const NumericVar *var1, const Nu
 	int			res_sign;
 	int			res_weight;
 	int			maxdigits;
-	int		   *dig;
-	int			carry;
+	uint32	   *dig,
+			   *dig_i1_2;
+	uint32		carry;
 	int			maxdig;
-	int			newdig;
+	uint32		newdig;
 	int			var1ndigits;
 	int			var2ndigits;
 	NumericDigit *var1digits;
 	NumericDigit *var2digits;
 	NumericDigit *res_digits;
+	NumericDigit var1digit;
 	int			i,
 				i1,
-				i2;
+				i2,
+				i2limit;
 
 	/*
 	 * Arrange for var1 to be the shorter of the two numbers.  This improves
@@ -8775,23 +8778,22 @@ mul_var(const NumericVar *var1, const Nu
 	}
 
 	/*
-	 * We do the arithmetic in an array "dig[]" of signed int's.  Since
-	 * INT_MAX is noticeably larger than NBASE*NBASE, this gives us headroom
-	 * to avoid normalizing carries immediately.
+	 * We do the arithmetic in an array "dig[]" of unsigned 32-bit integers.
+	 * Since PG_UINT32_MAX is noticeably larger than NBASE*NBASE, this gives
+	 * us headroom to avoid normalizing carries immediately.
 	 *
 	 * maxdig tracks the maximum possible value of any dig[] entry; when this
-	 * threatens to exceed INT_MAX, we take the time to propagate carries.
-	 * Furthermore, we need to ensure that overflow doesn't occur during the
-	 * carry propagation passes either.  The carry values could be as much as
-	 * INT_MAX/NBASE, so really we must normalize when digits threaten to
-	 * exceed INT_MAX - INT_MAX/NBASE.
+	 * threatens to exceed PG_UINT32_MAX, we take the time to propagate
+	 * carries.  Furthermore, we need to ensure that overflow doesn't occur
+	 * during the carry propagation passes either.  The carry values could be
+	 * as much as PG_UINT32_MAX/NBASE, so really we must normalize when digits
+	 * threaten to exceed PG_UINT32_MAX - PG_UINT32_MAX/NBASE.
 	 *
 	 * To avoid overflow in maxdig itself, it actually represents the max
 	 * possible value divided by NBASE-1, ie, at the top of the loop it is
 	 * known that no dig[] entry exceeds maxdig * (NBASE-1).
 	 */
-	dig = (int *) palloc0(res_ndigits * sizeof(int));
-	maxdig = 0;
+	dig = (uint32 *) palloc(res_ndigits * sizeof(uint32));
 
 	/*
 	 * The least significant digits of var1 should be ignored if they don't
@@ -8801,17 +8803,37 @@ mul_var(const NumericVar *var1, const Nu
 	 * Digit i1 of var1 and digit i2 of var2 are multiplied and added to digit
 	 * i1+i2+2 of the accumulator array, so we need only consider digits of
 	 * var1 for which i1 <= res_ndigits - 3.
+	 *
+	 * We process the least significant digit of var1 outside the main loop,
+	 * since the results can be applied directly to dig[], rather than being
+	 * added to it.
 	 */
-	for (i1 = Min(var1ndigits - 1, res_ndigits - 3); i1 >= 0; i1--)
+	i1 = Min(var1ndigits - 1, res_ndigits - 3);
+	var1digit = var1digits[i1];
+	maxdig = var1digit;
+
+	/*
+	 * Multiply by the least significant applicable digit of var1 -- see
+	 * comments below.  Any changes made there should be reflected here.
+	 */
+	i2limit = Min(var2ndigits, res_ndigits - i1 - 2);
+	dig_i1_2 = &dig[i1 + 2];
+
+	memset(dig, 0, sizeof(uint32) * (i1 + 2));
+	for (i2 = 0; i2 < i2limit; i2++)
+		dig_i1_2[i2] = var1digit * var2digits[i2];
+
+	/* Add contributions from remaining digits of var1 */
+	for (i1 = i1 - 1; i1 >= 0; i1--)
 	{
-		NumericDigit var1digit = var1digits[i1];
+		var1digit = var1digits[i1];
 
 		if (var1digit == 0)
 			continue;
 
 		/* Time to normalize? */
 		maxdig += var1digit;
-		if (maxdig > (INT_MAX - INT_MAX / NBASE) / (NBASE - 1))
+		if (maxdig > (PG_UINT32_MAX - PG_UINT32_MAX / NBASE) / (NBASE - 1))
 		{
 			/* Yes, do it */
 			carry = 0;
@@ -8845,13 +8867,11 @@ mul_var(const NumericVar *var1, const Nu
 		 * Since we aren't propagating carries in this loop, the order does
 		 * not matter.
 		 */
-		{
-			int			i2limit = Min(var2ndigits, res_ndigits - i1 - 2);
-			int		   *dig_i1_2 = &dig[i1 + 2];
+		i2limit = Min(var2ndigits, res_ndigits - i1 - 2);
+		dig_i1_2 = &dig[i1 + 2];
 
-			for (i2 = 0; i2 < i2limit; i2++)
-				dig_i1_2[i2] += var1digit * var2digits[i2];
-		}
+		for (i2 = 0; i2 < i2limit; i2++)
+			dig_i1_2[i2] += var1digit * var2digits[i2];
 	}
 
 	/*