supoport fetch pg_class and pg_statistic from videx-statistic-server

Author: TiNnNnnn

src/pg/videx/Makefile

@@ -5,6 +5,7 @@ OBJS = \
 	$(WIN32RES) \
 	stats.o \
 	videxam.o \
+	videx_json_item.o \
 
 CURRENT_DIR := $(shell pwd)
 $(info CURRENT_DIR = $(CURRENT_DIR))
@@ -23,6 +24,12 @@ REGRESS = videx
 CXX = g++
 CC = gcc
 
+ifdef DEBUG
+CFLAGS += -O0 -g3
+CXXFLAGS += -O0 -g3
+PG_CPPFLAGS += -O0 -g3
+endif
+
 TAP_TESTS = 1
 
 ifdef USE_PGXS

src/pg/videx/stats.cc

@@ -1,5 +1,5 @@
 #include "videx_json_item.h"
-
+#include <nlohmann/json.hpp>
 /**
  * A simple parsing function is written here instead,
  * since rapid_json always encounters strange segmentation faults across platforms,
@@ -11,70 +11,39 @@
  * @return
  */
 int videx_parse_simple_json(const std::string &json, int &code, std::string &message,
-                      std::map<std::string, std::string> &data_dict) {
-    try {
-        // find code and message
-        std::size_t pos_code = json.find("\"code\":");
-        std::size_t pos_message = json.find("\"message\":");
-        std::size_t pos_data = json.find("\"data\":");
-
-        if (pos_code == std::string::npos || pos_message == std::string::npos || pos_data == std::string::npos) {
-            throw std::invalid_argument("Missing essential components in JSON.");
-        }
-
-        // parse code
-        std::size_t start = json.find_first_of("0123456789", pos_code);
-        std::size_t end = json.find(',', start);
-        code = std::stoi(json.substr(start, end - start));
-
-        // parse message
-        start = json.find('\"', pos_message + 10) + 1;
-        end = json.find('\"', start);
-        message = json.substr(start, end - start);
-
-        // parse data
-        start = json.find('{', pos_data) + 1;
-        end = json.find('}', start);
-        std::string data_content = json.substr(start, end - start);
-        std::istringstream data_stream(data_content);
-        std::string line;
-
-        while (std::getline(data_stream, line, ',')) {
-            std::size_t colon_pos = line.find(':');
-            if (colon_pos == std::string::npos) {
-                continue; // Skip malformed line
-            }
-            std::string key = line.substr(0, colon_pos);
-            std::string value = line.substr(colon_pos + 1);
-
-            // clean key 和 value
-            auto trim_quotes_and_space = [](std::string &str) {
-                // Trim whitespace and surrounding quotes
-                size_t first = str.find_first_not_of(" \t\n\"");
-                size_t last = str.find_last_not_of(" \t\n\"");
-                if (first == std::string::npos || last == std::string::npos) {
-                    str.clear(); // All whitespace or empty
-                } else {
-                    str = str.substr(first, last - first + 1);
+                            std::map<std::string, std::string> &data_dict)
+{
+    try
+    {
+        auto root = nlohmann::json::parse(json);
+        code = root.at("code").get<int>();
+        message = root.at("message").get<std::string>();
+
+        if (root.contains("data"))
+        {
+            const auto &data = root["data"];
+            if (data.is_object())
+            {
+                for (auto it = data.begin(); it != data.end(); ++it)
+                {
+                    if (it->is_string())
+                        data_dict[it.key()] = it->get<std::string>();
+                    else
+                        data_dict[it.key()] = it->dump();
                 }
-            };
-
-            trim_quotes_and_space(key);
-            trim_quotes_and_space(value);
-
-            data_dict[key] = value;
+            }
         }
-
         return 0;
-    } catch (std::exception &e) {
+    }
+    catch (const std::exception &e)
+    {
         std::cerr << "Failed to parse JSON: " << e.what() << std::endl;
         message = e.what();
         code = -1;
         return 1;
     }
 }
 
-
 /**
  * This function is used to escape double quotes in a string.
  * @param input
@@ -127,21 +96,19 @@ size_t write_callback(void *contents, size_t size, size_t nmemb, std::string *ou
 
 int ask_from_videx_http(VidexJsonItem &request, VidexStringMap &res_json){
     const char *host_ip = "127.0.0.1:5001";
-    char value[1000];
 
     //VIDEX_SERVER
     if(videx_server)
-        host_ip = value;
+        host_ip = videx_server;
     std::string url = std::string("http://") + host_ip + "/ask_videx";
     CURL *curl;
     CURLcode res_code;
     std::string readBuffer;
-    curl = curl_easy_init(); // 初始化一个CURL easy handle。
+    curl = curl_easy_init();
     if(curl) {
         curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
         curl_easy_setopt(curl, CURLOPT_POST, 1);
 
-
         std::string request_str = request.to_json();
         curl_easy_setopt(curl, CURLOPT_POSTFIELDS, request_str.c_str());
 
@@ -171,6 +138,7 @@ int ask_from_videx_http(VidexJsonItem &request, VidexStringMap &res_json){
             std::string message;
             int error = videx_parse_simple_json(readBuffer.c_str(), code, message, res_json);
             if (error) {
+                std::cout << "videx_server raw response: " << readBuffer << std::endl;
                 std::cout << "!__!__!__!__!__! JSON parse error: " << message << '\n';
                 return 1;
             } else {

src/pg/videx/videx_json_item.cc

@@ -129,30 +129,36 @@ videx_table_block_relation_estimate_size(Relation rel, int32 *attr_widths,
 	/** try to fetch relation infos (pg_class) from videx-statistic-server, else use local pg_class instead*/
 	char *dbname = get_database_name(MyDatabaseId);
 	char *nspname   = get_namespace_name(rel->rd_rel->relnamespace);
+	char *relname = rel->rd_rel->relname.data;
+	std::string ns_relname = std::string(nspname) + "." + std::string(relname);
 	VidexStringMap res_json;
     VidexJsonItem request_item = construct_request(
-		dbname, rel->rd_rel->relname.data, nspname, __PRETTY_FUNCTION__);
+		dbname, nspname, ns_relname.c_str(), __PRETTY_FUNCTION__);
 	if (!ask_from_videx_http(request_item, res_json)) {
         /**fetch from videx-statistic-sever success, update local cache*/
+		curpages = (BlockNumber) std::stoi(res_json["relpages"]);
+		relpages = (BlockNumber) std::stoi(res_json["relpages"]);
+		reltuples = (double) std::stof(res_json["reltuples"]);
+		relallvisible = (BlockNumber) (res_json["relallvisible"] == "True" ? 1 : 0);
 		vac_update_relstats(rel,
-                        std::stoi(res_json["relpages"]),
-                        std::stof(res_json["reltuples"]),
-                        std::stoi(res_json["relallvisible"]),
-                        std::stoi(res_json["relhasindex"]),
+                        curpages,
+                        reltuples,
+                        relallvisible,
+                        res_json["relhasindex"] == "True" ? true : false,
                         InvalidTransactionId,
                         InvalidMultiXactId,
                         NULL,
                         NULL,
                         false);
-    }
+    } else {
+		/* it should have storage, so we can call the smgr */
+		curpages = (BlockNumber) rel->rd_rel->relpages;
 
-	/* it should have storage, so we can call the smgr */
-	curpages = (BlockNumber) rel->rd_rel->relpages;
-
-	/* coerce values in pg_class to more desirable types */
-	relpages = (BlockNumber) rel->rd_rel->relpages;
-	reltuples = (double) rel->rd_rel->reltuples;
-	relallvisible = (BlockNumber) rel->rd_rel->relallvisible;
+		/* coerce values in pg_class to more desirable types */
+		relpages = (BlockNumber) rel->rd_rel->relpages;
+		reltuples = (double) rel->rd_rel->reltuples;
+		relallvisible = (BlockNumber) rel->rd_rel->relallvisible;
+	}
 
 	/*
 	 * HACK: if the relation has never yet been vacuumed, use a minimum size
@@ -247,7 +253,7 @@ void videx_relation_set_new_filelocator (Relation rel,
 												 const RelFileLocator *newrlocator,
 												 char persistence,
 												 TransactionId *freezeXid,
-												 MultiXactId *minmulti){			
+												 MultiXactId *minmulti){		
 	return;									
 }
 

src/pg/videx/videxam.cc

@@ -166,7 +166,7 @@ def get_table_meta(self, db_name, schema_table_name):
         table = PGTable(
             dbname = db_name,
             table_schema = schema_name,
-            table_name = table_name,
+            table_name = schema_table_name,
             relpages = df['relpages'].values[0],
             reltuples = df['reltuples'].values[0],
             relallvisible = df['relallvisible'].values[0],

src/sub_platforms/sql_opt/databases/pg/pg_command.py

@@ -1,41 +1,112 @@
+import logging
+import json
 from typing import List
 
 from sub_platforms.sql_opt.videx.videx_metadata import VidexTableStats
 from sub_platforms.sql_opt.videx.model.videx_strategy import VidexModelBase, VidexStrategy
 from sub_platforms.sql_opt.videx.model.videx_model_innodb import VidexModelInnoDB
-from sub_platforms.sql_opt.videx.videx_metadata import VidexTableStats, VidexDBTaskStats
+from sub_platforms.sql_opt.videx.videx_metadata import VidexDBTaskStats
+from sub_platforms.sql_opt.videx.videx_pg_metadata import PGVidexTableStats
 from sub_platforms.sql_opt.videx.model.videx_strategy import VidexStrategy
 from sub_platforms.sql_opt.videx.videx_utils import IndexRangeCond
 from sub_platforms.sql_opt.pg_meta import PGTable
 
 class VidexModelPG(VidexModelBase):
-    def __init__(self, db_stats: VidexDBTaskStats, **kwargs):
+    def __init__(self, table_stats: PGVidexTableStats, **kwargs):
         super().__init__(None, VidexStrategy.postgresql)
-        self.videx_db_task_stats: VidexDBTaskStats = db_stats
+        self.table_stats: PGVidexTableStats = table_stats
 
     def scan_time(self, req_json_item: dict) -> float:
+        # no used in pg
         return 0.0
+
     def get_memory_buffer_size(self, req_json_item: dict) -> int:
+        # no used in pg
         return -1
 
+    def info_low(self, req_json_item: dict) -> int:
+        # no used in pg
+        return 0
+
     def cardinality(self, idx_range_cond: IndexRangeCond) -> int:
         return 0
 
     def ndv(self, index_name, field_list: List[str]) -> int:
-        return 0
+        if len(field_list) == 1:
+            colname = field_list[0]
+            col_stats_info = self.table_stats.table_statistic.statistic_dict.get(colname)
+            if col_stats_info is not None:
+                return int(col_stats_info.stadistinct)
+            else:
+                return 0
+        else:
+            #TODO: try to fetch from pg_statistic_ext for multi-column NDV,
+            # moreover, support ndv learned model (eg: PLM4NDV)
+            return 0
 
     def get_relation_stats(self, req_json_item: dict) -> dict:
-        return None
+        if self.table_stats.table_statistic is None:
+            logging.warning(f"Table statistic is None for "
+                            f"db_name: {self.table_stats.table_meta.dbname}, "
+                            f"table_name: {self.table_stats.table_meta.table_name}")
+            return {}
+        data_items = req_json_item.get("data")
+        if isinstance(data_items, list):
+            for item in data_items:
+                if item.get("item_type") == "colname":
+                    colname = (item.get("properties") or {}).get("name")
+                    if colname:
+                        break
+        if not colname:
+            logging.warning("Column name missing in request: %s", req_json_item)
+            return {}    
+        col_stats_info = self.table_stats.table_statistic.statistic_dict.get(colname)
+        if col_stats_info is None:
+            logging.warning(f"Column statistic not found for "
+                            f"db_name: {self.table_stats.table_meta.dbname}, "
+                            f"table_name: {self.table_stats.table_meta.table_name}, "
+                            f"req_json_item {req_json_item}")
+            return {}
+        ndv_value = self.ndv(None, [colname])
+        slots_payload = []
+        for slot in (col_stats_info.slots or []):
+            if hasattr(slot, "model_dump"):
+                slots_payload.append(slot.model_dump(exclude_none=True))
+            else:
+                slots_payload.append({
+                    "kind": getattr(slot, "kind", None),
+                    "op": getattr(slot, "op", None),
+                    "coll": getattr(slot, "coll", None),
+                    "numbers": getattr(slot, "numbers", None),
+                    "values": getattr(slot, "values", None),
+                })
+        res = {
+            "stanullfrac": col_stats_info.stanullfrac,
+            "stawidth": col_stats_info.stawidth,
+            "stainherit": col_stats_info.stainherit,
+            "stadistinct": ndv_value,
+            "slots": slots_payload,
+        }
+        logging.info(f"Get pg column statistic for "
+                     f"db_name: {self.table_stats.table_meta.dbname}, "
+                     f"table_name: {self.table_stats.table_meta.table_name}, "
+                     f"column {colname}: {res}")
+        return res
 
     def table_block_relation_estimate_size(self, req_json_item: dict) -> dict:
-        properties = req_json_item['properties']
-        videx_db = properties['dbname'].lower()
-        table_name = properties['table_name'].lower()
-        table: PGTable = self.videx_db_task_stats.get_table_meta(videx_db, table_name)
+        table : PGTable = self.table_stats.table_meta
+        logging.info(f"Start to get pg table block relation estimate size for "
+                     f"db_name: {table.dbname}, "
+                     f"table_name: {table.table_name}")
         res = {
             "relpages": table.relpages,
             "reltuples": table.reltuples,
             "relallvisible": table.relallvisible,
             "relhasindex": len(table.indexes) > 0,
         }
-        return res
+        logging.info(f"Get pg table block relation estimate size for "
+                     f"db_name: {table.dbname}, "
+                     f"table_name: {table.table_name}: {res}")
+        return res
+    def get_index_stats(self, req_json_item: dict) -> dict:
+        return NotImplementedError("get_index_stats is not implemented yet.")

src/sub_platforms/sql_opt/videx/model/videx_model_pg.py

@@ -155,12 +155,7 @@ def records_in_range(self, req_json_item: dict) -> int:
             }
         """
         return self.cardinality(idx_range_cond)
-    
-    def get_relation_stats(self, req_json_item: dict) -> dict:
-        pass
 
-    def table_block_relation_estimate_size(self, req_json_item: dict) -> dict:
-        pass
 
 def record_range_request_to_str(min_key: dict, max_key: dict) -> str:
     """