Qdrant Restful 服务注册 以查询（Search API）为例

1.服务注册

        #[cfg(feature = "web")]
    {
        let dispatcher_arc = dispatcher_arc.clone();
        let settings = settings.clone();
        let handle = thread::Builder::new()
            .name("web".to_string())
            .spawn(move || {
                log_err_if_any(
                    "REST",
                    // 服务初始化
                    actix::init(
                        dispatcher_arc.clone(),
                        telemetry_collector,
                        health_checker,
                        settings,
                        logger_handle,
                    ),
                )
            })
            .unwrap();
        handles.push(handle);
    }


    // 
    let mut server = HttpServer::new(move || {
 

            let mut app = App::new()
                .wrap(Compress::default())  
                .wrap(ConditionEx::from_option(auth_keys.as_ref().map(
                    |auth_keys| Auth::new(auth_keys.clone(), api_key_whitelist.clone()),
                )))
                .wrap(Condition::new(settings.service.enable_cors, cors))
                .wrap(
                    // Set up logger, but avoid logging hot status endpoints
                    Logger::default()
                        .exclude("/")
                        .exclude("/metrics")
                        .exclude("/telemetry")
                        .exclude("/healthz")
                        .exclude("/readyz")
                        .exclude("/livez"),
                )
                .wrap(actix_telemetry::ActixTelemetryTransform::new(
                    actix_telemetry_collector.clone(),
                ))
                .app_data(dispatcher_data.clone())
                .app_data(telemetry_collector_data.clone())
                .app_data(logger_handle_data.clone())
                .app_data(http_client.clone())
                .app_data(debugger_state.clone())
                .app_data(health_checker.clone())
                .app_data(validate_path_config)
                .app_data(validate_query_config)
                .app_data(validate_json_config)
                .app_data(TempFileConfig::default().directory(&upload_dir))
                .app_data(MultipartFormConfig::default().total_limit(usize::MAX))
                .app_data(service_config.clone())
                // 注册服务
                .service(index)
                .configure(config_collections_api)
                .configure(config_snapshots_api)
                .configure(config_update_api)
                .configure(config_cluster_api)
                .configure(config_service_api)
                // 查询API
                .configure(config_search_api)
                .configure(config_recommend_api)
                .configure(config_discovery_api)
                .configure(config_query_api)
                .configure(config_facet_api)
                .configure(config_shards_api)
                .configure(config_issues_api)
                .configure(config_debugger_api)
                .configure(config_local_shard_api)
                 
                .service(scroll_points)
                .service(count_points)
                .service(get_point)
                .service(get_points);


            app
        })
        .workers(max_web_workers(&settings));      

2. Search API

根据向量来返回查询相似记录。按照score来排序。

POST /collections/{collection_name}/points/query
{
    "query": [0.2, 0.1, 0.9, 0.79],
    "filter": {
        "must": [
            {
                "key": "city",
                "match": {
                    "value": "London"
                }
            }
        ]
    },
    "params": {
        "hnsw_ef": 128,
        "exact": false
    },
    "limit": 3
}

响应

{
  "result": [
    { "id": 10, "score": 0.81 }, // score 是非常重要的
    { "id": 14, "score": 0.75 },
    { "id": 11, "score": 0.73 }
  ],
  "status": "ok",
  "time": 0.001
}

函数入口

pub fn config_search_api(cfg: &mut web::ServiceConfig) {
    cfg.service(search_points)
        .service(batch_search_points)
        .service(search_point_groups)
        .service(search_points_matrix_pairs)
        .service(search_points_matrix_offsets);
}



#[post("/collections/{name}/points/search")]
async fn search_points(
    dispatcher: web::Data<Dispatcher>,
    collection: Path<CollectionPath>,
    request: Json<SearchRequest>,
    params: Query<ReadParams>,
    service_config: web::Data<ServiceConfig>,
    ActixAccess(access): ActixAccess,
) -> HttpResponse {
  
    // 查询入口
    let result = do_core_search_points(
        dispatcher.toc(&access, &pass),
        &collection.name,
        search_request.into(),
        params.consistency,
        shard_selection,
        access,
        params.timeout(),
        request_hw_counter.get_counter(),
    )
    .await
    .map(|scored_points| {
        scored_points
            .into_iter()
            .map(api::rest::ScoredPoint::from)
            .collect_vec()
    });

}


