Fix int32 overflow in ltree_compare()

The expression (len_diff * 10 * (an + 1)) used as the return value of
ltree_compare() is computed at int32 width.  With LTREE_MAX_LEVELS =
65535, the product can exceed INT32_MAX once an ltree has more than
~14,653 levels, which causes the result to wrap and invert its sign.
That corrupts btree ordering as well as the "magnitude" consumed by
ltree_penalty() for GiST page splits.

To fix, split ltree_compare() into two functions.  The new
ltree_compare_distance() function returns a float, which won't
overflow.  It's used by the ltree_penalty() caller.  All the other
callers only care about the sign of the return value, i.e. which of
the arguments is greater, so change ltree_compare() to not multiply
the result with (10 * (an + 1)), which avoids the overflow for those
callers.

Existing btree or GiST indexes on ltree columns containing values with
more than ~14,653 levels may be corrupt and should be REINDEXed.

Add a regression test based on the reporter's PoC.

Author: Ayush Tiwari <ayushtiwari.slg01@gmail.com>
Reported-by: 王跃林 <violin0613@tju.edu.cn>
Discussion: https://www.postgresql.org/message-id/AI6AnABgKW93Qbx1jVzi84r9.8.1781322625756.Hmail.3020001251%40tju.edu.cn
Backpatch-through: 14

diff --git a/contrib/ltree/expected/ltree.out b/contrib/ltree/expected/ltree.out
index 20f6ad7d0fec475e452991a5872ac0a1dfb03734..457263826578e09966098dbc65a2ae577430f8a1 100644 (file)
--- a/contrib/ltree/expected/ltree.out
+++ b/contrib/ltree/expected/ltree.out
@@ -8113,3 +8113,13 @@ DETAIL:  Total size of level exceeds the maximum allowed (65535 bytes).
 SELECT (repeat('a|', 65535) || 'a')::lquery;
 ERROR:  lquery level has too many variants
 DETAIL:  Number of variants exceeds the maximum allowed (65535).
+-- Test that ltree_compare() does not overflow with very deep paths.
+WITH s AS (SELECT 'a'::ltree AS v),
+     l AS (SELECT (repeat('a.', 14999) || 'a')::ltree AS v)
+SELECT (l.v > s.v) AS gt_ok, (l.v < s.v) AS lt_ok, (l.v = s.v) AS eq_ok
+  FROM s, l;
+ gt_ok | lt_ok | eq_ok 
+-------+-------+-------
+ t     | f     | f
+(1 row)
+

diff --git a/contrib/ltree/ltree.h b/contrib/ltree/ltree.h
index 3673b9d74c8be509a74dbfb4f46fcae41128bf61..2d738254f07989d92baddfb4391a6a1b4cd67dde 100644 (file)
--- a/contrib/ltree/ltree.h
+++ b/contrib/ltree/ltree.h
@@ -206,6 +206,7 @@ bool                ltree_execute(ITEM *curitem, void *checkval,
                                                  bool calcnot, bool (*chkcond) (void *checkval, ITEM *val));
 
 int                    ltree_compare(const ltree *a, const ltree *b);
+float          ltree_compare_distance(const ltree *a, const ltree *b);
 bool           inner_isparent(const ltree *c, const ltree *p);
 bool           compare_subnode(ltree_level *t, char *qn, int len, bool prefix, bool ci);
 ltree     *lca_inner(ltree **a, int len);

diff --git a/contrib/ltree/ltree_gist.c b/contrib/ltree/ltree_gist.c
index 1b893e9645d23d32d827db618cffcc499f415a3f..96c9774b2d81a511fd655f282effdbfe487a6b7d 100644 (file)
--- a/contrib/ltree/ltree_gist.c
+++ b/contrib/ltree/ltree_gist.c
@@ -261,11 +261,11 @@ ltree_penalty(PG_FUNCTION_ARGS)
        ltree_gist *newval = (ltree_gist *) DatumGetPointer(((GISTENTRY *) PG_GETARG_POINTER(1))->key);
        float      *penalty = (float *) PG_GETARG_POINTER(2);
        int                     siglen = LTREE_GET_SIGLEN();
-       int32           cmpr,
+       float           cmpr,
                                cmpl;
 
