* Transactions on Mathematical Software, Vol. 24, No. 4, December 1998,
* pages 359-367.
*
- * Copyright (c) 1998-2004, PostgreSQL Global Development Group
+ * Copyright (c) 1998-2010, PostgreSQL Global Development Group
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/utils/adt/numeric.c,v 1.78 2004/08/29 05:06:49 momjian Exp $
+ * $PostgreSQL: pgsql/src/backend/utils/adt/numeric.c,v 1.121 2010/01/07 04:53:34 tgl Exp $
*
*-------------------------------------------------------------------------
*/
#include <float.h>
#include <limits.h>
#include <math.h>
-#include <errno.h>
+#include "access/hash.h"
#include "catalog/pg_type.h"
#include "libpq/pqformat.h"
+#include "miscadmin.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/int8.h"
* NBASE that's less than sqrt(INT_MAX), in practice we are only interested
* in NBASE a power of ten, so that I/O conversions and decimal rounding
* are easy. Also, it's actually more efficient if NBASE is rather less than
- * sqrt(INT_MAX), so that there is "headroom" for mul_var and div_var to
+ * sqrt(INT_MAX), so that there is "headroom" for mul_var and div_var_fast to
* postpone processing carries.
* ----------
*/
/* ----------
+ * NumericVar is the format we use for arithmetic. The digit-array part
+ * is the same as the NumericData storage format, but the header is more
+ * complex.
+ *
* The value represented by a NumericVar is determined by the sign, weight,
* ndigits, and digits[] array.
* Note: the first digit of a NumericVar's value is assumed to be multiplied
* NumericVar. digits points at the first digit in actual use (the one
* with the specified weight). We normally leave an unused digit or two
* (preset to zeroes) between buf and digits, so that there is room to store
- * a carry out of the top digit without special pushups. We just need to
+ * a carry out of the top digit without reallocating space. We just need to
* decrement digits (and increment weight) to make room for the carry digit.
* (There is no such extra space in a numeric value stored in the database,
* only in a NumericVar in memory.)
{
int ndigits; /* # of digits in digits[] - can be 0! */
int weight; /* weight of first digit */
- int sign; /* NUMERIC_POS, NUMERIC_NEG, or
- * NUMERIC_NAN */
+ int sign; /* NUMERIC_POS, NUMERIC_NEG, or NUMERIC_NAN */
int dscale; /* display scale */
NumericDigit *buf; /* start of palloc'd space for digits[] */
NumericDigit *digits; /* base-NBASE digits */
#if DEC_DIGITS == 4
static NumericDigit const_zero_point_five_data[1] = {5000};
-
#elif DEC_DIGITS == 2
static NumericDigit const_zero_point_five_data[1] = {50};
-
#elif DEC_DIGITS == 1
static NumericDigit const_zero_point_five_data[1] = {5};
#endif
#if DEC_DIGITS == 4
static NumericDigit const_zero_point_nine_data[1] = {9000};
-
#elif DEC_DIGITS == 2
static NumericDigit const_zero_point_nine_data[1] = {90};
-
#elif DEC_DIGITS == 1
static NumericDigit const_zero_point_nine_data[1] = {9};
#endif
static NumericDigit const_zero_point_01_data[1] = {100};
static NumericVar const_zero_point_01 =
{1, -1, NUMERIC_POS, 2, NULL, const_zero_point_01_data};
-
#elif DEC_DIGITS == 2
static NumericDigit const_zero_point_01_data[1] = {1};
static NumericVar const_zero_point_01 =
{1, -1, NUMERIC_POS, 2, NULL, const_zero_point_01_data};
-
#elif DEC_DIGITS == 1
static NumericDigit const_zero_point_01_data[1] = {1};
static NumericVar const_zero_point_01 =
#if DEC_DIGITS == 4
static NumericDigit const_one_point_one_data[2] = {1, 1000};
-
#elif DEC_DIGITS == 2
static NumericDigit const_one_point_one_data[2] = {1, 10};
-
#elif DEC_DIGITS == 1
static NumericDigit const_one_point_one_data[2] = {1, 1};
#endif
#ifdef NUMERIC_DEBUG
static void dump_numeric(const char *str, Numeric num);
static void dump_var(const char *str, NumericVar *var);
-
#else
#define dump_numeric(s,n)
#define dump_var(s,v)
#define init_var(v) MemSetAligned(v, 0, sizeof(NumericVar))
+#define NUMERIC_DIGITS(num) ((NumericDigit *)(num)->n_data)
+#define NUMERIC_NDIGITS(num) \
+ ((VARSIZE(num) - NUMERIC_HDRSZ) / sizeof(NumericDigit))
+
static void alloc_var(NumericVar *var, int ndigits);
static void free_var(NumericVar *var);
static void zero_var(NumericVar *var);
-static void set_var_from_str(const char *str, NumericVar *dest);
+static const char *set_var_from_str(const char *str, const char *cp,
+ NumericVar *dest);
static void set_var_from_num(Numeric value, NumericVar *dest);
static void set_var_from_var(NumericVar *value, NumericVar *dest);
static char *get_str_from_var(NumericVar *var, int dscale);
+static char *get_str_from_var_sci(NumericVar *var, int rscale);
static Numeric make_result(NumericVar *var);
static int cmp_numerics(Numeric num1, Numeric num2);
static int cmp_var(NumericVar *var1, NumericVar *var2);
+static int cmp_var_common(const NumericDigit *var1digits, int var1ndigits,
+ int var1weight, int var1sign,
+ const NumericDigit *var2digits, int var2ndigits,
+ int var2weight, int var2sign);
static void add_var(NumericVar *var1, NumericVar *var2, NumericVar *result);
static void sub_var(NumericVar *var1, NumericVar *var2, NumericVar *result);
static void mul_var(NumericVar *var1, NumericVar *var2, NumericVar *result,
int rscale);
static void div_var(NumericVar *var1, NumericVar *var2, NumericVar *result,
- int rscale);
+ int rscale, bool round);
+static void div_var_fast(NumericVar *var1, NumericVar *var2, NumericVar *result,
+ int rscale, bool round);
static int select_div_scale(NumericVar *var1, NumericVar *var2);
static void mod_var(NumericVar *var1, NumericVar *var2, NumericVar *result);
static void ceil_var(NumericVar *var, NumericVar *result);
int rscale);
static int cmp_abs(NumericVar *var1, NumericVar *var2);
+static int cmp_abs_common(const NumericDigit *var1digits, int var1ndigits,
+ int var1weight,
+ const NumericDigit *var2digits, int var2ndigits,
+ int var2weight);
static void add_abs(NumericVar *var1, NumericVar *var2, NumericVar *result);
static void sub_abs(NumericVar *var1, NumericVar *var2, NumericVar *result);
static void round_var(NumericVar *var, int rscale);
Oid typelem = PG_GETARG_OID(1);
#endif
int32 typmod = PG_GETARG_INT32(2);
- NumericVar value;
Numeric res;
+ const char *cp;
+
+ /* Skip leading spaces */
+ cp = str;
+ while (*cp)
+ {
+ if (!isspace((unsigned char) *cp))
+ break;
+ cp++;
+ }
/*
* Check for NaN
*/
- if (pg_strcasecmp(str, "NaN") == 0)
- PG_RETURN_NUMERIC(make_result(&const_nan));
+ if (pg_strncasecmp(cp, "NaN", 3) == 0)
+ {
+ res = make_result(&const_nan);
- /*
- * Use set_var_from_str() to parse the input string and return it in
- * the packed DB storage format
- */
- init_var(&value);
- set_var_from_str(str, &value);
+ /* Should be nothing left but spaces */
+ cp += 3;
+ while (*cp)
+ {
+ if (!isspace((unsigned char) *cp))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
+ errmsg("invalid input syntax for type numeric: \"%s\"",
+ str)));
+ cp++;
+ }
+ }
+ else
+ {
+ /*
+ * Use set_var_from_str() to parse a normal numeric value
+ */
+ NumericVar value;
- apply_typmod(&value, typmod);
+ init_var(&value);
- res = make_result(&value);
- free_var(&value);
+ cp = set_var_from_str(str, cp, &value);
+
+ /*
+ * We duplicate a few lines of code here because we would like to
+ * throw any trailing-junk syntax error before any semantic error
+ * resulting from apply_typmod. We can't easily fold the two cases
+ * together because we mustn't apply apply_typmod to a NaN.
+ */
+ while (*cp)
+ {
+ if (!isspace((unsigned char) *cp))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
+ errmsg("invalid input syntax for type numeric: \"%s\"",
+ str)));
+ cp++;
+ }
+
+ apply_typmod(&value, typmod);
+
+ res = make_result(&value);
+ free_var(&value);
+ }
PG_RETURN_NUMERIC(res);
}
* Get the number in the variable format.
*
* Even if we didn't need to change format, we'd still need to copy the
- * value to have a modifiable copy for rounding. set_var_from_num()
- * also guarantees there is extra digit space in case we produce a
- * carry out from rounding.
+ * value to have a modifiable copy for rounding. set_var_from_num() also
+ * guarantees there is extra digit space in case we produce a carry out
+ * from rounding.
*/
init_var(&x);
set_var_from_num(num, &x);
PG_RETURN_CSTRING(str);
}
+/*
+ * numeric_out_sci() -
+ *
+ * Output function for numeric data type in scientific notation.
+ */
+char *
+numeric_out_sci(Numeric num, int scale)
+{
+ NumericVar x;
+ char *str;
+
+ /*
+ * Handle NaN
+ */
+ if (NUMERIC_IS_NAN(num))
+ return pstrdup("NaN");
+
+ init_var(&x);
+ set_var_from_num(num, &x);
+
+ str = get_str_from_var_sci(&x, scale);
+
+ free_var(&x);
+ return str;
+}
+
/*
* numeric_recv - converts external binary format to numeric
*
numeric_recv(PG_FUNCTION_ARGS)
{
StringInfo buf = (StringInfo) PG_GETARG_POINTER(0);
+
+#ifdef NOT_USED
+ Oid typelem = PG_GETARG_OID(1);
+#endif
+ int32 typmod = PG_GETARG_INT32(2);
NumericVar value;
Numeric res;
int len,
if (d < 0 || d >= NBASE)
ereport(ERROR,
(errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
- errmsg("invalid digit in external \"numeric\" value")));
+ errmsg("invalid digit in external \"numeric\" value")));
value.digits[i] = d;
}
+ apply_typmod(&value, typmod);
+
res = make_result(&value);
free_var(&value);
* scale of the attribute have to be applied on the value.
*/
Datum
-numeric(PG_FUNCTION_ARGS)
+numeric (PG_FUNCTION_ARGS)
{
Numeric num = PG_GETARG_NUMERIC(0);
int32 typmod = PG_GETARG_INT32(1);
PG_RETURN_NUMERIC(make_result(&const_nan));
/*
- * If the value isn't a valid type modifier, simply return a copy of
- * the input value
+ * If the value isn't a valid type modifier, simply return a copy of the
+ * input value
*/
if (typmod < (int32) (VARHDRSZ))
{
- new = (Numeric) palloc(num->varlen);
- memcpy(new, num, num->varlen);
+ new = (Numeric) palloc(VARSIZE(num));
+ memcpy(new, num, VARSIZE(num));
PG_RETURN_NUMERIC(new);
}
/*
* If the number is certainly in bounds and due to the target scale no
- * rounding could be necessary, just make a copy of the input and
- * modify its scale fields. (Note we assume the existing dscale is
- * honest...)
+ * rounding could be necessary, just make a copy of the input and modify
+ * its scale fields. (Note we assume the existing dscale is honest...)
*/
ddigits = (num->n_weight + 1) * DEC_DIGITS;
if (ddigits <= maxdigits && scale >= NUMERIC_DSCALE(num))
{
- new = (Numeric) palloc(num->varlen);
- memcpy(new, num, num->varlen);
+ new = (Numeric) palloc(VARSIZE(num));
+ memcpy(new, num, VARSIZE(num));
new->n_sign_dscale = NUMERIC_SIGN(new) |
((uint16) scale & NUMERIC_DSCALE_MASK);
PG_RETURN_NUMERIC(new);
PG_RETURN_NUMERIC(new);
}
+Datum
+numerictypmodin(PG_FUNCTION_ARGS)
+{
+ ArrayType *ta = PG_GETARG_ARRAYTYPE_P(0);
+ int32 *tl;
+ int n;
+ int32 typmod;
+
+ tl = ArrayGetIntegerTypmods(ta, &n);
+
+ if (n == 2)
+ {
+ if (tl[0] < 1 || tl[0] > NUMERIC_MAX_PRECISION)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("NUMERIC precision %d must be between 1 and %d",
+ tl[0], NUMERIC_MAX_PRECISION)));
+ if (tl[1] < 0 || tl[1] > tl[0])
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("NUMERIC scale %d must be between 0 and precision %d",
+ tl[1], tl[0])));
+ typmod = ((tl[0] << 16) | tl[1]) + VARHDRSZ;
+ }
+ else if (n == 1)
+ {
+ if (tl[0] < 1 || tl[0] > NUMERIC_MAX_PRECISION)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("NUMERIC precision %d must be between 1 and %d",
+ tl[0], NUMERIC_MAX_PRECISION)));
+ /* scale defaults to zero */
+ typmod = (tl[0] << 16) + VARHDRSZ;
+ }
+ else
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("invalid NUMERIC type modifier")));
+ typmod = 0; /* keep compiler quiet */
+ }
+
+ PG_RETURN_INT32(typmod);
+}
+
+Datum
+numerictypmodout(PG_FUNCTION_ARGS)
+{
+ int32 typmod = PG_GETARG_INT32(0);
+ char *res = (char *) palloc(64);
+
+ if (typmod >= 0)
+ snprintf(res, 64, "(%d,%d)",
+ ((typmod - VARHDRSZ) >> 16) & 0xffff,
+ (typmod - VARHDRSZ) & 0xffff);
+ else
+ *res = '\0';
+
+ PG_RETURN_CSTRING(res);
+}
+
/* ----------------------------------------------------------------------
*
/*
* Do it the easy way directly on the packed format
*/
- res = (Numeric) palloc(num->varlen);
- memcpy(res, num, num->varlen);
+ res = (Numeric) palloc(VARSIZE(num));
+ memcpy(res, num, VARSIZE(num));
res->n_sign_dscale = NUMERIC_POS | NUMERIC_DSCALE(num);
/*
* Do it the easy way directly on the packed format
*/
- res = (Numeric) palloc(num->varlen);
- memcpy(res, num, num->varlen);
+ res = (Numeric) palloc(VARSIZE(num));
+ memcpy(res, num, VARSIZE(num));
/*
- * The packed format is known to be totally zero digit trimmed always.
- * So we can identify a ZERO by the fact that there are no digits at
- * all. Do nothing to a zero.
+ * The packed format is known to be totally zero digit trimmed always. So
+ * we can identify a ZERO by the fact that there are no digits at all. Do
+ * nothing to a zero.
*/
- if (num->varlen != NUMERIC_HDRSZ)
+ if (VARSIZE(num) != NUMERIC_HDRSZ)
{
/* Else, flip the sign */
if (NUMERIC_SIGN(num) == NUMERIC_POS)
Numeric num = PG_GETARG_NUMERIC(0);
Numeric res;
- res = (Numeric) palloc(num->varlen);
- memcpy(res, num, num->varlen);
+ res = (Numeric) palloc(VARSIZE(num));
+ memcpy(res, num, VARSIZE(num));
PG_RETURN_NUMERIC(res);
}
init_var(&result);
/*
- * The packed format is known to be totally zero digit trimmed always.
- * So we can identify a ZERO by the fact that there are no digits at
- * all.
+ * The packed format is known to be totally zero digit trimmed always. So
+ * we can identify a ZERO by the fact that there are no digits at all.
*/
- if (num->varlen == NUMERIC_HDRSZ)
+ if (VARSIZE(num) == NUMERIC_HDRSZ)
set_var_from_var(&const_zero, &result);
else
{
/*
- * And if there are some, we return a copy of ONE with the sign of
- * our argument
+ * And if there are some, we return a copy of ONE with the sign of our
+ * argument
*/
set_var_from_var(&const_one, &result);
result.sign = NUMERIC_SIGN(num);
}
/*
- * width_bucket_numeric() -
+ * Implements the numeric version of the width_bucket() function
+ * defined by SQL2003. See also width_bucket_float8().
*
* 'bound1' and 'bound2' are the lower and upper bounds of the
* histogram's range, respectively. 'count' is the number of buckets
* in the histogram. width_bucket() returns an integer indicating the
- * bucket number that 'operand' belongs in for an equiwidth histogram
+ * bucket number that 'operand' belongs to in an equiwidth histogram
* with the specified characteristics. An operand smaller than the
* lower bound is assigned to bucket 0. An operand greater than the
* upper bound is assigned to an additional bucket (with number
- * count+1).
+ * count+1). We don't allow "NaN" for any of the numeric arguments.
*/
Datum
width_bucket_numeric(PG_FUNCTION_ARGS)
if (count <= 0)
ereport(ERROR,
- (errcode(ERRCODE_INVALID_ARGUMENT_FOR_WIDTH_BUCKET_FUNCTION),
- errmsg("count must be greater than zero")));
+ (errcode(ERRCODE_INVALID_ARGUMENT_FOR_WIDTH_BUCKET_FUNCTION),
+ errmsg("count must be greater than zero")));
+
+ if (NUMERIC_IS_NAN(operand) ||
+ NUMERIC_IS_NAN(bound1) ||
+ NUMERIC_IS_NAN(bound2))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_ARGUMENT_FOR_WIDTH_BUCKET_FUNCTION),
+ errmsg("operand, lower bound and upper bound cannot be NaN")));
init_var(&result_var);
init_var(&count_var);
{
case 0:
ereport(ERROR,
- (errcode(ERRCODE_INVALID_ARGUMENT_FOR_WIDTH_BUCKET_FUNCTION),
- errmsg("lower bound cannot equal upper bound")));
+ (errcode(ERRCODE_INVALID_ARGUMENT_FOR_WIDTH_BUCKET_FUNCTION),
+ errmsg("lower bound cannot equal upper bound")));
/* bound1 < bound2 */
case -1:
break;
}
+ /* if result exceeds the range of a legal int4, we ereport here */
result = numericvar_to_int4(&result_var);
free_var(&count_var);
}
/*
- * compute_bucket() -
- *
* If 'operand' is not outside the bucket range, determine the correct
* bucket for it to go. The calculations performed by this function
* are derived directly from the SQL2003 spec.
sub_var(&operand_var, &bound1_var, &operand_var);
sub_var(&bound2_var, &bound1_var, &bound2_var);
div_var(&operand_var, &bound2_var, result_var,
- select_div_scale(&operand_var, &bound2_var));
+ select_div_scale(&operand_var, &bound2_var), true);
}
else
{
sub_var(&bound1_var, &operand_var, &operand_var);
sub_var(&bound1_var, &bound2_var, &bound1_var);
div_var(&operand_var, &bound1_var, result_var,
- select_div_scale(&operand_var, &bound1_var));
+ select_div_scale(&operand_var, &bound1_var), true);
}
mul_var(result_var, count_var, result_var,
int result;
/*
- * We consider all NANs to be equal and larger than any non-NAN. This
- * is somewhat arbitrary; the important thing is to have a consistent
- * sort order.
+ * We consider all NANs to be equal and larger than any non-NAN. This is
+ * somewhat arbitrary; the important thing is to have a consistent sort
+ * order.
*/
if (NUMERIC_IS_NAN(num1))
{
}
else
{
- NumericVar arg1;
- NumericVar arg2;
+ result = cmp_var_common(NUMERIC_DIGITS(num1), NUMERIC_NDIGITS(num1),
+ num1->n_weight, NUMERIC_SIGN(num1),
+ NUMERIC_DIGITS(num2), NUMERIC_NDIGITS(num2),
+ num2->n_weight, NUMERIC_SIGN(num2));
+ }
- init_var(&arg1);
- init_var(&arg2);
+ return result;
+}
- set_var_from_num(num1, &arg1);
- set_var_from_num(num2, &arg2);
+Datum
+hash_numeric(PG_FUNCTION_ARGS)
+{
+ Numeric key = PG_GETARG_NUMERIC(0);
+ Datum digit_hash;
+ Datum result;
+ int weight;
+ int start_offset;
+ int end_offset;
+ int i;
+ int hash_len;
+
+ /* If it's NaN, don't try to hash the rest of the fields */
+ if (NUMERIC_IS_NAN(key))
+ PG_RETURN_UINT32(0);
+
+ weight = key->n_weight;
+ start_offset = 0;
+ end_offset = 0;
- result = cmp_var(&arg1, &arg2);
+ /*
+ * Omit any leading or trailing zeros from the input to the hash. The
+ * numeric implementation *should* guarantee that leading and trailing
+ * zeros are suppressed, but we're paranoid. Note that we measure the
+ * starting and ending offsets in units of NumericDigits, not bytes.
+ */
+ for (i = 0; i < NUMERIC_NDIGITS(key); i++)
+ {
+ if (NUMERIC_DIGITS(key)[i] != (NumericDigit) 0)
+ break;
+
+ start_offset++;
- free_var(&arg1);
- free_var(&arg2);
+ /*
+ * The weight is effectively the # of digits before the decimal point,
+ * so decrement it for each leading zero we skip.
+ */
+ weight--;
}
- return result;
+ /*
+ * If there are no non-zero digits, then the value of the number is zero,
+ * regardless of any other fields.
+ */
+ if (NUMERIC_NDIGITS(key) == start_offset)
+ PG_RETURN_UINT32(-1);
+
+ for (i = NUMERIC_NDIGITS(key) - 1; i >= 0; i--)
+ {
+ if (NUMERIC_DIGITS(key)[i] != (NumericDigit) 0)
+ break;
+
+ end_offset++;
+ }
+
+ /* If we get here, there should be at least one non-zero digit */
+ Assert(start_offset + end_offset < NUMERIC_NDIGITS(key));
+
+ /*
+ * Note that we don't hash on the Numeric's scale, since two numerics can
+ * compare equal but have different scales. We also don't hash on the
+ * sign, although we could: since a sign difference implies inequality,
+ * this shouldn't affect correctness.
+ */
+ hash_len = NUMERIC_NDIGITS(key) - start_offset - end_offset;
+ digit_hash = hash_any((unsigned char *) (NUMERIC_DIGITS(key) + start_offset),
+ hash_len * sizeof(NumericDigit));
+
+ /* Mix in the weight, via XOR */
+ result = digit_hash ^ weight;
+
+ PG_RETURN_DATUM(result);
}
/*
* Unpack the values, let mul_var() compute the result and return it.
- * Unlike add_var() and sub_var(), mul_var() will round its result. In
- * the case of numeric_mul(), which is invoked for the * operator on
- * numerics, we request exact representation for the product (rscale =
- * sum(dscale of arg1, dscale of arg2)).
+ * Unlike add_var() and sub_var(), mul_var() will round its result. In the
+ * case of numeric_mul(), which is invoked for the * operator on numerics,
+ * we request exact representation for the product (rscale = sum(dscale of
+ * arg1, dscale of arg2)).
*/
init_var(&arg1);
init_var(&arg2);
/*
* Do the divide and return the result
*/
- div_var(&arg1, &arg2, &result, rscale);
+ div_var(&arg1, &arg2, &result, rscale, true);
+
+ res = make_result(&result);
+
+ free_var(&arg1);
+ free_var(&arg2);
+ free_var(&result);
+
+ PG_RETURN_NUMERIC(res);
+}
+
+
+/*
+ * numeric_div_trunc() -
+ *
+ * Divide one numeric into another, truncating the result to an integer
+ */
+Datum
+numeric_div_trunc(PG_FUNCTION_ARGS)
+{
+ Numeric num1 = PG_GETARG_NUMERIC(0);
+ Numeric num2 = PG_GETARG_NUMERIC(1);
+ NumericVar arg1;
+ NumericVar arg2;
+ NumericVar result;
+ Numeric res;
+
+ /*
+ * Handle NaN
+ */
+ if (NUMERIC_IS_NAN(num1) || NUMERIC_IS_NAN(num2))
+ PG_RETURN_NUMERIC(make_result(&const_nan));
+
+ /*
+ * Unpack the arguments
+ */
+ init_var(&arg1);
+ init_var(&arg2);
+ init_var(&result);
+
+ set_var_from_num(num1, &arg1);
+ set_var_from_num(num2, &arg2);
+
+ /*
+ * Do the divide and return the result
+ */
+ div_var(&arg1, &arg2, &result, 0, false);
res = make_result(&result);
Numeric num2 = PG_GETARG_NUMERIC(1);
/*
- * Use cmp_numerics so that this will agree with the comparison
- * operators, particularly as regards comparisons involving NaN.
+ * Use cmp_numerics so that this will agree with the comparison operators,
+ * particularly as regards comparisons involving NaN.
*/
if (cmp_numerics(num1, num2) < 0)
PG_RETURN_NUMERIC(num1);
Numeric num2 = PG_GETARG_NUMERIC(1);
/*
- * Use cmp_numerics so that this will agree with the comparison
- * operators, particularly as regards comparisons involving NaN.
+ * Use cmp_numerics so that this will agree with the comparison operators,
+ * particularly as regards comparisons involving NaN.
*/
if (cmp_numerics(num1, num2) > 0)
PG_RETURN_NUMERIC(num1);
res = make_result(&const_one);
PG_RETURN_NUMERIC(res);
}
+ /* Fail immediately if the result would overflow */
+ if (num > 32177)
+ ereport(ERROR,
+ (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+ errmsg("value overflows numeric format")));
init_var(&fact);
init_var(&result);
for (num = num - 1; num > 1; num--)
{
+ /* this loop can take awhile, so allow it to be interrupted */
+ CHECK_FOR_INTERRUPTS();
+
int8_to_numericvar(num, &fact);
mul_var(&result, &fact, &result, 0);
PG_RETURN_NUMERIC(make_result(&const_nan));
/*
- * Unpack the argument and determine the result scale. We choose a
- * scale to give at least NUMERIC_MIN_SIG_DIGITS significant digits;
- * but in any case not less than the input's dscale.
+ * Unpack the argument and determine the result scale. We choose a scale
+ * to give at least NUMERIC_MIN_SIG_DIGITS significant digits; but in any
+ * case not less than the input's dscale.
*/
init_var(&arg);
init_var(&result);
PG_RETURN_NUMERIC(make_result(&const_nan));
/*
- * Unpack the argument and determine the result scale. We choose a
- * scale to give at least NUMERIC_MIN_SIG_DIGITS significant digits;
- * but in any case not less than the input's dscale.
+ * Unpack the argument and determine the result scale. We choose a scale
+ * to give at least NUMERIC_MIN_SIG_DIGITS significant digits; but in any
+ * case not less than the input's dscale.
*/
init_var(&arg);
init_var(&result);
val = numericvar_to_double_no_overflow(&arg);
/*
- * log10(result) = num * log10(e), so this is approximately the
- * decimal weight of the result:
+ * log10(result) = num * log10(e), so this is approximately the decimal
+ * weight of the result:
*/
val *= 0.434294481903252;
set_var_from_num(num2, &arg2);
/*
- * Call log_var() to compute and return the result; note it handles
- * scale selection itself.
+ * Call log_var() to compute and return the result; note it handles scale
+ * selection itself.
*/
log_var(&arg1, &arg2, &result);
trunc_var(&arg2_trunc, 0);
/*
- * Return special SQLSTATE error codes for a few conditions mandated
- * by the standard.
+ * The SQL spec requires that we emit a particular SQLSTATE error code for
+ * certain error conditions. Specifically, we don't return a
+ * divide-by-zero error code for 0 ^ -1.
*/
- if ((cmp_var(&arg1, &const_zero) == 0 &&
- cmp_var(&arg2, &const_zero) < 0) ||
- (cmp_var(&arg1, &const_zero) < 0 &&
- cmp_var(&arg2, &arg2_trunc) != 0))
+ if (cmp_var(&arg1, &const_zero) == 0 &&
+ cmp_var(&arg2, &const_zero) < 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_ARGUMENT_FOR_POWER_FUNCTION),
+ errmsg("zero raised to a negative power is undefined")));
+
+ if (cmp_var(&arg1, &const_zero) < 0 &&
+ cmp_var(&arg2, &arg2_trunc) != 0)
ereport(ERROR,
(errcode(ERRCODE_INVALID_ARGUMENT_FOR_POWER_FUNCTION),
- errmsg("invalid argument for power function")));
+ errmsg("a negative number raised to a non-integer power yields a complex result")));
/*
* Call power_var() to compute and return the result; note it handles
if (NUMERIC_IS_NAN(num))
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("cannot convert NaN to integer")));
+ errmsg("cannot convert NaN to bigint")));
/* Convert to variable format and thence to int8 */
init_var(&x);
if (!numericvar_to_int8(&x, &result))
ereport(ERROR,
(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
- errmsg("integer out of range")));
+ errmsg("bigint out of range")));
free_var(&x);
if (NUMERIC_IS_NAN(num))
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("cannot convert NaN to integer")));
+ errmsg("cannot convert NaN to smallint")));
/* Convert to variable format and thence to int8 */
init_var(&x);
if (!numericvar_to_int8(&x, &val))
ereport(ERROR,
(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
- errmsg("integer out of range")));
+ errmsg("smallint out of range")));
free_var(&x);
if ((int64) result != val)
ereport(ERROR,
(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
- errmsg("integer out of range")));
+ errmsg("smallint out of range")));
PG_RETURN_INT16(result);
}
init_var(&result);
- set_var_from_str(buf, &result);
+ /* Assume we need not worry about leading/trailing spaces */
+ (void) set_var_from_str(buf, buf, &result);
+
res = make_result(&result);
free_var(&result);
init_var(&result);
- set_var_from_str(buf, &result);
+ /* Assume we need not worry about leading/trailing spaces */
+ (void) set_var_from_str(buf, buf, &result);
+
res = make_result(&result);
free_var(&result);
}
-Datum
-text_numeric(PG_FUNCTION_ARGS)
-{
- text *str = PG_GETARG_TEXT_P(0);
- int len;
- char *s;
- Datum result;
-
- len = (VARSIZE(str) - VARHDRSZ);
- s = palloc(len + 1);
- memcpy(s, VARDATA(str), len);
- *(s + len) = '\0';
-
- result = DirectFunctionCall3(numeric_in, CStringGetDatum(s),
- ObjectIdGetDatum(0), Int32GetDatum(-1));
-
- pfree(s);
-
- return result;
-}
-
-Datum
-numeric_text(PG_FUNCTION_ARGS)
-{
- /* val is numeric, but easier to leave it as Datum */
- Datum val = PG_GETARG_DATUM(0);
- char *s;
- int len;
- text *result;
-
- s = DatumGetCString(DirectFunctionCall1(numeric_out, val));
- len = strlen(s);
-
- result = (text *) palloc(VARHDRSZ + len);
-
- VARATT_SIZEP(result) = len + VARHDRSZ;
- memcpy(VARDATA(result), s, len);
-
- pfree(s);
-
- PG_RETURN_TEXT_P(result);
-}
-
-
/* ----------------------------------------------------------------------
*
* Aggregate functions
/* We assume the input is array of numeric */
deconstruct_array(transarray,
NUMERICOID, -1, false, 'i',
- &transdatums, &ndatums);
+ &transdatums, NULL, &ndatums);
if (ndatums != 3)
elog(ERROR, "expected 3-element numeric array");
N = transdatums[0];
NumericGetDatum(newval));
sumX2 = DirectFunctionCall2(numeric_add, sumX2,
DirectFunctionCall2(numeric_mul,
- NumericGetDatum(newval),
- NumericGetDatum(newval)));
+ NumericGetDatum(newval),
+ NumericGetDatum(newval)));
transdatums[0] = N;
transdatums[1] = sumX;
return result;
}
+/*
+ * Improve avg performance by not caclulating sum(X*X).
+ */
+static ArrayType *
+do_numeric_avg_accum(ArrayType *transarray, Numeric newval)
+{
+ Datum *transdatums;
+ int ndatums;
+ Datum N,
+ sumX;
+ ArrayType *result;
+
+ /* We assume the input is array of numeric */
+ deconstruct_array(transarray,
+ NUMERICOID, -1, false, 'i',
+ &transdatums, NULL, &ndatums);
+ if (ndatums != 2)
+ elog(ERROR, "expected 2-element numeric array");
+ N = transdatums[0];
+ sumX = transdatums[1];
+
+ N = DirectFunctionCall1(numeric_inc, N);
+ sumX = DirectFunctionCall2(numeric_add, sumX,
+ NumericGetDatum(newval));
+
+ transdatums[0] = N;
+ transdatums[1] = sumX;
+
+ result = construct_array(transdatums, 2,
+ NUMERICOID, -1, false, 'i');
+
+ return result;
+}
+
Datum
numeric_accum(PG_FUNCTION_ARGS)
{
PG_RETURN_ARRAYTYPE_P(do_numeric_accum(transarray, newval));
}
+/*
+ * Optimized case for average of numeric.
+ */
+Datum
+numeric_avg_accum(PG_FUNCTION_ARGS)
+{
+ ArrayType *transarray = PG_GETARG_ARRAYTYPE_P(0);
+ Numeric newval = PG_GETARG_NUMERIC(1);
+
+ PG_RETURN_ARRAYTYPE_P(do_numeric_avg_accum(transarray, newval));
+}
+
/*
* Integer data types all use Numeric accumulators to share code and
* avoid risk of overflow. For int2 and int4 inputs, Numeric accumulation
PG_RETURN_ARRAYTYPE_P(do_numeric_accum(transarray, newval));
}
+/*
+ * Optimized case for average of int8.
+ */
+Datum
+int8_avg_accum(PG_FUNCTION_ARGS)
+{
+ ArrayType *transarray = PG_GETARG_ARRAYTYPE_P(0);
+ Datum newval8 = PG_GETARG_DATUM(1);
+ Numeric newval;
+
+ newval = DatumGetNumeric(DirectFunctionCall1(int8_numeric, newval8));
+
+ PG_RETURN_ARRAYTYPE_P(do_numeric_avg_accum(transarray, newval));
+}
+
+
Datum
numeric_avg(PG_FUNCTION_ARGS)
{
/* We assume the input is array of numeric */
deconstruct_array(transarray,
NUMERICOID, -1, false, 'i',
- &transdatums, &ndatums);
- if (ndatums != 3)
- elog(ERROR, "expected 3-element numeric array");
+ &transdatums, NULL, &ndatums);
+ if (ndatums != 2)
+ elog(ERROR, "expected 2-element numeric array");
N = DatumGetNumeric(transdatums[0]);
sumX = DatumGetNumeric(transdatums[1]);
- /* ignore sumX2 */
/* SQL92 defines AVG of no values to be NULL */
/* N is zero iff no digits (cf. numeric_uminus) */
- if (N->varlen == NUMERIC_HDRSZ)
+ if (VARSIZE(N) == NUMERIC_HDRSZ)
PG_RETURN_NULL();
PG_RETURN_DATUM(DirectFunctionCall2(numeric_div,
NumericGetDatum(N)));
}
-Datum
-numeric_variance(PG_FUNCTION_ARGS)
+/*
+ * Workhorse routine for the standard deviance and variance
+ * aggregates. 'transarray' is the aggregate's transition
+ * array. 'variance' specifies whether we should calculate the
+ * variance or the standard deviation. 'sample' indicates whether the
+ * caller is interested in the sample or the population
+ * variance/stddev.
+ *
+ * If appropriate variance statistic is undefined for the input,
+ * *is_null is set to true and NULL is returned.
+ */
+static Numeric
+numeric_stddev_internal(ArrayType *transarray,
+ bool variance, bool sample,
+ bool *is_null)
{
- ArrayType *transarray = PG_GETARG_ARRAYTYPE_P(0);
Datum *transdatums;
int ndatums;
Numeric N,
vsumX,
vsumX2,
vNminus1;
+ NumericVar *comp;
int rscale;
+ *is_null = false;
+
/* We assume the input is array of numeric */
deconstruct_array(transarray,
NUMERICOID, -1, false, 'i',
- &transdatums, &ndatums);
+ &transdatums, NULL, &ndatums);
if (ndatums != 3)
elog(ERROR, "expected 3-element numeric array");
N = DatumGetNumeric(transdatums[0]);
sumX2 = DatumGetNumeric(transdatums[2]);
if (NUMERIC_IS_NAN(N) || NUMERIC_IS_NAN(sumX) || NUMERIC_IS_NAN(sumX2))
- PG_RETURN_NUMERIC(make_result(&const_nan));
+ return make_result(&const_nan);
- /* Sample variance is undefined when N is 0 or 1, so return NULL */
init_var(&vN);
set_var_from_num(N, &vN);
- if (cmp_var(&vN, &const_one) <= 0)
+ /*
+ * Sample stddev and variance are undefined when N <= 1; population stddev
+ * is undefined when N == 0. Return NULL in either case.
+ */
+ if (sample)
+ comp = &const_one;
+ else
+ comp = &const_zero;
+
+ if (cmp_var(&vN, comp) <= 0)
{
free_var(&vN);
- PG_RETURN_NULL();
+ *is_null = true;
+ return NULL;
}
init_var(&vNminus1);
}
else
{
- mul_var(&vN, &vNminus1, &vNminus1, 0); /* N * (N - 1) */
+ if (sample)
+ mul_var(&vN, &vNminus1, &vNminus1, 0); /* N * (N - 1) */
+ else
+ mul_var(&vN, &vN, &vNminus1, 0); /* N * N */
rscale = select_div_scale(&vsumX2, &vNminus1);
- div_var(&vsumX2, &vNminus1, &vsumX, rscale); /* variance */
+ div_var(&vsumX2, &vNminus1, &vsumX, rscale, true); /* variance */
+ if (!variance)
+ sqrt_var(&vsumX, &vsumX, rscale); /* stddev */
res = make_result(&vsumX);
}
free_var(&vsumX);
free_var(&vsumX2);
- PG_RETURN_NUMERIC(res);
+ return res;
}
Datum
-numeric_stddev(PG_FUNCTION_ARGS)
+numeric_var_samp(PG_FUNCTION_ARGS)
{
- ArrayType *transarray = PG_GETARG_ARRAYTYPE_P(0);
- Datum *transdatums;
- int ndatums;
- Numeric N,
- sumX,
- sumX2,
- res;
- NumericVar vN,
- vsumX,
- vsumX2,
- vNminus1;
- int rscale;
-
- /* We assume the input is array of numeric */
- deconstruct_array(transarray,
- NUMERICOID, -1, false, 'i',
- &transdatums, &ndatums);
- if (ndatums != 3)
- elog(ERROR, "expected 3-element numeric array");
- N = DatumGetNumeric(transdatums[0]);
- sumX = DatumGetNumeric(transdatums[1]);
- sumX2 = DatumGetNumeric(transdatums[2]);
-
- if (NUMERIC_IS_NAN(N) || NUMERIC_IS_NAN(sumX) || NUMERIC_IS_NAN(sumX2))
- PG_RETURN_NUMERIC(make_result(&const_nan));
+ Numeric res;
+ bool is_null;
- /* Sample stddev is undefined when N is 0 or 1, so return NULL */
- init_var(&vN);
- set_var_from_num(N, &vN);
+ res = numeric_stddev_internal(PG_GETARG_ARRAYTYPE_P(0),
+ true, true, &is_null);
- if (cmp_var(&vN, &const_one) <= 0)
- {
- free_var(&vN);
+ if (is_null)
PG_RETURN_NULL();
- }
+ else
+ PG_RETURN_NUMERIC(res);
+}
- init_var(&vNminus1);
- sub_var(&vN, &const_one, &vNminus1);
+Datum
+numeric_stddev_samp(PG_FUNCTION_ARGS)
+{
+ Numeric res;
+ bool is_null;
- init_var(&vsumX);
- set_var_from_num(sumX, &vsumX);
- init_var(&vsumX2);
- set_var_from_num(sumX2, &vsumX2);
+ res = numeric_stddev_internal(PG_GETARG_ARRAYTYPE_P(0),
+ false, true, &is_null);
- /* compute rscale for mul_var calls */
- rscale = vsumX.dscale * 2;
+ if (is_null)
+ PG_RETURN_NULL();
+ else
+ PG_RETURN_NUMERIC(res);
+}
- mul_var(&vsumX, &vsumX, &vsumX, rscale); /* vsumX = sumX * sumX */
- mul_var(&vN, &vsumX2, &vsumX2, rscale); /* vsumX2 = N * sumX2 */
- sub_var(&vsumX2, &vsumX, &vsumX2); /* N * sumX2 - sumX * sumX */
+Datum
+numeric_var_pop(PG_FUNCTION_ARGS)
+{
+ Numeric res;
+ bool is_null;
- if (cmp_var(&vsumX2, &const_zero) <= 0)
- {
- /* Watch out for roundoff error producing a negative numerator */
- res = make_result(&const_zero);
- }
+ res = numeric_stddev_internal(PG_GETARG_ARRAYTYPE_P(0),
+ true, false, &is_null);
+
+ if (is_null)
+ PG_RETURN_NULL();
else
- {
- mul_var(&vN, &vNminus1, &vNminus1, 0); /* N * (N - 1) */
- rscale = select_div_scale(&vsumX2, &vNminus1);
- div_var(&vsumX2, &vNminus1, &vsumX, rscale); /* variance */
- sqrt_var(&vsumX, &vsumX, rscale); /* stddev */
+ PG_RETURN_NUMERIC(res);
+}
- res = make_result(&vsumX);
- }
+Datum
+numeric_stddev_pop(PG_FUNCTION_ARGS)
+{
+ Numeric res;
+ bool is_null;
- free_var(&vN);
- free_var(&vNminus1);
- free_var(&vsumX);
- free_var(&vsumX2);
+ res = numeric_stddev_internal(PG_GETARG_ARRAYTYPE_P(0),
+ false, false, &is_null);
- PG_RETURN_NUMERIC(res);
+ if (is_null)
+ PG_RETURN_NULL();
+ else
+ PG_RETURN_NUMERIC(res);
}
-
/*
* SUM transition functions for integer datatypes.
*
Datum
int2_sum(PG_FUNCTION_ARGS)
{
- int64 oldsum;
int64 newval;
if (PG_ARGISNULL(0))
PG_RETURN_INT64(newval);
}
- oldsum = PG_GETARG_INT64(0);
+ /*
+ * If we're invoked by nodeAgg, we can cheat and modify our first
+ * parameter in-place to avoid palloc overhead. If not, we need to return
+ * the new value of the transition variable. (If int8 is pass-by-value,
+ * then of course this is useless as well as incorrect, so just ifdef it
+ * out.)
+ */
+#ifndef USE_FLOAT8_BYVAL /* controls int8 too */
+ if (fcinfo->context &&
+ (IsA(fcinfo->context, AggState) ||
+ IsA(fcinfo->context, WindowAggState)))
+ {
+ int64 *oldsum = (int64 *) PG_GETARG_POINTER(0);
+
+ /* Leave the running sum unchanged in the new input is null */
+ if (!PG_ARGISNULL(1))
+ *oldsum = *oldsum + (int64) PG_GETARG_INT16(1);
- /* Leave sum unchanged if new input is null. */
- if (PG_ARGISNULL(1))
- PG_RETURN_INT64(oldsum);
+ PG_RETURN_POINTER(oldsum);
+ }
+ else
+#endif
+ {
+ int64 oldsum = PG_GETARG_INT64(0);
- /* OK to do the addition. */
- newval = oldsum + (int64) PG_GETARG_INT16(1);
+ /* Leave sum unchanged if new input is null. */
+ if (PG_ARGISNULL(1))
+ PG_RETURN_INT64(oldsum);
+
+ /* OK to do the addition. */
+ newval = oldsum + (int64) PG_GETARG_INT16(1);
- PG_RETURN_INT64(newval);
+ PG_RETURN_INT64(newval);
+ }
}
Datum
int4_sum(PG_FUNCTION_ARGS)
{
- int64 oldsum;
int64 newval;
if (PG_ARGISNULL(0))
PG_RETURN_INT64(newval);
}
- oldsum = PG_GETARG_INT64(0);
+ /*
+ * If we're invoked by nodeAgg, we can cheat and modify our first
+ * parameter in-place to avoid palloc overhead. If not, we need to return
+ * the new value of the transition variable. (If int8 is pass-by-value,
+ * then of course this is useless as well as incorrect, so just ifdef it
+ * out.)
+ */
+#ifndef USE_FLOAT8_BYVAL /* controls int8 too */
+ if (fcinfo->context &&
+ (IsA(fcinfo->context, AggState) ||
+ IsA(fcinfo->context, WindowAggState)))
+ {
+ int64 *oldsum = (int64 *) PG_GETARG_POINTER(0);
- /* Leave sum unchanged if new input is null. */
- if (PG_ARGISNULL(1))
- PG_RETURN_INT64(oldsum);
+ /* Leave the running sum unchanged in the new input is null */
+ if (!PG_ARGISNULL(1))
+ *oldsum = *oldsum + (int64) PG_GETARG_INT32(1);
+
+ PG_RETURN_POINTER(oldsum);
+ }
+ else
+#endif
+ {
+ int64 oldsum = PG_GETARG_INT64(0);
+
+ /* Leave sum unchanged if new input is null. */
+ if (PG_ARGISNULL(1))
+ PG_RETURN_INT64(oldsum);
- /* OK to do the addition. */
- newval = oldsum + (int64) PG_GETARG_INT32(1);
+ /* OK to do the addition. */
+ newval = oldsum + (int64) PG_GETARG_INT32(1);
- PG_RETURN_INT64(newval);
+ PG_RETURN_INT64(newval);
+ }
}
Datum
PG_RETURN_DATUM(newval);
}
+ /*
+ * Note that we cannot special-case the nodeAgg case here, as we do for
+ * int2_sum and int4_sum: numeric is of variable size, so we cannot modify
+ * our first parameter in-place.
+ */
+
oldsum = PG_GETARG_NUMERIC(0);
/* Leave sum unchanged if new input is null. */
typedef struct Int8TransTypeData
{
-#ifndef INT64_IS_BUSTED
int64 count;
int64 sum;
-#else
- /* "int64" isn't really 64 bits, so fake up properly-aligned fields */
- int32 count;
- int32 pad1;
- int32 sum;
- int32 pad2;
-#endif
} Int8TransTypeData;
Datum
int2_avg_accum(PG_FUNCTION_ARGS)
{
- ArrayType *transarray = PG_GETARG_ARRAYTYPE_P_COPY(0);
+ ArrayType *transarray;
int16 newval = PG_GETARG_INT16(1);
Int8TransTypeData *transdata;
/*
- * We copied the input array, so it's okay to scribble on it directly.
+ * If we're invoked by nodeAgg, we can cheat and modify our first
+ * parameter in-place to reduce palloc overhead. Otherwise we need to make
+ * a copy of it before scribbling on it.
*/
- if (ARR_SIZE(transarray) != ARR_OVERHEAD(1) + sizeof(Int8TransTypeData))
+ if (fcinfo->context &&
+ (IsA(fcinfo->context, AggState) ||
+ IsA(fcinfo->context, WindowAggState)))
+ transarray = PG_GETARG_ARRAYTYPE_P(0);
+ else
+ transarray = PG_GETARG_ARRAYTYPE_P_COPY(0);
+
+ if (ARR_HASNULL(transarray) ||
+ ARR_SIZE(transarray) != ARR_OVERHEAD_NONULLS(1) + sizeof(Int8TransTypeData))
elog(ERROR, "expected 2-element int8 array");
- transdata = (Int8TransTypeData *) ARR_DATA_PTR(transarray);
+ transdata = (Int8TransTypeData *) ARR_DATA_PTR(transarray);
transdata->count++;
transdata->sum += newval;
Datum
int4_avg_accum(PG_FUNCTION_ARGS)
{
- ArrayType *transarray = PG_GETARG_ARRAYTYPE_P_COPY(0);
+ ArrayType *transarray;
int32 newval = PG_GETARG_INT32(1);
Int8TransTypeData *transdata;
/*
- * We copied the input array, so it's okay to scribble on it directly.
+ * If we're invoked by nodeAgg, we can cheat and modify our first
+ * parameter in-place to reduce palloc overhead. Otherwise we need to make
+ * a copy of it before scribbling on it.
*/
- if (ARR_SIZE(transarray) != ARR_OVERHEAD(1) + sizeof(Int8TransTypeData))
+ if (fcinfo->context &&
+ (IsA(fcinfo->context, AggState) ||
+ IsA(fcinfo->context, WindowAggState)))
+ transarray = PG_GETARG_ARRAYTYPE_P(0);
+ else
+ transarray = PG_GETARG_ARRAYTYPE_P_COPY(0);
+
+ if (ARR_HASNULL(transarray) ||
+ ARR_SIZE(transarray) != ARR_OVERHEAD_NONULLS(1) + sizeof(Int8TransTypeData))
elog(ERROR, "expected 2-element int8 array");
- transdata = (Int8TransTypeData *) ARR_DATA_PTR(transarray);
+ transdata = (Int8TransTypeData *) ARR_DATA_PTR(transarray);
transdata->count++;
transdata->sum += newval;
Datum countd,
sumd;
- if (ARR_SIZE(transarray) != ARR_OVERHEAD(1) + sizeof(Int8TransTypeData))
+ if (ARR_HASNULL(transarray) ||
+ ARR_SIZE(transarray) != ARR_OVERHEAD_NONULLS(1) + sizeof(Int8TransTypeData))
elog(ERROR, "expected 2-element int8 array");
transdata = (Int8TransTypeData *) ARR_DATA_PTR(transarray);
static void
dump_numeric(const char *str, Numeric num)
{
- NumericDigit *digits = (NumericDigit *) num->n_data;
+ NumericDigit *digits = NUMERIC_DIGITS(num);
int ndigits;
int i;
- ndigits = (num->varlen - NUMERIC_HDRSZ) / sizeof(NumericDigit);
+ ndigits = NUMERIC_NDIGITS(num);
printf("%s: NUMERIC w=%d d=%d ", str, num->n_weight, NUMERIC_DSCALE(num));
switch (NUMERIC_SIGN(num))
* set_var_from_str()
*
* Parse a string and put the number into a variable
+ *
+ * This function does not handle leading or trailing spaces, and it doesn't
+ * accept "NaN" either. It returns the end+1 position so that caller can
+ * check for trailing spaces/garbage if deemed necessary.
+ *
+ * cp is the place to actually start parsing; str is what to use in error
+ * reports. (Typically cp would be the same except advanced over spaces.)
*/
-static void
-set_var_from_str(const char *str, NumericVar *dest)
+static const char *
+set_var_from_str(const char *str, const char *cp, NumericVar *dest)
{
- const char *cp = str;
bool have_dp = FALSE;
int i;
unsigned char *decdigits;
NumericDigit *digits;
/*
- * We first parse the string to extract decimal digits and determine
- * the correct decimal weight. Then convert to NBASE representation.
+ * We first parse the string to extract decimal digits and determine the
+ * correct decimal weight. Then convert to NBASE representation.
*/
-
- /* skip leading spaces */
- while (*cp)
- {
- if (!isspace((unsigned char) *cp))
- break;
- cp++;
- }
-
switch (*cp)
{
case '+':
if (!isdigit((unsigned char) *cp))
ereport(ERROR,
(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
- errmsg("invalid input syntax for type numeric: \"%s\"", str)));
+ errmsg("invalid input syntax for type numeric: \"%s\"", str)));
decdigits = (unsigned char *) palloc(strlen(cp) + DEC_DIGITS * 2);
if (have_dp)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
- errmsg("invalid input syntax for type numeric: \"%s\"",
- str)));
+ errmsg("invalid input syntax for type numeric: \"%s\"",
+ str)));
have_dp = TRUE;
cp++;
}
if (endptr == cp)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
- errmsg("invalid input syntax for type numeric: \"%s\"",
- str)));
+ errmsg("invalid input syntax for type numeric: \"%s\"",
+ str)));
cp = endptr;
if (exponent > NUMERIC_MAX_PRECISION ||
exponent < -NUMERIC_MAX_PRECISION)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
- errmsg("invalid input syntax for type numeric: \"%s\"",
- str)));
+ errmsg("invalid input syntax for type numeric: \"%s\"",
+ str)));
dweight += (int) exponent;
dscale -= (int) exponent;
if (dscale < 0)
dscale = 0;
}
- /* Should be nothing left but spaces */
- while (*cp)
- {
- if (!isspace((unsigned char) *cp))
- ereport(ERROR,
- (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
- errmsg("invalid input syntax for type numeric: \"%s\"",
- str)));
- cp++;
- }
-
/*
- * Okay, convert pure-decimal representation to base NBASE. First we
- * need to determine the converted weight and ndigits. offset is the
- * number of decimal zeroes to insert before the first given digit to
- * have a correctly aligned first NBASE digit.
+ * Okay, convert pure-decimal representation to base NBASE. First we need
+ * to determine the converted weight and ndigits. offset is the number of
+ * decimal zeroes to insert before the first given digit to have a
+ * correctly aligned first NBASE digit.
*/
if (dweight >= 0)
weight = (dweight + 1 + DEC_DIGITS - 1) / DEC_DIGITS - 1;
/* Strip any leading/trailing zeroes, and normalize weight if zero */
strip_var(dest);
+
+ /* Return end+1 position for caller */
+ return cp;
}
{
int ndigits;
- ndigits = (num->varlen - NUMERIC_HDRSZ) / sizeof(NumericDigit);
+ ndigits = NUMERIC_NDIGITS(num);
alloc_var(dest, ndigits);
* Allocate space for the result.
*
* i is set to to # of decimal digits before decimal point. dscale is the
- * # of decimal digits we will print after decimal point. We may
- * generate as many as DEC_DIGITS-1 excess digits at the end, and in
- * addition we need room for sign, decimal point, null terminator.
+ * # of decimal digits we will print after decimal point. We may generate
+ * as many as DEC_DIGITS-1 excess digits at the end, and in addition we
+ * need room for sign, decimal point, null terminator.
*/
i = (var->weight + 1) * DEC_DIGITS;
if (i <= 0)
}
/*
- * If requested, output a decimal point and all the digits that follow
- * it. We initially put out a multiple of DEC_DIGITS digits, then
- * truncate if needed.
+ * If requested, output a decimal point and all the digits that follow it.
+ * We initially put out a multiple of DEC_DIGITS digits, then truncate if
+ * needed.
*/
if (dscale > 0)
{
return str;
}
+/*
+ * get_str_from_var_sci() -
+ *
+ * Convert a var to a normalised scientific notation text representation.
+ * This function does the heavy lifting for numeric_out_sci().
+ *
+ * This notation has the general form a * 10^b, where a is known as the
+ * "significand" and b is known as the "exponent".
+ *
+ * Because we can't do superscript in ASCII (and because we want to copy
+ * printf's behaviour) we display the exponent using E notation, with a
+ * minimum of two exponent digits.
+ *
+ * For example, the value 1234 could be output as 1.2e+03.
+ *
+ * We assume that the exponent can fit into an int32.
+ *
+ * rscale is the number of decimal digits desired after the decimal point in
+ * the output, negative values will be treated as meaning zero.
+ *
+ * CAUTION: var's contents may be modified by rounding!
+ *
+ * Returns a palloc'd string.
+ */
+static char *
+get_str_from_var_sci(NumericVar *var, int rscale)
+{
+ int32 exponent;
+ NumericVar denominator;
+ NumericVar significand;
+ int denom_scale;
+ size_t len;
+ char *str;
+ char *sig_out;
+
+ if (rscale < 0)
+ rscale = 0;
+
+ /*
+ * Determine the exponent of this number in normalised form.
+ *
+ * This is the exponent required to represent the number with only one
+ * significant digit before the decimal place.
+ */
+ if (var->ndigits > 0)
+ {
+ exponent = (var->weight + 1) * DEC_DIGITS;
+
+ /*
+ * Compensate for leading decimal zeroes in the first numeric digit by
+ * decrementing the exponent.
+ */
+ exponent -= DEC_DIGITS - (int) log10(var->digits[0]);
+ }
+ else
+ {
+ /*
+ * If var has no digits, then it must be zero.
+ *
+ * Zero doesn't technically have a meaningful exponent in normalised
+ * notation, but we just display the exponent as zero for consistency
+ * of output.
+ */
+ exponent = 0;
+ }
+
+ /*
+ * The denominator is set to 10 raised to the power of the exponent.
+ *
+ * We then divide var by the denominator to get the significand, rounding
+ * to rscale decimal digits in the process.
+ */
+ if (exponent < 0)
+ denom_scale = -exponent;
+ else
+ denom_scale = 0;
+
+ init_var(&denominator);
+ init_var(&significand);
+
+ int8_to_numericvar((int64) 10, &denominator);
+ power_var_int(&denominator, exponent, &denominator, denom_scale);
+ div_var(var, &denominator, &significand, rscale, true);
+ sig_out = get_str_from_var(&significand, rscale);
+
+ free_var(&denominator);
+ free_var(&significand);
+
+ /*
+ * Allocate space for the result.
+ *
+ * In addition to the significand, we need room for the exponent decoration
+ * ("e"), the sign of the exponent, up to 10 digits for the exponent
+ * itself, and of course the null terminator.
+ */
+ len = strlen(sig_out) + 13;
+ str = palloc(len);
+ snprintf(str, len, "%se%+03d", sig_out, exponent);
+
+ pfree(sig_out);
+
+ return str;
+}
+
/*
* make_result() -
{
result = (Numeric) palloc(NUMERIC_HDRSZ);
- result->varlen = NUMERIC_HDRSZ;
+ SET_VARSIZE(result, NUMERIC_HDRSZ);
result->n_weight = 0;
result->n_sign_dscale = NUMERIC_NAN;
/* Build the result */
len = NUMERIC_HDRSZ + n * sizeof(NumericDigit);
result = (Numeric) palloc(len);
- result->varlen = len;
+ SET_VARSIZE(result, len);
result->n_weight = weight;
result->n_sign_dscale = sign | (var->dscale & NUMERIC_DSCALE_MASK);
/*
* Check for overflow - note we can't do this before rounding, because
- * rounding could raise the weight. Also note that the var's weight
- * could be inflated by leading zeroes, which will be stripped before
- * storage but perhaps might not have been yet. In any case, we must
- * recognize a true zero, whose weight doesn't mean anything.
+ * rounding could raise the weight. Also note that the var's weight could
+ * be inflated by leading zeroes, which will be stripped before storage
+ * but perhaps might not have been yet. In any case, we must recognize a
+ * true zero, whose weight doesn't mean anything.
*/
ddigits = (var->weight + 1) * DEC_DIGITS;
if (ddigits > maxdigits)
ereport(ERROR,
(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
errmsg("numeric field overflow"),
- errdetail("The absolute value is greater than or equal to 10^%d for field with precision %d, scale %d.",
- ddigits - 1, precision, scale)));
+ errdetail("A field with precision %d, scale %d must round to an absolute value less than %s%d.",
+ precision, scale,
+ /* Display 10^0 as 1 */
+ maxdigits ? "10^" : "",
+ maxdigits ? maxdigits : 1
+ )));
break;
}
ddigits -= DEC_DIGITS;
}
/*
- * For input like 10000000000, we must treat stripped digits as real.
- * So the loop assumes there are weight+1 digits before the decimal
- * point.
+ * For input like 10000000000, we must treat stripped digits as real. So
+ * the loop assumes there are weight+1 digits before the decimal point.
*/
weight = var->weight;
Assert(weight >= 0 && ndigits <= weight + 1);
/*
* The overflow check is a bit tricky because we want to accept
- * INT64_MIN, which will overflow the positive accumulator. We
- * can detect this case easily though because INT64_MIN is the
- * only nonzero value for which -val == val (on a two's complement
- * machine, anyway).
+ * INT64_MIN, which will overflow the positive accumulator. We can
+ * detect this case easily though because INT64_MIN is the only
+ * nonzero value for which -val == val (on a two's complement machine,
+ * anyway).
*/
if ((val / NBASE) != oldval) /* possible overflow? */
{
/* shouldn't happen ... */
ereport(ERROR,
(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
- errmsg("invalid input syntax for type double precision: \"%s\"",
- tmp)));
+ errmsg("invalid input syntax for type double precision: \"%s\"",
+ tmp)));
}
pfree(tmp);
/* shouldn't happen ... */
ereport(ERROR,
(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
- errmsg("invalid input syntax for type double precision: \"%s\"",
- tmp)));
+ errmsg("invalid input syntax for type double precision: \"%s\"",
+ tmp)));
}
pfree(tmp);
static int
cmp_var(NumericVar *var1, NumericVar *var2)
{
- if (var1->ndigits == 0)
+ return cmp_var_common(var1->digits, var1->ndigits,
+ var1->weight, var1->sign,
+ var2->digits, var2->ndigits,
+ var2->weight, var2->sign);
+}
+
+/*
+ * cmp_var_common() -
+ *
+ * Main routine of cmp_var(). This function can be used by both
+ * NumericVar and Numeric.
+ */
+static int
+cmp_var_common(const NumericDigit *var1digits, int var1ndigits,
+ int var1weight, int var1sign,
+ const NumericDigit *var2digits, int var2ndigits,
+ int var2weight, int var2sign)
+{
+ if (var1ndigits == 0)
{
- if (var2->ndigits == 0)
+ if (var2ndigits == 0)
return 0;
- if (var2->sign == NUMERIC_NEG)
+ if (var2sign == NUMERIC_NEG)
return 1;
return -1;
}
- if (var2->ndigits == 0)
+ if (var2ndigits == 0)
{
- if (var1->sign == NUMERIC_POS)
+ if (var1sign == NUMERIC_POS)
return 1;
return -1;
}
- if (var1->sign == NUMERIC_POS)
+ if (var1sign == NUMERIC_POS)
{
- if (var2->sign == NUMERIC_NEG)
+ if (var2sign == NUMERIC_NEG)
return 1;
- return cmp_abs(var1, var2);
+ return cmp_abs_common(var1digits, var1ndigits, var1weight,
+ var2digits, var2ndigits, var2weight);
}
- if (var2->sign == NUMERIC_POS)
+ if (var2sign == NUMERIC_POS)
return -1;
- return cmp_abs(var2, var1);
+ return cmp_abs_common(var2digits, var2ndigits, var2weight,
+ var1digits, var1ndigits, var1weight);
}
else
{
/*
- * var1 is positive, var2 is negative Must compare absolute
- * values
+ * var1 is positive, var2 is negative Must compare absolute values
*/
switch (cmp_abs(var1, var2))
{
/*
* Determine number of result digits to compute. If the exact result
- * would have more than rscale fractional digits, truncate the
- * computation with MUL_GUARD_DIGITS guard digits. We do that by
- * pretending that one or both inputs have fewer digits than they
- * really do.
+ * would have more than rscale fractional digits, truncate the computation
+ * with MUL_GUARD_DIGITS guard digits. We do that by pretending that one
+ * or both inputs have fewer digits than they really do.
*/
res_ndigits = var1ndigits + var2ndigits + 1;
maxdigits = res_weight + 1 + (rscale * DEC_DIGITS) + MUL_GUARD_DIGITS;
/*
* 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.
+ * INT_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.
- * To avoid overflow in maxdig itself, it actually represents the max
+ * threatens to exceed INT_MAX, we take the time to propagate carries. To
+ * avoid overflow in maxdig itself, it actually represents the max
* possible value divided by NBASE-1.
*/
dig = (int *) palloc0(res_ndigits * sizeof(int));
}
/*
- * Now we do a final carry propagation pass to normalize the result,
- * which we combine with storing the result digits into the output.
- * Note that this is still done at full precision w/guard digits.
+ * Now we do a final carry propagation pass to normalize the result, which
+ * we combine with storing the result digits into the output. Note that
+ * this is still done at full precision w/guard digits.
*/
alloc_var(result, res_ndigits);
res_digits = result->digits;
/*
* div_var() -
*
- * Division on variable level. Quotient of var1 / var2 is stored
- * in result. Result is rounded to no more than rscale fractional digits.
+ * Division on variable level. Quotient of var1 / var2 is stored in result.
+ * The quotient is figured to exactly rscale fractional digits.
+ * If round is true, it is rounded at the rscale'th digit; if false, it
+ * is truncated (towards zero) at that digit.
*/
static void
div_var(NumericVar *var1, NumericVar *var2, NumericVar *result,
- int rscale)
+ int rscale, bool round)
+{
+ int div_ndigits;
+ int res_ndigits;
+ int res_sign;
+ int res_weight;
+ int carry;
+ int borrow;
+ int divisor1;
+ int divisor2;
+ NumericDigit *dividend;
+ NumericDigit *divisor;
+ NumericDigit *res_digits;
+ int i;
+ int j;
+
+ /* copy these values into local vars for speed in inner loop */
+ int var1ndigits = var1->ndigits;
+ int var2ndigits = var2->ndigits;
+
+ /*
+ * First of all division by zero check; we must not be handed an
+ * unnormalized divisor.
+ */
+ if (var2ndigits == 0 || var2->digits[0] == 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_DIVISION_BY_ZERO),
+ errmsg("division by zero")));
+
+ /*
+ * Now result zero check
+ */
+ if (var1ndigits == 0)
+ {
+ zero_var(result);
+ result->dscale = rscale;
+ return;
+ }
+
+ /*
+ * Determine the result sign, weight and number of digits to calculate.
+ * The weight figured here is correct if the emitted quotient has no
+ * leading zero digits; otherwise strip_var() will fix things up.
+ */
+ if (var1->sign == var2->sign)
+ res_sign = NUMERIC_POS;
+ else
+ res_sign = NUMERIC_NEG;
+ res_weight = var1->weight - var2->weight;
+ /* The number of accurate result digits we need to produce: */
+ res_ndigits = res_weight + 1 + (rscale + DEC_DIGITS - 1) / DEC_DIGITS;
+ /* ... but always at least 1 */
+ res_ndigits = Max(res_ndigits, 1);
+ /* If rounding needed, figure one more digit to ensure correct result */
+ if (round)
+ res_ndigits++;
+
+ /*
+ * The working dividend normally requires res_ndigits + var2ndigits
+ * digits, but make it at least var1ndigits so we can load all of var1
+ * into it. (There will be an additional digit dividend[0] in the
+ * dividend space, but for consistency with Knuth's notation we don't
+ * count that in div_ndigits.)
+ */
+ div_ndigits = res_ndigits + var2ndigits;
+ div_ndigits = Max(div_ndigits, var1ndigits);
+
+ /*
+ * We need a workspace with room for the working dividend (div_ndigits+1
+ * digits) plus room for the possibly-normalized divisor (var2ndigits
+ * digits). It is convenient also to have a zero at divisor[0] with the
+ * actual divisor data in divisor[1 .. var2ndigits]. Transferring the
+ * digits into the workspace also allows us to realloc the result (which
+ * might be the same as either input var) before we begin the main loop.
+ * Note that we use palloc0 to ensure that divisor[0], dividend[0], and
+ * any additional dividend positions beyond var1ndigits, start out 0.
+ */
+ dividend = (NumericDigit *)
+ palloc0((div_ndigits + var2ndigits + 2) * sizeof(NumericDigit));
+ divisor = dividend + (div_ndigits + 1);
+ memcpy(dividend + 1, var1->digits, var1ndigits * sizeof(NumericDigit));
+ memcpy(divisor + 1, var2->digits, var2ndigits * sizeof(NumericDigit));
+
+ /*
+ * Now we can realloc the result to hold the generated quotient digits.
+ */
+ alloc_var(result, res_ndigits);
+ res_digits = result->digits;
+
+ if (var2ndigits == 1)
+ {
+ /*
+ * If there's only a single divisor digit, we can use a fast path (cf.
+ * Knuth section 4.3.1 exercise 16).
+ */
+ divisor1 = divisor[1];
+ carry = 0;
+ for (i = 0; i < res_ndigits; i++)
+ {
+ carry = carry * NBASE + dividend[i + 1];
+ res_digits[i] = carry / divisor1;
+ carry = carry % divisor1;
+ }
+ }
+ else
+ {
+ /*
+ * The full multiple-place algorithm is taken from Knuth volume 2,
+ * Algorithm 4.3.1D.
+ *
+ * We need the first divisor digit to be >= NBASE/2. If it isn't,
+ * make it so by scaling up both the divisor and dividend by the
+ * factor "d". (The reason for allocating dividend[0] above is to
+ * leave room for possible carry here.)
+ */
+ if (divisor[1] < HALF_NBASE)
+ {
+ int d = NBASE / (divisor[1] + 1);
+
+ carry = 0;
+ for (i = var2ndigits; i > 0; i--)
+ {
+ carry += divisor[i] * d;
+ divisor[i] = carry % NBASE;
+ carry = carry / NBASE;
+ }
+ Assert(carry == 0);
+ carry = 0;
+ /* at this point only var1ndigits of dividend can be nonzero */
+ for (i = var1ndigits; i >= 0; i--)
+ {
+ carry += dividend[i] * d;
+ dividend[i] = carry % NBASE;
+ carry = carry / NBASE;
+ }
+ Assert(carry == 0);
+ Assert(divisor[1] >= HALF_NBASE);
+ }
+ /* First 2 divisor digits are used repeatedly in main loop */
+ divisor1 = divisor[1];
+ divisor2 = divisor[2];
+
+ /*
+ * Begin the main loop. Each iteration of this loop produces the j'th
+ * quotient digit by dividing dividend[j .. j + var2ndigits] by the
+ * divisor; this is essentially the same as the common manual
+ * procedure for long division.
+ */
+ for (j = 0; j < res_ndigits; j++)
+ {
+ /* Estimate quotient digit from the first two dividend digits */
+ int next2digits = dividend[j] * NBASE + dividend[j + 1];
+ int qhat;
+
+ /*
+ * If next2digits are 0, then quotient digit must be 0 and there's
+ * no need to adjust the working dividend. It's worth testing
+ * here to fall out ASAP when processing trailing zeroes in a
+ * dividend.
+ */
+ if (next2digits == 0)
+ {
+ res_digits[j] = 0;
+ continue;
+ }
+
+ if (dividend[j] == divisor1)
+ qhat = NBASE - 1;
+ else
+ qhat = next2digits / divisor1;
+
+ /*
+ * Adjust quotient digit if it's too large. Knuth proves that
+ * after this step, the quotient digit will be either correct or
+ * just one too large. (Note: it's OK to use dividend[j+2] here
+ * because we know the divisor length is at least 2.)
+ */
+ while (divisor2 * qhat >
+ (next2digits - qhat * divisor1) * NBASE + dividend[j + 2])
+ qhat--;
+
+ /* As above, need do nothing more when quotient digit is 0 */
+ if (qhat > 0)
+ {
+ /*
+ * Multiply the divisor by qhat, and subtract that from the
+ * working dividend. "carry" tracks the multiplication,
+ * "borrow" the subtraction (could we fold these together?)
+ */
+ carry = 0;
+ borrow = 0;
+ for (i = var2ndigits; i >= 0; i--)
+ {
+ carry += divisor[i] * qhat;
+ borrow -= carry % NBASE;
+ carry = carry / NBASE;
+ borrow += dividend[j + i];
+ if (borrow < 0)
+ {
+ dividend[j + i] = borrow + NBASE;
+ borrow = -1;
+ }
+ else
+ {
+ dividend[j + i] = borrow;
+ borrow = 0;
+ }
+ }
+ Assert(carry == 0);
+
+ /*
+ * If we got a borrow out of the top dividend digit, then
+ * indeed qhat was one too large. Fix it, and add back the
+ * divisor to correct the working dividend. (Knuth proves
+ * that this will occur only about 3/NBASE of the time; hence,
+ * it's a good idea to test this code with small NBASE to be
+ * sure this section gets exercised.)
+ */
+ if (borrow)
+ {
+ qhat--;
+ carry = 0;
+ for (i = var2ndigits; i >= 0; i--)
+ {
+ carry += dividend[j + i] + divisor[i];
+ if (carry >= NBASE)
+ {
+ dividend[j + i] = carry - NBASE;
+ carry = 1;
+ }
+ else
+ {
+ dividend[j + i] = carry;
+ carry = 0;
+ }
+ }
+ /* A carry should occur here to cancel the borrow above */
+ Assert(carry == 1);
+ }
+ }
+
+ /* And we're done with this quotient digit */
+ res_digits[j] = qhat;
+ }
+ }
+
+ pfree(dividend);
+
+ /*
+ * Finally, round or truncate the result to the requested precision.
+ */
+ result->weight = res_weight;
+ result->sign = res_sign;
+
+ /* Round or truncate to target rscale (and set result->dscale) */
+ if (round)
+ round_var(result, rscale);
+ else
+ trunc_var(result, rscale);
+
+ /* Strip leading and trailing zeroes */
+ strip_var(result);
+}
+
+
+/*
+ * div_var_fast() -
+ *
+ * This has the same API as div_var, but is implemented using the division
+ * algorithm from the "FM" library, rather than Knuth's schoolbook-division
+ * approach. This is significantly faster but can produce inaccurate
+ * results, because it sometimes has to propagate rounding to the left,
+ * and so we can never be entirely sure that we know the requested digits
+ * exactly. We compute DIV_GUARD_DIGITS extra digits, but there is
+ * no certainty that that's enough. We use this only in the transcendental
+ * function calculation routines, where everything is approximate anyway.
+ */
+static void
+div_var_fast(NumericVar *var1, NumericVar *var2, NumericVar *result,
+ int rscale, bool round)
{
int div_ndigits;
int res_sign;
/*
* We do the arithmetic in an array "div[]" of signed int's. Since
- * INT_MAX is noticeably larger than NBASE*NBASE, this gives us
- * headroom to avoid normalizing carries immediately.
+ * INT_MAX is noticeably larger than NBASE*NBASE, this gives us headroom
+ * to avoid normalizing carries immediately.
*
* We start with div[] containing one zero digit followed by the
- * dividend's digits (plus appended zeroes to reach the desired
- * precision including guard digits). Each step of the main loop
- * computes an (approximate) quotient digit and stores it into div[],
- * removing one position of dividend space. A final pass of carry
- * propagation takes care of any mistaken quotient digits.
+ * dividend's digits (plus appended zeroes to reach the desired precision
+ * including guard digits). Each step of the main loop computes an
+ * (approximate) quotient digit and stores it into div[], removing one
+ * position of dividend space. A final pass of carry propagation takes
+ * care of any mistaken quotient digits.
*/
div = (int *) palloc0((div_ndigits + 1) * sizeof(int));
for (i = 0; i < var1ndigits; i++)
div[i + 1] = var1digits[i];
/*
- * We estimate each quotient digit using floating-point arithmetic,
- * taking the first four digits of the (current) dividend and divisor.
- * This must be float to avoid overflow.
+ * We estimate each quotient digit using floating-point arithmetic, taking
+ * the first four digits of the (current) dividend and divisor. This must
+ * be float to avoid overflow.
*/
fdivisor = (double) var2digits[0];
for (i = 1; i < 4; i++)
fdivisorinverse = 1.0 / fdivisor;
/*
- * maxdiv tracks the maximum possible absolute value of any div[]
- * entry; when this threatens to exceed INT_MAX, we take the time to
- * propagate carries. To avoid overflow in maxdiv itself, it actually
- * represents the max possible abs. value divided by NBASE-1.
+ * maxdiv tracks the maximum possible absolute value of any div[] entry;
+ * when this threatens to exceed INT_MAX, we take the time to propagate
+ * carries. To avoid overflow in maxdiv itself, it actually represents
+ * the max possible abs. value divided by NBASE-1.
*/
maxdiv = 1;
div[qi] = newdig;
/*
- * All the div[] digits except possibly div[qi] are now in
- * the range 0..NBASE-1.
+ * All the div[] digits except possibly div[qi] are now in the
+ * range 0..NBASE-1.
*/
maxdiv = Abs(newdig) / (NBASE - 1);
maxdiv = Max(maxdiv, 1);
/* Compute the (approximate) quotient digit */
fquotient = fdividend * fdivisorinverse;
qdigit = (fquotient >= 0.0) ? ((int) fquotient) :
- (((int) fquotient) - 1); /* truncate towards
- * -infinity */
+ (((int) fquotient) - 1); /* truncate towards -infinity */
maxdiv += Abs(qdigit);
}
}
/*
- * The dividend digit we are about to replace might still be
- * nonzero. Fold it into the next digit position. We don't need
- * to worry about overflow here since this should nearly cancel
- * with the subtraction of the divisor.
+ * The dividend digit we are about to replace might still be nonzero.
+ * Fold it into the next digit position. We don't need to worry about
+ * overflow here since this should nearly cancel with the subtraction
+ * of the divisor.
*/
div[qi + 1] += div[qi] * NBASE;
div[qi] = qdigit;
/*
- * Now we do a final carry propagation pass to normalize the result,
- * which we combine with storing the result digits into the output.
- * Note that this is still done at full precision w/guard digits.
+ * Now we do a final carry propagation pass to normalize the result, which
+ * we combine with storing the result digits into the output. Note that
+ * this is still done at full precision w/guard digits.
*/
alloc_var(result, div_ndigits + 1);
res_digits = result->digits;
result->sign = res_sign;
/* Round to target rscale (and set result->dscale) */
- round_var(result, rscale);
+ if (round)
+ round_var(result, rscale);
+ else
+ trunc_var(result, rscale);
/* Strip leading and trailing zeroes */
strip_var(result);
int rscale;
/*
- * The result scale of a division isn't specified in any SQL standard.
- * For PostgreSQL we select a result scale that will give at least
+ * The result scale of a division isn't specified in any SQL standard. For
+ * PostgreSQL we select a result scale that will give at least
* NUMERIC_MIN_SIG_DIGITS significant digits, so that numeric gives a
* result no less accurate than float8; but use a scale not less than
* either input's display scale.
mod_var(NumericVar *var1, NumericVar *var2, NumericVar *result)
{
NumericVar tmp;
- int rscale;
init_var(&tmp);
/* ---------
* We do this using the equation
* mod(x,y) = x - trunc(x/y)*y
- * We set rscale the same way numeric_div and numeric_mul do
- * to get the right answer from the equation. The final result,
- * however, need not be displayed to more precision than the inputs.
+ * div_var can be persuaded to give us trunc(x/y) directly.
* ----------
*/
- rscale = select_div_scale(var1, var2);
-
- div_var(var1, var2, &tmp, rscale);
-
- trunc_var(&tmp, 0);
+ div_var(var1, var2, &tmp, 0, false);
- mul_var(var2, &tmp, &tmp, var2->dscale + tmp.dscale);
+ mul_var(var2, &tmp, &tmp, var2->dscale);
sub_var(var1, &tmp, result);
- round_var(result, Max(var1->dscale, var2->dscale));
-
free_var(&tmp);
}
}
/*
- * SQL2003 defines sqrt() in terms of power, so we need to emit the
- * right SQLSTATE error code if the operand is negative.
+ * SQL2003 defines sqrt() in terms of power, so we need to emit the right
+ * SQLSTATE error code if the operand is negative.
*/
if (stat < 0)
ereport(ERROR,
for (;;)
{
- div_var(&tmp_arg, result, &tmp_val, local_rscale);
+ div_var_fast(&tmp_arg, result, &tmp_val, local_rscale, true);
add_var(result, &tmp_val, result);
mul_var(result, &const_zero_point_five, result, local_rscale);
/* Compensate for input sign, and round to requested rscale */
if (xneg)
- div_var(&const_one, result, result, rscale);
+ div_var_fast(&const_one, result, result, rscale, true);
else
round_var(result, rscale);
* exp(x) = 1 + x + x^2/2! + x^3/3! + ...
*
* Given the limited range of x, this should converge reasonably quickly.
- * We run the series until the terms fall below the local_rscale
- * limit.
+ * We run the series until the terms fall below the local_rscale limit.
*/
add_var(&const_one, &x, result);
set_var_from_var(&x, &xpow);
add_var(&ni, &const_one, &ni);
mul_var(&xpow, &x, &xpow, local_rscale);
mul_var(&ifac, &ni, &ifac, 0);
- div_var(&xpow, &ifac, &elem, local_rscale);
+ div_var_fast(&xpow, &ifac, &elem, local_rscale, true);
if (elem.ndigits == 0)
break;
*/
sub_var(&x, &const_one, result);
add_var(&x, &const_one, &elem);
- div_var(result, &elem, result, local_rscale);
+ div_var_fast(result, &elem, result, local_rscale, true);
set_var_from_var(result, &xx);
mul_var(result, result, &x, local_rscale);
{
add_var(&ni, &const_two, &ni);
mul_var(&xx, &x, &xx, local_rscale);
- div_var(&xx, &ni, &elem, local_rscale);
+ div_var_fast(&xx, &ni, &elem, local_rscale, true);
if (elem.ndigits == 0)
break;
/* Select scale for division result */
rscale = select_div_scale(&ln_num, &ln_base);
- div_var(&ln_num, &ln_base, result, rscale);
+ div_var_fast(&ln_num, &ln_base, result, rscale, true);
free_var(&ln_num);
free_var(&ln_base);
free_var(&x);
}
+ /*
+ * This avoids log(0) for cases of 0 raised to a non-integer. 0 ^ 0
+ * handled by power_var_int().
+ */
+ if (cmp_var(base, &const_zero) == 0)
+ {
+ set_var_from_var(&const_zero, result);
+ result->dscale = NUMERIC_MIN_SIG_DIGITS; /* no need to round */
+ return;
+ }
+
init_var(&ln_base);
init_var(&ln_num);
val = numericvar_to_double_no_overflow(&ln_num);
/*
- * log10(result) = num * log10(e), so this is approximately the
- * weight:
+ * log10(result) = num * log10(e), so this is approximately the weight:
*/
val *= 0.434294481903252;
NumericVar base_prod;
int local_rscale;
- /* Detect some special cases, particularly 0^0. */
-
switch (exp)
{
case 0:
- if (base->ndigits == 0)
- ereport(ERROR,
- (errcode(ERRCODE_FLOATING_POINT_EXCEPTION),
- errmsg("zero raised to zero is undefined")));
+
+ /*
+ * While 0 ^ 0 can be either 1 or indeterminate (error), we treat
+ * it as 1 because most programming languages do this. SQL:2003
+ * also requires a return value of 1.
+ * http://en.wikipedia.org/wiki/Exponentiation#Zero_to_the_zero_pow
+ * er
+ */
set_var_from_var(&const_one, result);
result->dscale = rscale; /* no need to round */
return;
round_var(result, rscale);
return;
case -1:
- div_var(&const_one, base, result, rscale);
+ div_var(&const_one, base, result, rscale, true);
return;
case 2:
mul_var(base, base, result, rscale);
/*
* The general case repeatedly multiplies base according to the bit
- * pattern of exp. We do the multiplications with some extra
- * precision.
+ * pattern of exp. We do the multiplications with some extra precision.
*/
neg = (exp < 0);
exp = Abs(exp);
/* Compensate for input sign, and round to requested rscale */
if (neg)
- div_var(&const_one, result, result, rscale);
+ div_var_fast(&const_one, result, result, rscale, true);
else
round_var(result, rscale);
}
static int
cmp_abs(NumericVar *var1, NumericVar *var2)
{
- NumericDigit *var1digits = var1->digits;
- NumericDigit *var2digits = var2->digits;
+ return cmp_abs_common(var1->digits, var1->ndigits, var1->weight,
+ var2->digits, var2->ndigits, var2->weight);
+}
+
+/* ----------
+ * cmp_abs_common() -
+ *
+ * Main routine of cmp_abs(). This function can be used by both
+ * NumericVar and Numeric.
+ * ----------
+ */
+static int
+cmp_abs_common(const NumericDigit *var1digits, int var1ndigits, int var1weight,
+ const NumericDigit *var2digits, int var2ndigits, int var2weight)
+{
int i1 = 0;
int i2 = 0;
- int w1 = var1->weight;
- int w2 = var2->weight;
/* Check any digits before the first common digit */
- while (w1 > w2 && i1 < var1->ndigits)
+ while (var1weight > var2weight && i1 < var1ndigits)
{
if (var1digits[i1++] != 0)
return 1;
- w1--;
+ var1weight--;
}
- while (w2 > w1 && i2 < var2->ndigits)
+ while (var2weight > var1weight && i2 < var2ndigits)
{
if (var2digits[i2++] != 0)
return -1;
- w2--;
+ var2weight--;
}
/* At this point, either w1 == w2 or we've run out of digits */
- if (w1 == w2)
+ if (var1weight == var2weight)
{
- while (i1 < var1->ndigits && i2 < var2->ndigits)
+ while (i1 < var1ndigits && i2 < var2ndigits)
{
int stat = var1digits[i1++] - var2digits[i2++];
}
/*
- * At this point, we've run out of digits on one side or the other; so
- * any remaining nonzero digits imply that side is larger
+ * At this point, we've run out of digits on one side or the other; so any
+ * remaining nonzero digits imply that side is larger
*/
- while (i1 < var1->ndigits)
+ while (i1 < var1ndigits)
{
if (var1digits[i1++] != 0)
return 1;
}
- while (i2 < var2->ndigits)
+ while (i2 < var2ndigits)
{
if (var2digits[i2++] != 0)
return -1;
di = (var->weight + 1) * DEC_DIGITS + rscale;
/*
- * If di = 0, the value loses all digits, but could round up to 1 if
- * its first extra digit is >= 5. If di < 0 the result must be 0.
+ * If di = 0, the value loses all digits, but could round up to 1 if its
+ * first extra digit is >= 5. If di < 0 the result must be 0.
*/
if (di < 0)
{
/*
* trunc_var
*
- * Truncate the value of a variable at rscale decimal digits after the
- * decimal point. NOTE: we allow rscale < 0 here, implying
+ * Truncate (towards zero) the value of a variable at rscale decimal digits
+ * after the decimal point. NOTE: we allow rscale < 0 here, implying
* truncation before the decimal point.
*/
static void
projects / postgresql / blobdiff