reduce overhead index requests (#342)

* reduce overhead requests

Signed-off-by: ruiyi.jiang <ruiyi.jiang@zilliz.com>

* update index cache ttl

Signed-off-by: ruiyi.jiang <ruiyi.jiang@zilliz.com>

* code clean

Signed-off-by: ruiyi.jiang <ruiyi.jiang@zilliz.com>

* support clear cache

Signed-off-by: ruiyi.jiang <ruiyi.jiang@zilliz.com>

* let => const

Signed-off-by: ryjiang <jiangruiyi@gmail.com>

---------

Signed-off-by: ruiyi.jiang <ruiyi.jiang@zilliz.com>
Signed-off-by: ryjiang <jiangruiyi@gmail.com>

client/src/http/BaseModel.ts

@@ -50,6 +50,8 @@ export default class BaseModel {
  } as any;
  if (timeout) httpConfig.timeout = timeout;
  const res = await http(httpConfig);
+ // conflict with collection view data structure, status is useless, so delete here.
+ delete res.data.data.status;
  return new this(res.data.data || {}) as T;
  }
 

client/src/http/Collection.ts

@@ -42,13 +42,20 @@ export class Collection extends BaseModel implements CollectionData {
  return super.findAll({ path: this.COLLECTIONS_URL, params: data || {} });
  }
 
- static getCollection(name: string) {
+ static getCollectionWithIndexInfo(name: string) {
  return super.search<Collection>({
  path: `${this.COLLECTIONS_URL}/${name}`,
  params: {},
  });
  }
 
+ static getCollectionInfo(collectionName: string) {
+ return super.search<Collection>({
+ path: `${this.COLLECTIONS_URL}/${collectionName}/info`,
+ params: {},
+ });
+ }
+
  static createCollection(data: any) {
  return super.create({ path: this.COLLECTIONS_URL, data });
  }

client/src/http/MilvusIndex.ts

@@ -49,6 +49,12 @@ export class MilvusIndex extends BaseModel {
  return super.batchDelete({ path, data: { ...param, type } });
  }
 
+ static async flush() {
+ const path = `${this.BASE_URL}/flush`;
+
+ return super.query({ path, data: {} });
+ }
+
  get indexType() {
  return this.params.find(p => p.key === 'index_type')?.value || '';
  }

client/src/pages/collections/Collections.tsx

@@ -9,7 +9,7 @@ import {
  dataContext,
  webSocketContext,
 } from '@/context';
-import { Collection, MilvusService, DataService } from '@/http';
+import { Collection, MilvusService, DataService, MilvusIndex } from '@/http';
 import { useNavigationHook, usePaginationHook } from '@/hooks';
 import { ALL_ROUTER_TYPES } from '@/router/Types';
 import AttuGrid from '@/components/grid/Grid';
@@ -123,6 +123,10 @@ const Collections = () => {
  }
  }, [setCollections, checkCollectionStatus]);
 
+ const clearIndexCache = useCallback(async () => {
+ await MilvusIndex.flush();
+ }, []);
+
  useEffect(() => {
  fetchData();
  }, [fetchData, database]);