-       cmpl = ltree_compare(LTG_GETLNODE(origval, siglen), LTG_GETLNODE(newval, siglen));
-       cmpr = ltree_compare(LTG_GETRNODE(newval, siglen), LTG_GETRNODE(origval, siglen));
+       cmpl = ltree_compare_distance(LTG_GETLNODE(origval, siglen), LTG_GETLNODE(newval, siglen));
+       cmpr = ltree_compare_distance(LTG_GETRNODE(newval, siglen), LTG_GETRNODE(origval, siglen));
 
        *penalty = Max(cmpl, 0) + Max(cmpr, 0);
 

diff --git a/contrib/ltree/ltree_op.c b/contrib/ltree/ltree_op.c
index da1db5fcd2229b23fd8d2c0fbc1497416586022a..a2845a928be29c24f9cd523a65da47184ee0ce9c 100644 (file)
--- a/contrib/ltree/ltree_op.c
+++ b/contrib/ltree/ltree_op.c
@@ -38,6 +38,9 @@ PG_FUNCTION_INFO_V1(ltree2text);
 PG_FUNCTION_INFO_V1(text2ltree);
 PG_FUNCTION_INFO_V1(ltreeparentsel);
 
+/*
+ * btree-comparison function.
+ */
 int
 ltree_compare(const ltree *a, const ltree *b)
 {
@@ -50,18 +53,52 @@ ltree_compare(const ltree *a, const ltree *b)
        {
                int                     res;
 
-               if ((res = memcmp(al->name, bl->name, Min(al->len, bl->len))) == 0)
+               res = memcmp(al->name, bl->name, Min(al->len, bl->len));
+               if (res == 0)
+               {
+                       if (al->len != bl->len)
+                               return (int) al->len - (int) bl->len;
+               }
+               else
+                       return res;
+
+               an--;
+               bn--;
+               al = LEVEL_NEXT(al);
+               bl = LEVEL_NEXT(bl);
+       }
+
+       return a->numlevel - b->numlevel;
+}
+
+/*
+ * Returns a "distance" between a and b.  If a < b, the distance is negative,
+ * consistent with the ltree_compare() ordering.
+ */
+float
+ltree_compare_distance(const ltree *a, const ltree *b)
+{
+       ltree_level *al = LTREE_FIRST(a);
+       ltree_level *bl = LTREE_FIRST(b);
+       int                     an = a->numlevel;
+       int                     bn = b->numlevel;
+
+       while (an > 0 && bn > 0)
+       {
+               int                     res;
+
+               res = memcmp(al->name, bl->name, Min(al->len, bl->len));
+               if (res == 0)
                {
                        if (al->len != bl->len)
-                               return (al->len - bl->len) * 10 * (an + 1);
+                               return (float) (al->len - bl->len) * 10.0 * (an + 1);
                }
                else
                {
                        if (res < 0)
-                               res = -1;
+                               return -1.0 * 10.0 * (an + 1);
                        else
-                               res = 1;
-                       return res * 10 * (an + 1);
+                               return 1.0 * 10.0 * (an + 1);
                }
 
                an--;
@@ -70,7 +107,7 @@ ltree_compare(const ltree *a, const ltree *b)
                bl = LEVEL_NEXT(bl);
        }
 
-       return (a->numlevel - b->numlevel) * 10 * (an + 1);
+       return ((float) (a->numlevel - b->numlevel)) * 10.0 * (an + 1);
 }
 
 #define RUNCMP                                         \

diff --git a/contrib/ltree/sql/ltree.sql b/contrib/ltree/sql/ltree.sql
index b187b53c52cdb41159538a02575792778be8556a..f47f81238de0e5530cfaa3930db2f96d38685e75 100644 (file)
--- a/contrib/ltree/sql/ltree.sql
+++ b/contrib/ltree/sql/ltree.sql
@@ -404,3 +404,9 @@ SELECT (repeat('x', 255) || repeat('|' || repeat('x', 255), 256))::lquery;
 --- Test for overflow of lquery_level.numvar, with a set of single-char
 --- variants in one level.
 SELECT (repeat('a|', 65535) || 'a')::lquery;
+
+-- Test that ltree_compare() does not overflow with very deep paths.
+WITH s AS (SELECT 'a'::ltree AS v),
+     l AS (SELECT (repeat('a.', 14999) || 'a')::ltree AS v)
+SELECT (l.v > s.v) AS gt_ok, (l.v < s.v) AS lt_ok, (l.v = s.v) AS eq_ok
+  FROM s, l;