Re: fix for BUG #3720: wrong results at using ltree - Mailing list pgsql-hackers

Nikita Glukhov <n.gluhov@postgrespro.ru> writes:
> On 24.01.2020 21:29, Tomas Vondra wrote:
>> Unfortunately, the current code is somewhat undercommented :-(

> The main problem is that no one really understands how it works now.

Indeed.  I was disturbed to realize that lquery_op.c, despite being
far from trivial code, contained NOT ONE SINGLE COMMENT before today,
other than the content-free file header and a commented-out (visibly
unsafe, too) debugging printing function.  This is a long way south
of minimally acceptable, in my book.

Anyway, I concur that Nikita's two patches are bug fixes, so I pushed
them.  Nonetheless, he *did* hijack this thread, so in hopes of restoring
attention to the original topic, here's a rebased version of the original
patch.

My main complaint about it remains the same, that it changes a
disturbingly large number of existing regression-test results,
suggesting that there's not a meeting of the minds about what
this logic is supposed to do.  Maybe it's okay or maybe it's
not, but who's going to decide?

Also, now that I've looked at it a bit more, I'd be inclined to
strip out the parts of the patch that remove setting up the
LQUERY_HASNOT flag.  Even if we're not using that right now,
we might want it again someday, and we're not saving much of
anything by introducing a minor on-disk incompatibility.

            regards, tom lane

diff --git a/contrib/ltree/expected/ltree.out b/contrib/ltree/expected/ltree.out
index 1b31dbc..1eef086 100644
--- a/contrib/ltree/expected/ltree.out
+++ b/contrib/ltree/expected/ltree.out
@@ -722,7 +722,7 @@ SELECT 'a.b.c.d.e'::ltree ~ '*.a.*.d.*';
 SELECT 'a.b.c.d.e'::ltree ~ '*.!d.*';
  ?column?
 ----------
- f
+ t
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ '*.!d';
@@ -752,7 +752,7 @@ SELECT 'a.b.c.d.e'::ltree ~ '*.!e';
 SELECT 'a.b.c.d.e'::ltree ~ '*.!e.*';
  ?column?
 ----------
- f
+ t
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ 'a.*.!e';
@@ -770,7 +770,7 @@ SELECT 'a.b.c.d.e'::ltree ~ 'a.*.!d';
 SELECT 'a.b.c.d.e'::ltree ~ 'a.*.!d.*';
  ?column?
 ----------
- f
+ t
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ 'a.*.!f.*';
@@ -788,7 +788,7 @@ SELECT 'a.b.c.d.e'::ltree ~ '*.a.*.!f.*';
 SELECT 'a.b.c.d.e'::ltree ~ '*.a.*.!d.*';
  ?column?
 ----------
- f
+ t
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ '*.a.!d.*';
@@ -812,13 +812,13 @@ SELECT 'a.b.c.d.e'::ltree ~ 'a.!d.*';
 SELECT 'a.b.c.d.e'::ltree ~ '*.a.*.!d.*';
  ?column?
 ----------
- f
+ t
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ '*.!b.*';
  ?column?
 ----------
- f
+ t
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ '*.!b.c.*';
@@ -830,7 +830,7 @@ SELECT 'a.b.c.d.e'::ltree ~ '*.!b.c.*';
 SELECT 'a.b.c.d.e'::ltree ~ '*.!b.*.c.*';
  ?column?
 ----------
- f
+ t
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ '!b.*.c.*';
@@ -878,31 +878,31 @@ SELECT 'a.b.c.d.e'::ltree ~ '*{1}.!b.*.!c.*.e';
 SELECT 'a.b.c.d.e'::ltree ~ '*{1}.!b.*{1}.!c.*.e';
  ?column?
 ----------
- t
+ f
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ 'a.!b.*{1}.!c.*.e';
  ?column?
 ----------
- t
+ f
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ '!b.*{1}.!c.*.e';
  ?column?
 ----------
- t
+ f
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ '*.!b.*{1}.!c.*.e';
  ?column?
 ----------
- t
+ f
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ '*.!b.*.!c.*.e';
  ?column?
 ----------
- f
+ t
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ '!b.!c.*';
@@ -932,19 +932,19 @@ SELECT 'a.b.c.d.e'::ltree ~ '*{1}.!b.*.!c.*';
 SELECT 'a.b.c.d.e'::ltree ~ '*{1}.!b.*{1}.!c.*';
  ?column?
 ----------
- t
+ f
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ 'a.!b.*{1}.!c.*';
  ?column?
 ----------
- t
+ f
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ '!b.*{1}.!c.*';
  ?column?
 ----------
- t
+ f
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ '*.!b.*{1}.!c.*';
@@ -956,7 +956,7 @@ SELECT 'a.b.c.d.e'::ltree ~ '*.!b.*{1}.!c.*';
 SELECT 'a.b.c.d.e'::ltree ~ '*.!b.*.!c.*';
  ?column?
 ----------
- f
+ t
 (1 row)

 SELECT 'a.b.c.d.e'::ltree ~ 'a.*{2}.*{2}';
@@ -983,6 +983,18 @@ SELECT 'a.b.c.d.e'::ltree ~ 'a.*{5}.*';
  f
 (1 row)

+SELECT '5.0.1.0'::ltree ~ '5.!0.!0.0';
+ ?column?
+----------
+ f
+(1 row)
+
+SELECT 'a.b'::ltree ~ '!a.!a';
+ ?column?
+----------
+ f
+(1 row)
+
 SELECT 'QWER_TY'::ltree ~ 'q%@*';
  ?column?
 ----------
diff --git a/contrib/ltree/lquery_op.c b/contrib/ltree/lquery_op.c
index 5c7afe5..81fcb0e 100644
--- a/contrib/ltree/lquery_op.c
+++ b/contrib/ltree/lquery_op.c
@@ -19,16 +19,6 @@ PG_FUNCTION_INFO_V1(lt_q_rregex);

 #define NEXTVAL(x) ( (lquery*)( (char*)(x) + INTALIGN( VARSIZE(x) ) ) )

-typedef struct
-{
-    lquery_level *q;
-    int            nq;
-    ltree_level *t;
-    int            nt;
-    int            posq;
-    int            post;
-} FieldNot;
-
 static char *
 getlexeme(char *start, char *end, int *len)
 {
@@ -109,6 +99,9 @@ checkLevel(lquery_level *curq, ltree_level *curt)
     int            (*cmpptr) (const char *, const char *, size_t);
     lquery_variant *curvar = LQL_FIRST(curq);
     int            i;
+    bool        success;
+
+    success = (curq->flag & LQL_NOT) ? false : true;

     for (i = 0; i < curq->numvar; i++)
     {
@@ -116,39 +109,26 @@ checkLevel(lquery_level *curq, ltree_level *curt)

         if (curvar->flag & LVAR_SUBLEXEME)
         {
-            if (compare_subnode(curt, curvar->name, curvar->len, cmpptr, (curvar->flag & LVAR_ANYEND)))
-                return true;
+            if (compare_subnode(curt, curvar->name, curvar->len, cmpptr,
+                                (curvar->flag & LVAR_ANYEND)))
+                return success;
         }
         else if ((curvar->len == curt->len ||
                   (curt->len > curvar->len && (curvar->flag & LVAR_ANYEND))) &&
                  (*cmpptr) (curvar->name, curt->name, curvar->len) == 0)
         {

-            return true;
+            return success;
         }
         curvar = LVAR_NEXT(curvar);
     }
-    return false;
+    return !success;
 }

 /*
-void
-printFieldNot(FieldNot *fn ) {
-    while(fn->q) {
-        elog(NOTICE,"posQ:%d lenQ:%d posT:%d lenT:%d", fn->posq,fn->nq,fn->post,fn->nt);
-        fn++;
-    }
-}
-*/
-
-/*
  * Try to match an lquery (of query_numlevel items) to an ltree (of
  * tree_numlevel items)
  *
- * If the query contains any NOT flags, "ptr" must point to a FieldNot
- * workspace initialized with ptr->q == NULL.  Otherwise it can be NULL.
- * (LQL_NOT flags will be ignored if ptr == NULL.)
- *
  * high_pos is the last ltree position the first lquery item is allowed
  * to match at; it should be zero for external calls.
  *
@@ -157,7 +137,6 @@ printFieldNot(FieldNot *fn ) {
 static bool
 checkCond(lquery_level *curq, int query_numlevel,
           ltree_level *curt, int tree_numlevel,
-          FieldNot *ptr,
           uint32 high_pos,
           bool force_advance)
 {
@@ -181,7 +160,7 @@ checkCond(lquery_level *curq, int query_numlevel,
         if (curq->numvar)
         {
             /* Current query item is not '*' */
-            ltree_level *prevt = curt;
+            bool        isok = false;

             /* skip tree items that must be ignored due to prior * items */
             while (cur_tpos < low_pos)
@@ -191,82 +170,28 @@ checkCond(lquery_level *curq, int query_numlevel,
                 cur_tpos++;
                 if (tlen == 0)
                     return false;
-                if (ptr && ptr->q)
-                    ptr->nt++;
             }

-            if (ptr && (curq->flag & LQL_NOT))
+            while (cur_tpos <= high_pos && tlen > 0 && !isok)
             {
-                /* Deal with a NOT item */
-                if (!(prevq && prevq->numvar == 0))
-                    prevq = curq;
-                if (ptr->q == NULL)
-                {
-                    ptr->t = prevt;
-                    ptr->q = prevq;
-                    ptr->nt = 1;
-                    ptr->nq = 1 + ((prevq == curq) ? 0 : 1);
-                    ptr->posq = query_numlevel - qlen - ((prevq == curq) ? 0 : 1);
-                    ptr->post = cur_tpos;
-                }
-                else
-                {
-                    ptr->nt++;
-                    ptr->nq++;
-                }
-
-                if (qlen == 1 && ptr->q->numvar == 0)
-                    ptr->nt = tree_numlevel - ptr->post;
+                isok = checkLevel(curq, curt);
                 curt = LEVEL_NEXT(curt);
                 tlen--;
                 cur_tpos++;
-                if (high_pos < cur_tpos)
-                    high_pos++;
-            }
-            else
-            {
-                /* Not a NOT item, check for normal match */
-                bool        isok = false;
-
-                while (cur_tpos <= high_pos && tlen > 0 && !isok)
+                if (isok && prevq && prevq->numvar == 0 && tlen > 0 && cur_tpos <= high_pos)
                 {
-                    isok = checkLevel(curq, curt);
-                    curt = LEVEL_NEXT(curt);
-                    tlen--;
-                    cur_tpos++;
-                    if (isok && prevq && prevq->numvar == 0 &&
-                        tlen > 0 && cur_tpos <= high_pos)
-                    {
-                        FieldNot    tmpptr;
-
-                        if (ptr)
-                            memcpy(&tmpptr, ptr, sizeof(FieldNot));
-                        if (checkCond(prevq, qlen + 1,
-                                      curt, tlen,
-                                      (ptr) ? &tmpptr : NULL,
-                                      high_pos - cur_tpos,
-                                      true))
-                            return true;
-                    }
-                    if (!isok && ptr && ptr->q)
-                        ptr->nt++;
+                    if (checkCond(prevq, qlen + 1,
+                                  curt, tlen,
+                                  high_pos - cur_tpos,
+                                  true))
+                        return true;
                 }
-                if (!isok)
-                    return false;
-
-                if (ptr && ptr->q)
-                {
-                    if (checkCond(ptr->q, ptr->nq,
-                                  ptr->t, ptr->nt,
-                                  NULL,
-                                  0,
-                                  false))
-                        return false;
-                    ptr->q = NULL;
-                }
-                low_pos = cur_tpos;
-                high_pos = cur_tpos;
             }
+            if (!isok)
+                return false;
+
+            low_pos = cur_tpos;
+            high_pos = cur_tpos;
         }
         else
         {
@@ -277,13 +202,6 @@ checkCond(lquery_level *curq, int query_numlevel,
                 return false;

             high_pos = Min(high_pos + curq->high, tree_numlevel);
-
-            if (ptr && ptr->q)
-            {
-                ptr->nq++;
-                if (qlen == 1)
-                    ptr->nt = tree_numlevel - ptr->post;
-            }
         }

         prevq = curq;
@@ -321,15 +239,6 @@ checkCond(lquery_level *curq, int query_numlevel,
     if (low_pos > tree_numlevel || tree_numlevel > high_pos)
         return false;

-    /* Finish pending NOT check, if any */
-    if (ptr && ptr->q &&
-        checkCond(ptr->q, ptr->nq,
-                  ptr->t, ptr->nt,
-                  NULL,
-                  0,
-                  false))
-        return false;
-
     return true;
 }

@@ -340,26 +249,10 @@ ltq_regex(PG_FUNCTION_ARGS)
     lquery       *query = PG_GETARG_LQUERY_P(1);
     bool        res = false;

-    if (query->flag & LQUERY_HASNOT)
-    {
-        FieldNot    fn;
-
-        fn.q = NULL;
-
-        res = checkCond(LQUERY_FIRST(query), query->numlevel,
-                        LTREE_FIRST(tree), tree->numlevel,
-                        &fn,
-                        0,
-                        false);
-    }
-    else
-    {
-        res = checkCond(LQUERY_FIRST(query), query->numlevel,
-                        LTREE_FIRST(tree), tree->numlevel,
-                        NULL,
-                        0,
-                        false);
-    }
+    res = checkCond(LQUERY_FIRST(query), query->numlevel,
+                    LTREE_FIRST(tree), tree->numlevel,
+                    0,
+                    false);

     PG_FREE_IF_COPY(tree, 0);
     PG_FREE_IF_COPY(query, 1);
diff --git a/contrib/ltree/ltree.h b/contrib/ltree/ltree.h
index 2c48304..02f05fc 100644
--- a/contrib/ltree/ltree.h
+++ b/contrib/ltree/ltree.h
@@ -83,8 +83,6 @@ typedef struct
 #define LQUERY_FIRST(x)   ( (lquery_level*)( ((char*)(x))+LQUERY_HDRSIZE ) )
 #define LQUERY_MAX_LEVELS    PG_UINT16_MAX    /* lquery.numlevel is uint16 */

-#define LQUERY_HASNOT        0x01
-
 #define ISALNUM(x)    ( t_isalpha(x) || t_isdigit(x)    || ( pg_mblen(x) == 1 && t_iseq((x), '_') ) )

 /* full text query */
diff --git a/contrib/ltree/ltree_io.c b/contrib/ltree/ltree_io.c
index 3ae7dc0..43cbb66 100644
--- a/contrib/ltree/ltree_io.c
+++ b/contrib/ltree/ltree_io.c
@@ -205,7 +205,6 @@ lquery_in(PG_FUNCTION_ARGS)
                *curqlevel,
                *tmpql;
     lquery_variant *lrptr = NULL;
-    bool        hasnot = false;
     bool        wasbad = false;
     int            charlen;
     int            pos = 0;
@@ -254,7 +253,6 @@ lquery_in(PG_FUNCTION_ARGS)
                 state = LQPRS_WAITDELIM;
                 curqlevel->numvar = 1;
                 curqlevel->flag |= LQL_NOT;
-                hasnot = true;
             }
             else if (charlen == 1 && t_iseq(ptr, '*'))
                 state = LQPRS_WAITOPEN;