pub async fn do_core_search_points(
    toc: &TableOfContent,
    collection_name: &str,
    request: CoreSearchRequest,
    read_consistency: Option<ReadConsistency>,
    shard_selection: ShardSelectorInternal,
    access: Access,
    timeout: Option<Duration>,
    hw_measurement_acc: HwMeasurementAcc,
) -> Result<Vec<ScoredPoint>, StorageError> {
    let batch_res = do_core_search_batch_points(
        toc,
        collection_name,
        CoreSearchRequestBatch {
            searches: vec![request],
        },
        read_consistency,
        shard_selection,
        access,
        timeout,
        hw_measurement_acc,
    )
    .await?;
}


    pub async fn core_search_batch(
        &self,
        collection_name: &str,
        mut request: CoreSearchRequestBatch,
        read_consistency: Option<ReadConsistency>,
        shard_selection: ShardSelectorInternal,
        access: Access,
        timeout: Option<Duration>,
        hw_measurement_acc: HwMeasurementAcc,
    ) -> StorageResult<Vec<Vec<ScoredPoint>>> {

        // 批量查询
        collection
            .core_search_batch(
                request,
                read_consistency,
                shard_selection,
                timeout,
                hw_measurement_acc,
            )
            .await
            .map_err(|err| err.into())
    }

    // 最终调用函数

        pub async fn core_search(
        &self,
        request: Arc<CoreSearchRequestBatch>,
        read_consistency: Option<ReadConsistency>,
        local_only: bool,
        timeout: Option<Duration>,
        hw_measurement_acc: HwMeasurementAcc,
    ) -> CollectionResult<Vec<Vec<ScoredPoint>>> {
        self.execute_and_resolve_read_operation(
            |shard| {
                let request = Arc::clone(&request);
                let search_runtime = self.search_runtime.clone();
                let hw_measurement_acc_clone = hw_measurement_acc.clone();
                async move {
                    shard
                        .core_search(request, &search_runtime, timeout, hw_measurement_acc_clone)
                        .await
                }
                .boxed()
            },
            read_consistency,
            local_only,
        )
        .await
    }

    // 不同的方式会有不同的查询实现以local为例
    async fn core_search(
        &self,
        request: Arc<CoreSearchRequestBatch>,
        search_runtime_handle: &Handle,
        timeout: Option<Duration>,
        hw_measurement_acc: HwMeasurementAcc,
    ) -> CollectionResult<Vec<Vec<ScoredPoint>>>;


    impl LocalShard {
    pub async fn do_search(
        &self,
        core_request: Arc<CoreSearchRequestBatch>,
        search_runtime_handle: &Handle,
        timeout: Option<Duration>,
        hw_counter_acc: HwMeasurementAcc,
    ) -> CollectionResult<Vec<Vec<ScoredPoint>>> {


        let search_request = SegmentsSearcher::search(
            Arc::clone(&self.segments),
            Arc::clone(&core_request),
            search_runtime_handle,
            true,
            query_context,
        );
    }
}



 pub async fn search(
        segments: LockedSegmentHolder,
        batch_request: Arc<CoreSearchRequestBatch>,
        runtime_handle: &Handle,
        sampling_enabled: bool,
        query_context: QueryContext,
    ) -> CollectionResult<Vec<Vec<ScoredPoint>>> {
        let search = runtime_handle.spawn_blocking({
                        let (segment, batch_request) = (segment.clone(), batch_request.clone());
                        move || {
                            let segment_query_context =
                                query_context_arc_segment.get_segment_query_context();

                            search_in_segment(
                                segment,
                                batch_request,
                                use_sampling,
                                &segment_query_context,
                            )
                        }
                    });
    }



