From 60fb34cda558a6977b29415bb90717329f013427 Mon Sep 17 00:00:00 2001
From: hyc520 <huyicheng1@xfusion.com>
Date: Thu, 4 Aug 2022 19:17:08 +0800
Subject: [PATCH] improve the performance of numerics in libpq

---
 .../libpq/client_logic_fmt/numeric.cpp        | 75 +++++++++++--------
 .../libpq/client_logic_fmt/numeric.h          |  3 +
 2 files changed, 47 insertions(+), 31 deletions(-)

diff --git a/src/common/interfaces/libpq/client_logic_fmt/numeric.cpp b/src/common/interfaces/libpq/client_logic_fmt/numeric.cpp
index 36967763f..657e8e5f2 100644
--- a/src/common/interfaces/libpq/client_logic_fmt/numeric.cpp
+++ b/src/common/interfaces/libpq/client_logic_fmt/numeric.cpp
@@ -113,13 +113,47 @@ typedef union {
 } varattrib_4b_fe;
 
 
-#define init_var(v) MemSetAligned(v, 0, sizeof(NumericVar))
-#define digitbuf_alloc(ndigits) ((NumericDigit *)palloc((ndigits) * sizeof(NumericDigit)))
+#define quick_init_var(v)    \
+    do {                     \
+        (v)->buf = (v)->ndb; \
+        (v)->digits = NULL;  \
+    } while (0)
 
+#define init_var(v)          \
+    do {                     \
+        quick_init_var((v)); \
+        (v)->ndigits = 0;    \
+        (v)->weight = 0;     \
+        (v)->sign = 0;       \
+        (v)->dscale = 0;     \
+    } while (0)
 
-#define digitbuf_free(buf) do { \
-        if ((buf) != NULL)      \
-            FREE_POINTER(buf);  \
+#define digitbuf_alloc(ndigits) \
+    ((NumericDigit*) palloc((ndigits) * sizeof(NumericDigit)))
+
+#define digitbuf_free(v)            \
+    do {                            \
+        if ((v)->buf != (v)->ndb) { \
+            pfree((v)->buf);        \
+            (v)->buf = (v)->ndb;    \
+        }                           \
+    } while (0)
+
+#define free_var(v) digitbuf_free((v));
+
+/*
+ * Init a var and allocate digit buffer of ndigits digits (plus a spare digit for rounding).
+ * Called when first using a var.
+ */
+#define init_alloc_var(v, n)                    \
+    do  {                                       \
+        (v)->buf = (v)->ndb;                    \
+        (v)->ndigits = (n);                     \
+        if ((n) > NUMERIC_LOCAL_NMAX) {         \
+            (v)->buf = digitbuf_alloc((n) + 1); \
+        }                                       \
+        (v)->buf[0] = 0;                        \
+        (v)->digits = (v)->buf + 1;             \
     } while (0)
 
 
@@ -128,7 +162,6 @@ static void alloc_var(NumericVar *var, int ndigits);
 static void strip_var(NumericVar *var);
 
 static const char *set_var_from_str(const char *str, const char *cp, NumericVar *dest, char *err_msg);
-static void free_var(NumericVar *var);
 static void init_var_from_num(NumericData* num, NumericVar *dest);
 static bool get_str_from_var(const NumericVar *var, char *str, size_t max_size);
 static NumericVar const_nan = { 0, 0, NUMERIC_NAN, 0, NULL, NULL };
@@ -155,27 +188,8 @@ static void dump_numeric(const char *str, NumericData* num);
  */
 static void alloc_var(NumericVar *var, int ndigits)
 {
-    digitbuf_free(var->buf);
-    var->buf = digitbuf_alloc(ndigits + 1);
-    if (var->buf == NULL) {
-        return;
-    }
-    var->buf[0] = 0; /* spare digit for rounding */
-    var->digits = var->buf + 1;
-    var->ndigits = ndigits;
-}
-
-
-/*
- * free_var() -
- * Return the digit buffer of a variable to the free pool
- */
-static void free_var(NumericVar *var)
-{
-    digitbuf_free(var->buf);
-    var->buf = NULL;
-    var->digits = NULL;
-    var->sign = NUMERIC_NAN;
+    digitbuf_free(var);
+    init_alloc_var(var, ndigits);
 }
 
 
@@ -187,9 +201,8 @@ static void free_var(NumericVar *var)
  */
 static void zero_var(NumericVar *var)
 {
-    digitbuf_free(var->buf);
-    var->buf = NULL;
-    var->digits = NULL;
+    digitbuf_free(var);
+    quick_init_var(var);
     var->ndigits = 0;
     var->weight = 0;         /* by convention; doesn't really matter */
     var->sign = NUMERIC_POS; /* anything but NAN... */
@@ -758,7 +771,7 @@ static inline void init_var_from_num(NumericData* num, NumericVar* dest)
     dest->sign = NUMERIC_SIGN(num);
     dest->dscale = NUMERIC_DSCALE(num);
     dest->digits = NUMERIC_DIGITS(num);
-    dest->buf = NULL; /* digits array is not palloc'd */
+    dest->buf = dest->ndb;
 }
 
 /*
diff --git a/src/common/interfaces/libpq/client_logic_fmt/numeric.h b/src/common/interfaces/libpq/client_logic_fmt/numeric.h
index c7f327c7f..5cf0f8ca2 100644
--- a/src/common/interfaces/libpq/client_logic_fmt/numeric.h
+++ b/src/common/interfaces/libpq/client_logic_fmt/numeric.h
@@ -39,6 +39,8 @@ typedef int16 NumericDigit;
 #define DEC_DIGITS 4 /* decimal digits per NBASE digit */
 #define NUMERIC_MAX_PRECISION 1000
 #define NUMERIC_MAX_RESULT_SCALE (NUMERIC_MAX_PRECISION * 2)
+#define NUMERIC_LOCAL_NDIG  36  /* number of 'digits' in local digits[] */
+#define NUMERIC_LOCAL_NMAX  (NUMERIC_LOCAL_NDIG - 2)
 
 typedef struct NumericVar {
     int ndigits;          /* # of digits in digits[] - can be 0! */
@@ -47,6 +49,7 @@ typedef struct NumericVar {
     int dscale;           /* display scale */
     NumericDigit *buf;    /* start of palloc'd space for digits[] */
     NumericDigit *digits; /* base-NBASE digits */
+    NumericDigit ndb[NUMERIC_LOCAL_NDIG]; /* local space for digits[] */
 } NumericVar;
 struct NumericShort {
     uint16 n_header;        /* Sign + display scale + weight */