@@ -498,8 +496,6 @@ lquery_in(PG_FUNCTION_ARGS)
     result->numlevel = num;
     result->firstgood = 0;
     result->flag = 0;
-    if (hasnot)
-        result->flag |= LQUERY_HASNOT;
     cur = LQUERY_FIRST(result);
     curqlevel = tmpql;
     while ((char *) curqlevel - (char *) tmpql < num * ITEMSIZE)
diff --git a/contrib/ltree/sql/ltree.sql b/contrib/ltree/sql/ltree.sql
index 1de0b2a..d1a4d6f 100644
--- a/contrib/ltree/sql/ltree.sql
+++ b/contrib/ltree/sql/ltree.sql
@@ -183,6 +183,8 @@ SELECT 'a.b.c.d.e'::ltree ~ 'a.*{2}.*{2}';
 SELECT 'a.b.c.d.e'::ltree ~ 'a.*{1}.*{2}.e';
 SELECT 'a.b.c.d.e'::ltree ~ 'a.*{1}.*{4}';
 SELECT 'a.b.c.d.e'::ltree ~ 'a.*{5}.*';
+SELECT '5.0.1.0'::ltree ~ '5.!0.!0.0';
+SELECT 'a.b'::ltree ~ '!a.!a';

 SELECT 'QWER_TY'::ltree ~ 'q%@*';
 SELECT 'QWER_TY'::ltree ~ 'Q_t%@*';
diff --git a/doc/src/sgml/ltree.sgml b/doc/src/sgml/ltree.sgml
index 0c74aa7..7794068 100644
--- a/doc/src/sgml/ltree.sgml
+++ b/doc/src/sgml/ltree.sgml
@@ -603,7 +603,7 @@ ltreetest=> SELECT path FROM test WHERE path ~ '*.Astronomy.*';
  Top.Collections.Pictures.Astronomy.Astronauts
 (7 rows)

-ltreetest=> SELECT path FROM test WHERE path ~ '*.!pictures@.*.Astronomy.*';
+ltreetest=> SELECT path FROM test WHERE path ~ '*.!pictures@.Astronomy.*';
                 path
 ------------------------------------
  Top.Science.Astronomy