fn search_in_segment(
    segment: LockedSegment,
    request: Arc<CoreSearchRequestBatch>,
    use_sampling: bool,
    segment_query_context: &SegmentQueryContext,
) -> CollectionResult<(Vec<Vec<ScoredPoint>>, Vec<bool>)> {

    for search_query in &request.searches {
        let with_payload_interface = search_query
            .with_payload
            .as_ref()
            .unwrap_or(&WithPayloadInterface::Bool(false));
        // 查询条件 和 RESTFUL 接口中对应
        let params = BatchSearchParams {
            search_type: search_query.query.as_ref().into(),
            vector_name: search_query.query.get_vector_name(),
            filter: search_query.filter.as_ref(),
            with_payload: WithPayload::from(with_payload_interface),
            with_vector: search_query.with_vector.clone().unwrap_or_default(),
            top: search_query.limit + search_query.offset,
            params: search_query.params.as_ref(),
        };

        let query = search_query.query.clone().into();

        if params == prev_params {
            vectors_batch.push(query);
        } else {
            if !vectors_batch.is_empty() {
                // 批量查询
                let (mut res, mut further) = execute_batch_search(
                    &segment,
                    &vectors_batch,
                    &prev_params,
                    use_sampling,
                    segment_query_context,
                )?;
                further_results.append(&mut further);
                result.append(&mut res);
                vectors_batch.clear()
            }
            vectors_batch.push(query);
            prev_params = params;
        }
    }


}


  fn search_batch(
        &self,
        vector_name: &VectorName,
        query_vectors: &[&QueryVector],
        with_payload: &WithPayload,
        with_vector: &WithVector,
        filter: Option<&Filter>,
        top: usize,
        params: Option<&SearchParams>,
        query_context: &SegmentQueryContext,
    ) -> OperationResult<Vec<Vec<ScoredPoint>>> {
        // 查询
        let internal_results = vector_data.vector_index.borrow().search(
            query_vectors,
            filter,
            top,
            params,
            &vector_query_context,
        )?;
    }


    // 返回值中的分数
    fn search(
        &self,
        vectors: &[&QueryVector],
        filter: Option<&Filter>,
        top: usize,
        params: Option<&SearchParams>,
        query_context: &VectorQueryContext,
    ) -> OperationResult<Vec<Vec<ScoredPointOffset>>> {
        match self {
            VectorIndexEnum::Plain(index) => {
                index.search(vectors, filter, top, params, query_context)
            }
           
        }
    }

    // 不同类型计算分数不一样
    VectorStorageEnum::DenseSimple(vs) => {
            raw_scorer_impl(query, vs, point_deleted, is_stopped, hc)
    }

    pub fn raw_scorer_impl<'a, TVectorStorage: DenseVectorStorage<VectorElementType>>(
    query: QueryVector,
    vector_storage: &'a TVectorStorage,
    point_deleted: &'a BitSlice,
    is_stopped: &'a AtomicBool,
    hardware_counter: HardwareCounterCell,
    ) -> OperationResult<Box<dyn RawScorer + 'a>> {
        match vector_storage.distance() {
            // 余弦
            Distance::Cosine => new_scorer_with_metric::<CosineMetric, _>(
                query,
                vector_storage,
                point_deleted,
                is_stopped,
                hardware_counter,
            ),
            // 欧拉距离
            Distance::Euclid => new_scorer_with_metric::<EuclidMetric, _>(
                query,
                vector_storage,
                point_deleted,
                is_stopped,
                hardware_counter,
            ),
            // 点乘
            Distance::Dot => new_scorer_with_metric::<DotProductMetric, _>(
                query,
                vector_storage,
                point_deleted,
                is_stopped,
                hardware_counter,
            ),
            Distance::Manhattan => new_scorer_with_metric::<ManhattanMetric, _>(
                query,
                vector_storage,
                point_deleted,
                is_stopped,
                hardware_counter,
            ),
        }
    }