@@ -412,6 +416,7 @@ const Collections = () => {
  {
  type: 'iconBtn',
  onClick: () => {
+ clearIndexCache();
  fetchData();
  },
  label: collectionTrans('delete'),

client/src/pages/partitions/Partitions.tsx

@@ -81,7 +81,7 @@ const Partitions: FC<{
  };
 
  const fetchCollectionDetail = async (name: string) => {
- const res = await Collection.getCollection(name);
+ const res = await Collection.getCollectionInfo(name);
  return res;
  };
 

client/src/pages/preview/Preview.tsx

@@ -45,7 +45,7 @@ const Preview: FC<{
 
  const loadData = async (collectionName: string) => {
  // get schema list
- const collection = await Collection.getCollection(collectionName);
+ const collection = await Collection.getCollectionInfo(collectionName);
 
  const schemaList = collection.fields!;
  let nameList = schemaList.map(v => ({

client/src/pages/query/Query.tsx

@@ -66,7 +66,7 @@ const Query: FC<{
  };
 
  const getFields = async (collectionName: string) => {
- const collection = await Collection.getCollection(collectionName);
+ const collection = await Collection.getCollectionInfo(collectionName);
  const schemaList = collection.fields;
 
  const nameList = schemaList.map(v => ({

client/src/pages/schema/Schema.tsx

@@ -85,7 +85,9 @@ const Schema: FC<{
  const KeyIcon = icons.key;
 
  try {
- const collection = await Collection.getCollection(collectionName);
+ const collection = await Collection.getCollectionWithIndexInfo(
+ collectionName
+ );
  const fields = collection.fieldWithIndexInfo.map(f =>
  Object.assign(f, {
  _fieldNameElement: (

client/src/pages/search/VectorSearch.tsx

@@ -15,7 +15,7 @@ import SimpleMenu from '@/components/menu/SimpleMenu';
 import { Option } from '@/components/customSelector/Types';
 import Filter from '@/components/advancedSearch';
 import { Field } from '@/components/advancedSearch/Types';
-import { Collection, MilvusIndex } from '@/http';
+import { Collection } from '@/http';
 import {
  parseValue,
  parseLocationSearch,
@@ -253,15 +253,17 @@ const VectorSearch = () => {
 
  const fetchFieldsWithIndex = useCallback(
  async (collectionName: string, collections: Collection[]) => {
- const fields =
- collections.find(c => c.collectionName === collectionName)?.fields ||
- [];
- const indexes = await MilvusIndex.getIndexInfo(collectionName);
+ const col = collections.find(c => c.collectionName === collectionName);
+
+ const fields = col?.fields ?? [];
 
  const { vectorFields, nonVectorFields } = classifyFields(fields);
 
  // only vector type fields can be select
- const fieldOptions = getVectorFieldOptions(vectorFields, indexes);
+ const fieldOptions = getVectorFieldOptions(
+ vectorFields,
+ col?.indexes ?? []
+ );
  setFieldOptions(fieldOptions);
  if (fieldOptions.length > 0) {
  // set first option value as default field value
@@ -351,10 +353,7 @@ const VectorSearch = () => {
 
  setTableLoading(true);
  try {
- const res = await Collection.vectorSearchData(
- selectedCollection,
- params
- );
+ const res = await Collection.vectorSearchData(selectedCollection, params);
  setTableLoading(false);
  setSearchResult(res.results);
  setLatency(res.latency);

client/src/pages/segments/Segments.tsx

@@ -28,7 +28,8 @@ const Segments: FC<{
  const qsegments = (await Segement.getQSegments(collectionName)) || {};
 
  const combinedArray = psegments.infos.map(p => {
- const q = qsegments.infos.find(q => q.segmentID === p.segmentID)! as any;
+ const q: any =
+ qsegments.infos.find(q => q.segmentID === p.segmentID)! || {};
  return {
  ...p,
  ...Object.keys(q).reduce((acc, key) => {

package.json

@@ -14,5 +14,8 @@
  "private": true,
  "dependencies": {
  "react-router-dom": "^6.20.0"
+ },
+ "devDependencies": {
+ "@types/lru-cache": "^7.10.10"
  }
 }

server/package.json

@@ -26,7 +26,7 @@
  "glob": "^7.2.0",
  "helmet": "^7.0.0",
  "http-errors": "^2.0.0",
- "lru-cache": "^6.0.0",
+ "lru-cache": "^10.1.0",
  "morgan": "^1.10.0",
  "node-cron": "^3.0.2",
  "rimraf": "^5.0.1",
@@ -55,7 +55,7 @@
  "@types/glob": "^8.1.0",
  "@types/http-errors": "^2.0.1",
  "@types/jest": "^29.5.3",
- "@types/lru-cache": "^5.1.1",
+ "@types/lru-cache": "^7.10.10",
  "@types/morgan": "^1.9.4",
  "@types/node": "^20.4.2",
  "@types/node-cron": "^3.0.8",

server/src/app.ts

@@ -3,7 +3,7 @@ import cors from 'cors';
 import helmet from 'helmet';
 import * as http from 'http';
 import { Server, Socket } from 'socket.io';
-import LruCache from 'lru-cache';
+import { LRUCache } from 'lru-cache';
 import * as path from 'path';
 import chalk from 'chalk';
 import { router as connectRouter } from './milvus';
@@ -21,15 +21,21 @@ import {
  ErrorMiddleware,
  ReqHeaderMiddleware,
 } from './middleware';
-import { EXPIRED_TIME, CACHE_KEY } from './utils';
+import { CLIENT_TTL, INDEX_TTL } from './utils';
 import { getIp } from './utils/Network';
+import { DescribeIndexResponse, MilvusClient } from './types';
 // initialize express app
 export const app = express();
 
 // initialize cache store
-const cache = new LruCache({
- maxAge: EXPIRED_TIME,
- updateAgeOnGet: true,
+export const clientCache = new LRUCache<string, MilvusClient>({
+ ttl: CLIENT_TTL,
+ ttlAutopurge: true,
+});
+
+export const indexCache = new LRUCache<string, DescribeIndexResponse>({
+ ttl: INDEX_TTL,
+ ttlAutopurge: true,
 });
 
 // initialize express router
@@ -52,9 +58,8 @@ router.get('/healthy', (req, res, next) => {
 const server = http.createServer(app);
 // default port 3000
 const PORT = 3000;
+
 // setup middlewares
-// use cache
-app.set(CACHE_KEY, cache);
 // use cors https://expressjs.com/en/resources/middleware/cors.html
 app.use(cors());
 // use helmet https://github.com/helmetjs/helmet

server/src/collections/collections.controller.ts

@@ -50,7 +50,10 @@ export class CollectionController {
  this.renameCollection.bind(this)
  );
  this.router.delete('/:name/alias/:alias', this.dropAlias.bind(this));
+ // collection with index info
  this.router.get('/:name', this.describeCollection.bind(this));
+ // just collection info
+ this.router.get('/:name/info', this.getCollectionInfo.bind(this));
  this.router.get('/:name/count', this.count.bind(this));
 
  // load / release
@@ -164,6 +167,18 @@ export class CollectionController {
  }
  }
 
+ async getCollectionInfo(req: Request, res: Response, next: NextFunction) {
+ const name = req.params?.name;
+ try {
+ const result = await this.collectionsService.describeCollection({
+ collection_name: name,
+ });
+ res.send(result);
+ } catch (error) {
+ next(error);
+ }
+ }
+
  async getCollectionStatistics(
  req: Request,
  res: Response,

server/src/collections/collections.service.ts

@@ -4,7 +4,6 @@ import {
  DescribeCollectionReq,
  DropCollectionReq,
  GetCollectionStatisticsReq,
- GetIndexStateReq,
  InsertReq,
  LoadCollectionReq,
  ReleaseLoadCollectionReq,
@@ -27,9 +26,14 @@ import { Parser } from '@json2csv/plainjs';
 import { throwErrorFromSDK, findKeyValue, genRows, ROW_COUNT } from '../utils';
 import { QueryDto, ImportSampleDto, GetReplicasDto } from './dto';
 import { CollectionData } from '../types';
+import { SchemaService } from '../schema/schema.service';
 
 export class CollectionsService {
- constructor(private milvusService: MilvusService) {}
+ private schemaService: SchemaService;
+
+ constructor(private milvusService: MilvusService) {
+ this.schemaService = new SchemaService(milvusService);
+ }
 
  async getCollections(data?: ShowCollectionsReq) {
  const res = await this.milvusService.client.showCollections(data);
@@ -151,18 +155,6 @@ export class CollectionsService {
  return res;
  }
 
- /**
- * We do not throw error for this.
- * Because if collection dont have index, it will throw error.
- * We need wait for milvus error code.
- * @param data
- * @returns
- */
- async getIndexInfo(data: GetIndexStateReq) {
- const res = await this.milvusService.client.describeIndex(data);
- return res;
- }
-
  /**
  * Get all collections meta data
  * @returns {id:string, collection_name:string, schema:Field[], autoID:boolean, rowCount: string, consistency_level:string}
@@ -190,7 +182,7 @@ export class CollectionsService {
  });
 
  // get index info for collections
- const indexRes = await this.getIndexInfo({
+ const indexRes = await this.schemaService.describeIndex({
  collection_name: item.name,
  });
 
@@ -295,7 +287,7 @@ export class CollectionsService {
  const res = await this.getCollections();
  if (res.data.length > 0) {
  for (const item of res.data) {
- const indexRes = await this.getIndexInfo({
+ const indexRes = await this.schemaService.describeIndex({
  collection_name: item.name,
  });
  data.push({

server/src/middleware/index.ts

@@ -2,16 +2,16 @@ import { Request, Response, NextFunction } from 'express';
 import morgan from 'morgan';
 import chalk from 'chalk';
 import { MilvusService } from '../milvus/milvus.service';
-import { CACHE_KEY, MILVUS_ADDRESS, HTTP_STATUS_CODE } from '../utils';
+import { MILVUS_ADDRESS, HTTP_STATUS_CODE } from '../utils';
 import { HttpError } from 'http-errors';
 import HttpErrors from 'http-errors';
+import { clientCache } from '../app';
 
 export const ReqHeaderMiddleware = (
  req: Request,
  res: Response,
  next: NextFunction
 ) => {
- const cache = req.app.get(CACHE_KEY);
  // all ape requests need set milvus address in header.
  // server will set active address in milvus service.
  const milvusAddress = (req.headers[MILVUS_ADDRESS] as string) || '';
@@ -20,10 +20,10 @@ export const ReqHeaderMiddleware = (
  // only api request has MILVUS_ADDRESS.
  // When client run in express, we dont need static files like: xx.js run this logic.
  // Otherwise will cause 401 error.
- if (milvusAddress && cache.has(milvusAddress)) {
+ if (milvusAddress && clientCache.has(milvusAddress)) {
  MilvusService.activeAddress = milvusAddress;
  // insight cache will update expire time when use insightCache.get
- MilvusService.activeMilvusClient = cache.get(milvusAddress);
+ MilvusService.activeMilvusClient = clientCache.get(milvusAddress);
  }
 
  const CONNECT_URL = `/api/v1/milvus/connect`;

server/src/milvus/milvus.controller.ts

@@ -2,7 +2,6 @@ import { NextFunction, Request, Response, Router } from 'express';
 import { dtoValidationMiddleware } from '../middleware/validation';
 import { MilvusService } from './milvus.service';
 import { ConnectMilvusDto, FlushDto, UseDatabaseDto } from './dto';
-import { CACHE_KEY } from '../utils';
 import packageJson from '../../package.json';
 
 export class MilvusController {
@@ -44,12 +43,13 @@ export class MilvusController {
 
  async connectMilvus(req: Request, res: Response, next: NextFunction) {
  const { address, username, password, database } = req.body;
- const cache = req.app.get(CACHE_KEY);
  try {
- const result = await this.milvusService.connectMilvus(
- { address, username, password, database },
- cache
- );
+ const result = await this.milvusService.connectMilvus({
+ address,
+ username,
+ password,
+ database,
+ });
 
  res.send(result);
  } catch (error) {
@@ -60,10 +60,9 @@ export class MilvusController {
 
  async checkConnect(req: Request, res: Response, next: NextFunction) {
  const address = '' + req.query?.address;
- const cache = req.app.get(CACHE_KEY);
 
  try {
- const result = await this.milvusService.checkConnect(address, cache);
+ const result = await this.milvusService.checkConnect(address);
  res.send(result);
  } catch (error) {
  next(error);