recursive query to update comments parents

hive/db/db_state.py

@@ -13,6 +13,7 @@ from hive.db.schema import (setup, reset_autovac, build_metadata,
 from hive.db.adapter import Db
 
 from hive.utils.trends import update_all_hot_and_tranding
+from hive.utils.post_active import update_all_posts_active
 
 log = logging.getLogger(__name__)
 
@@ -197,6 +198,12 @@ class DbState:
         time_end = perf_counter()
         log.info("[INIT] update_all_hot_and_tranding executed in %fs", time_end - time_start)
 
+        time_start = perf_counter()
+
+        update_all_posts_active()
+
+        time_end = perf_counter()
+        log.info("[INIT] update_all_posts_active executed in %fs", time_end - time_start)
         # TODO: #111
         #for key in cls._all_foreign_keys():
         #    log.info("Create fk %s", key.name)

hive/db/schema.py

@@ -1228,6 +1228,24 @@ def setup(db):
     """
     db.query_no_return(sql)
 
+    sql = """
+        DROP FUNCTION IF EXISTS public.max_time_stamp() CASCADE;
+        CREATE OR REPLACE FUNCTION public.max_time_stamp( _first TIMESTAMP, _second TIMESTAMP )
+        RETURNS TIMESTAMP
+        LANGUAGE 'plpgsql'
+        IMMUTABLE
+        AS $BODY$
+        BEGIN
+	        IF _first > _second THEN
+	             RETURN _first;
+            ELSE
+	             RETURN _second;
+            END IF;
+        END
+        $BODY$;
+        """
+    db.query_no_return(sql)
+
 def reset_autovac(db):
     """Initializes/resets per-table autovacuum/autoanalyze params.
 

hive/indexer/blocks.py

@@ -17,6 +17,7 @@ from time import perf_counter
 
 from hive.utils.stats import OPStatusManager as OPSM
 from hive.utils.stats import FlushStatusManager as FSM
+from hive.utils.post_active import update_active_starting_from_posts_on_block
 
 log = logging.getLogger(__name__)
 
@@ -60,6 +61,8 @@ class Blocks:
         Tags.flush()
         Votes.flush()
         Posts.flush()
+        block_num = int(block['block_id'][:8], base=16)
+        cls.on_live_blocks_processed( block_num, block_num, is_initial_sync )
         time_end = perf_counter()
         log.info("[PROCESS BLOCK] %fs", time_end - time_start)
         return ret
@@ -71,8 +74,11 @@ class Blocks:
         DB.query("START TRANSACTION")
 
         last_num = 0
+        first_block = -1
         try:
             for block in blocks:
+                if first_block == -1:
+                    first_block = int(block['block_id'][:8], base=16)
                 last_num = cls._process(block, vops, hived, is_initial_sync)
         except Exception as e:
             log.error("exception encountered block %d", last_num + 1)
@@ -97,6 +103,9 @@ class Blocks:
         flush_time = register_time(flush_time, "Follow", folllow_items)
         flush_time = register_time(flush_time, "Posts", Posts.flush())
 
+        if first_block > -1:
+            cls.on_live_blocks_processed( first_block, last_num, is_initial_sync )
+
         DB.query("COMMIT")
 
         log.info(f"[PROCESS MULTI] {len(blocks)} blocks in {OPSM.stop(time_start) :.4f}s")
@@ -382,3 +391,13 @@ class Blocks:
         DB.query("COMMIT")
         log.warning("[FORK] recovery complete")
         # TODO: manually re-process here the blocks which were just popped.
+
+    @classmethod
+    def on_live_blocks_processed( cls, first_block, last_block, is_initial_sync ):
+        """Is invoked when processing of block range is done and received
+           informations from hived are already stored in db
+        """
+        if is_initial_sync:
+            return;
+
+        update_active_starting_from_posts_on_block( first_block, last_block )

hive/utils/post_active.py

+from hive.db.adapter import Db
+
+DB = Db.instance()
+"""
+There are three cases when 'active' field in post is updated:
+1) when a descendant post comment was added (recursivly on any depth)
+2) when a descendant post comment was deleted (recursivly on any depth)
+3) when the post is updated
+
+It means that, when the comment for posts is updated then its 'active' field
+does not propagate for its ancestors.
+"""
+
+update_active_sql = """
+    WITH RECURSIVE parent_posts ( parent_id, post_id, intrusive_active) AS (
+    	SELECT
+    		parent_id as parent_id,
+    		id as post_id,
+    		CASE WHEN hp1.active = hp1.created_at OR hp1.counter_deleted > 0 THEN hp1.active
+    		ELSE hp1.created_at
+    		END as intrusive_active
+    	FROM hive_posts hp1 {}
+    	UNION
+    	SELECT
+    		hp2.parent_id as parent_id,
+    		id as post_id,
+    		max_time_stamp(
+    			CASE WHEN hp2.active = hp2.created_at OR hp2.counter_deleted > 0 THEN hp2.active
+    			ELSE hp2.created_at
+    			END
+    			, pp.intrusive_active
+    		) as intrusive_active
+    	FROM parent_posts pp
+    	JOIN hive_posts hp2 ON pp.parent_id = hp2.id
+    	WHERE hp2.depth > 0 AND pp.intrusive_active > hp2.active
+    )
+   UPDATE
+	   hive_posts
+   SET
+	   active = new_active
+   FROM
+   (
+        SELECT hp.id as post_id, MAX(pp.intrusive_active) as new_active
+        FROM parent_posts pp
+        JOIN hive_posts hp ON pp.parent_id = hp.id GROUP BY hp.id
+    ) as dataset
+    WHERE dataset.post_id = hive_posts.id;
+    """
+
+def update_all_posts_active():
+    DB.query_no_return(update_active_sql.format( "WHERE ( children = 0 OR hp1.counter_deleted > 0 ) AND depth > 0" ))
+
+def update_active_starting_from_posts_on_block( first_block_num, last_block_num ):
+    if first_block_num == last_block_num:
+            DB.query_no_return(update_active_sql.format( "WHERE block_num={} AND depth > 0" ).format(first_block_num) )
+            return
+    DB.query_no_return(update_active_sql.format( "WHERE block_num>={} AND block_num <={} AND depth > 0" ).format(first_block_num, last_block_num) )

tests_api @ 385a6826

-Subproject commit bae23397a6538852645dc8f3c4f08eca06f18144
+Subproject commit 385a6826f33d7818ba0e04bb2144bf5f275d1acf