Report indexing failures on ft.info - MOD-5364

CMakeLists.txt

@@ -87,6 +87,7 @@ file(GLOB SOURCES
     "src/query_parser/v2/*.c"
     "src/util/*.c"
     "src/trie/*.c"
+    "src/info/*.c"
 
     "deps/cndict/cndict_data.c"
     "deps/libnu/*.c"
@@ -111,15 +112,15 @@ set(FINAL_OBJECTS
 
 if (BUILD_COORDINATOR)
     if (BUILD_COORD_OSS)
-        add_library(redisearch-oss STATIC ${SOURCES} ${FINAL_OBJECTS} src/module-init/module-init.c)
+        add_library(redisearch-oss STATIC ${SOURCES} ${FINAL_OBJECTS} src/module-init/module-init.c coord/src/rmr/reply.c)
 
         target_compile_definitions(redisearch-oss PRIVATE
             RS_NO_ONLOAD
             RS_NO_RMAPI
             RS_CLUSTER_OSS)
 
     elseif (BUILD_COORD_RLEC)
-        add_library(redisearch-enterprise STATIC ${SOURCES} ${FINAL_OBJECTS} src/module-init/module-init.c)
+        add_library(redisearch-enterprise STATIC ${SOURCES} ${FINAL_OBJECTS} src/module-init/module-init.c coord/src/rmr/reply.c)
 
         target_compile_definitions(redisearch-enterprise PRIVATE
             RS_NO_ONLOAD

coord/src/dist_aggregate.c

@@ -178,7 +178,7 @@ RSValue *MRReply_ToValue(MRReply *r) {
     case MR_REPLY_STATUS:
     case MR_REPLY_STRING: {
       size_t l;
-      char *s = MRReply_String(r, &l);
+      const char *s = MRReply_String(r, &l);
       v = RS_NewCopiedString(s, l);
       // v = RS_StringValT(s, l, RSString_Volatile);
       break;
@@ -397,7 +397,7 @@ static int rpnetNext(ResultProcessor *self, SearchResult *r) {
 
     // If an error was returned, propagate it
     if(MRReply_Type(nc->current.root) == MR_REPLY_ERROR) {
-      char *strErr = MRReply_String(nc->current.root, NULL);
+      const char *strErr = MRReply_String(nc->current.root, NULL);
       if (!strErr
           || strcmp(strErr, "Timeout limit was reached")
           || nc->areq->reqConfig.timeoutPolicy == TimeoutPolicy_Fail) {

coord/src/info_command.c

@@ -6,13 +6,14 @@
 
 #include "info_command.h"
 #include "resp3.h"
+#include "info/field_spec_info.h"
 
 // Type of field returned in INFO
 typedef enum {
   InfoField_WholeSum,
   InfoField_DoubleSum,
   InfoField_DoubleAverage,
-  InfoField_Max
+  InfoField_Max,
 } InfoFieldType;
 
 // Field specification
@@ -76,9 +77,13 @@
     size_t total_l;
     double total_d;
     struct {
-      double avg;
+      double sum;
       double count;
     } avg;
+    struct {
+      char *str;
+      size_t len;
+    } str;
   } u;
 } InfoValue;
 
@@ -91,6 +96,8 @@
   MRReply *indexOptions;
   size_t *errorIndexes;
   InfoValue toplevelValues[NUM_FIELDS_SPEC];
+  FieldSpecInfo *fieldSpecInfo_arr;
+  IndexError indexError;
   InfoValue gcValues[NUM_GC_FIELDS_SPEC];
   InfoValue cursorValues[NUM_CURSOR_FIELDS_SPEC];
   InfoValue dialectValues[NUM_DIALECT_FIELDS_SPEC];
@@ -115,29 +122,72 @@
 static void convertField(InfoValue *dst, MRReply *src, const InfoFieldSpec *spec) {
   int type = spec->type;
 
-  if (type == InfoField_WholeSum) {
-    long long tmp;
-    MRReply_ToInteger(src, &tmp);
-    dst->u.total_l += tmp;
-  } else if (type == InfoField_DoubleSum) {
-    double d;
-    MRReply_ToDouble(src, &d);
-    dst->u.total_d += d;
-  } else if (type == InfoField_DoubleAverage) {
-    dst->u.avg.count++;
-    double d;
-    MRReply_ToDouble(src, &d);
-    dst->u.avg.avg += d;
-  } else if (type == InfoField_Max) {
-    long long newVal;
-    MRReply_ToInteger(src, &newVal);
-    if (dst->u.total_l < newVal) {
-      dst->u.total_l = newVal;
+  switch (type) {
+    case InfoField_WholeSum: {
+      long long tmp;
+      MRReply_ToInteger(src, &tmp);
+      dst->u.total_l += tmp;
+      break;
+    }
+    case InfoField_DoubleSum: {
+      double d;
+      MRReply_ToDouble(src, &d);
+      dst->u.total_d += d;
+      break;
+    }
+    case InfoField_DoubleAverage: {
+      dst->u.avg.count++;
+      double d;
+      MRReply_ToDouble(src, &d);
+      dst->u.avg.sum += d;
+      break;
+    }
+    case InfoField_Max: {
+      long long newVal;
+      MRReply_ToInteger(src, &newVal);
+      if (dst->u.total_l < newVal) {
+        dst->u.total_l = newVal;
+      }
+      break;
     }
   }
   dst->isSet = 1;
 }
 
+// Extract an array of FieldSpecInfo from MRReply
+void handleFieldStatistics(MRReply *src, InfoFields *fields) {
+  // Input validations
+  RedisModule_Assert(src && fields);
+  RedisModule_Assert(MRReply_Type(src) == MR_REPLY_ARRAY);
+
+  size_t len = MRReply_Length(src);
+  if (!fields->fieldSpecInfo_arr) {
+    // Lazy initialization
+    fields->fieldSpecInfo_arr = array_new(FieldSpecInfo, len);
+    for (size_t i = 0; i < len; i++) {
+      FieldSpecInfo fieldSpecInfo = FieldSpecInfo_Init();
+      fields->fieldSpecInfo_arr = array_append(fields->fieldSpecInfo_arr, fieldSpecInfo);
+    }
+  }
+
+  for (size_t i = 0; i < len; i++) {
+    MRReply *serializedFieldSpecInfo = MRReply_ArrayElement(src, i);
+    FieldSpecInfo fieldSpecInfo = FieldSpecInfo_Deserialize(serializedFieldSpecInfo);
+    FieldSpecInfo_OpPlusEquals(&fields->fieldSpecInfo_arr[i], &fieldSpecInfo);
+    FieldSpecInfo_Clear(&fieldSpecInfo); // Free Resources
+  }
+}
+
+static void handleIndexError(InfoFields *fields, MRReply *src) {
+  // Check if indexError is initialized
+  if (!IndexError_LastError(&fields->indexError)) {
+    fields->indexError = IndexError_Init();
+  }
+  IndexError indexError = IndexError_Deserialize(src);
+  IndexError_OpPlusEquals(&fields->indexError, &indexError);
+  IndexError_Clear(indexError); // Free Resources
+}
+
 // Handle fields which aren't InfoValue types
 static void handleSpecialField(InfoFields *fields, const char *name, MRReply *value) {
   if (!strcmp(name, "index_name")) {
@@ -168,6 +218,10 @@
     processKvArray(fields, value, fields->cursorValues, cursorSpecs, NUM_CURSOR_FIELDS_SPEC, 1);
   } else if (!strcmp(name, "dialect_stats")) {
     processKvArray(fields, value, fields->dialectValues, dialectSpecs, NUM_DIALECT_FIELDS_SPEC, 1);
+  } else if (!strcmp(name, "field statistics")) {
+    handleFieldStatistics(value, fields);
+  } else if (!strcmp(name, IndexError_ObjectName)) {
+    handleIndexError(fields, value);
   }
 }
 
@@ -204,6 +258,15 @@
 }
 
 static void cleanInfoReply(InfoFields *fields) {
+  if (fields->fieldSpecInfo_arr) {
+    // Clear the info fields
+    for (size_t i = 0; i < array_len(fields->fieldSpecInfo_arr); i++) {
+      FieldSpecInfo_Clear(&fields->fieldSpecInfo_arr[i]);
+    }
+    array_free(fields->fieldSpecInfo_arr);
+    fields->fieldSpecInfo_arr = NULL;
+  }
+  IndexError_Clear(fields->indexError);
   rm_free(fields->errorIndexes);
 }
 
@@ -224,7 +287,7 @@
       RedisModule_ReplyKV_Double(reply, key, source->u.total_d);
     } else if (type == InfoField_DoubleAverage) {
       if (source->u.avg.count) {
-        RedisModule_ReplyKV_Double(reply, key, source->u.avg.avg / source->u.avg.count);
+        RedisModule_ReplyKV_Double(reply, key, source->u.avg.sum / source->u.avg.count);
       } else {
         RedisModule_ReplyKV_Double(reply, key, 0);
       }
@@ -241,7 +304,7 @@
   if (fields->indexName) {
     RedisModule_ReplyKV_StringBuffer(reply, "index_name", fields->indexName, fields->indexNameLen);
   }
-  
+
   if (fields->indexDef) {
     RedisModule_ReplyKV_MRReply(reply, "index_definition", fields->indexDef);
   }
@@ -268,12 +331,26 @@
 
   replyKvArray(reply, fields, fields->toplevelValues, toplevelSpecs_g, NUM_FIELDS_SPEC);
 
+
+  // Global index error stats
+  RedisModule_Reply_SimpleString(reply, IndexError_ObjectName);
+  IndexError_Reply(&fields->indexError, reply, 0);
+
+  if (fields->fieldSpecInfo_arr) {
+    RedisModule_ReplyKV_Array(reply, "field statistics"); //Field statistics
+    for (size_t i = 0; i < array_len(fields->fieldSpecInfo_arr); ++i) {
+      FieldSpecInfo_Reply(&fields->fieldSpecInfo_arr[i], reply, 0);
+    }
+    RedisModule_Reply_ArrayEnd(reply); // >Field statistics
+  }
+
+
   RedisModule_Reply_MapEnd(reply);
 }
 
 int InfoReplyReducer(struct MRCtx *mc, int count, MRReply **replies) {
   // Summarize all aggregate replies
-  InfoFields fields = {0};
+  InfoFields fields = { .indexError = IndexError_Init() };
   size_t numErrored = 0;
   MRReply *firstError = NULL;
   RedisModuleCtx *ctx = MRCtx_GetRedisCtx(mc);

coord/src/module.c

@@ -91,9 +91,9 @@ int uniqueStringsReducer(struct MRCtx *mc, int count, MRReply **replies) {
       nArrs++;
       for (size_t j = 0; j < MRReply_Length(replies[i]); j++) {
         size_t sl = 0;
-        char *s = MRReply_String(MRReply_ArrayElement(replies[i], j), &sl);
+        const char *s = MRReply_String(MRReply_ArrayElement(replies[i], j), &sl);
         if (s && sl) {
-          TrieMap_Add(dict, s, sl, NULL, NULL);
+          TrieMap_Add(dict, (char*)s, sl, NULL, NULL);
         }
       }
     } else if (MRReply_Type(replies[i]) == MR_REPLY_ERROR && err == NULL) {
@@ -782,7 +782,7 @@ searchResult *newResult_resp2(searchResult *cached, MRReply *arr, int j, searchR
     res->id = NULL;
     return res;
   }
-  res->id = MRReply_String(MRReply_ArrayElement(arr, j), &res->idLen);
+  res->id = (char*)MRReply_String(MRReply_ArrayElement(arr, j), &res->idLen);
   if (!res->id) {
     return res;
   }
@@ -844,7 +844,7 @@ searchResult *newResult_resp3(searchResult *cached, MRReply *results, int j, sea
   }
 
   MRReply *result_id = MRReply_MapElement(result_j, "id");
-  res->id = MRReply_String(result_id, &res->idLen);
+  res->id = (char*)MRReply_String(result_id, &res->idLen);
   if (!res->id) {
     return res;
   }
@@ -1513,14 +1513,11 @@ static int searchResultReducer_background(struct MRCtx *mc, int count, MRReply *
 }
 
 static bool should_return_error(MRReply *reply) {
-  // TODO: Replace second condition with a var instead of hard-coded string
-  char *errStr = MRReply_String(reply, NULL);
-  if (!errStr
-      || strcmp(errStr, "Timeout limit was reached")
-      || RSGlobalConfig.requestConfigParams.timeoutPolicy == TimeoutPolicy_Fail) {
-        return true;
-  }
-  return false;
+  // TODO: Replace third condition with a var instead of hard-coded string
+  const char *errStr = MRReply_String(reply, NULL);
+  return (!errStr
+          || RSGlobalConfig.requestConfigParams.timeoutPolicy == TimeoutPolicy_Fail
+          || strcmp(errStr, "Timeout limit was reached"));
 }
 
 static bool should_return_timeout_error(searchRequestCtx *req) {
@@ -1927,6 +1924,7 @@ int InfoCommandHandler(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
   }
   RS_AutoMemory(ctx);
   MRCommand cmd = MR_NewCommandFromRedisStrings(argc, argv);
+  MRCommand_Append(&cmd, WITH_INDEX_ERROR_TIME, strlen(WITH_INDEX_ERROR_TIME));
   MRCommand_SetProtocol(&cmd, ctx);
   MRCommand_SetPrefix(&cmd, "_FT");