Optimize gRPC Response Parsing Performance (#553)

## Problem

The current implementation uses `json_format.MessageToDict` to convert
entire protobuf messages to dictionaries when parsing gRPC responses.
This is a significant CPU bottleneck when processing large numbers of
vectors, as reported in PR #537 where users experienced ~100
vectors/second throughput.

The `MessageToDict` conversion is expensive because it:
1. Serializes the entire protobuf message to JSON
2. Deserializes it back into a Python dictionary
3. Does this for every field, even when we only need specific fields

Additionally, several other performance issues were identified:
- Metadata conversion using `MessageToDict` on `Struct` messages
- Inefficient list construction (append vs pre-allocation)
- Unnecessary dict creation for `SparseValues` parsing
- Response header processing overhead

## Solution

Optimized all gRPC response parsing functions in
`pinecone/grpc/utils.py` to directly access protobuf fields instead of
converting entire messages to dictionaries. This approach:

1. **Directly accesses protobuf fields**: Uses `response.vectors`,
`response.matches`, `response.namespace`, etc. directly
2. **Optimized metadata conversion**: Created `_struct_to_dict()` helper
that directly accesses `Struct` fields (~1.5-2x faster than
`MessageToDict`)
3. **Pre-allocates lists**: Uses `[None] * len()` for known-size lists
(~6.5% improvement)
4. **Direct SparseValues creation**: Creates `SparseValues` objects
directly instead of going through dict conversion (~410x faster)
5. **Caches protobuf attributes**: Stores repeated attribute accesses in
local variables
6. **Optimized response info extraction**: Improved
`extract_response_info()` performance with module-level constants and
early returns
7. **Maintains backward compatibility**: Output format remains identical
to the previous implementation

## Performance Impact

Performance testing of the response parsing functions show significant
improvements across all optimized functions.

## Changes

### Modified Files
- `pinecone/grpc/utils.py`: Optimized 9 response parsing functions with
direct protobuf field access
- Added `_struct_to_dict()` helper for optimized metadata conversion
(~1.5-2x faster)
  - Pre-allocated lists where size is known (~6.5% improvement)
  - Direct `SparseValues` creation (removed dict conversion overhead)
  - Cached protobuf message attributes
  - Removed dead code paths (dict fallback in `parse_usage`)
- `pinecone/grpc/index_grpc.py`: Updated to pass protobuf messages
directly to parse functions
- `pinecone/grpc/resources/vector_grpc.py`: Updated to pass protobuf
messages directly to parse functions
- `pinecone/utils/response_info.py`: Optimized `extract_response_info()`
with module-level constants and early returns
- `tests/perf/test_fetch_response_optimization.py`: New performance
tests for fetch response parsing
- `tests/perf/test_query_response_optimization.py`: New performance
tests for query response parsing
- `tests/perf/test_other_parse_methods.py`: New performance tests for
all other parse methods
- `tests/perf/test_grpc_parsing_perf.py`: Extended with additional
benchmarks

### Technical Details

**Core Optimizations**:

1. **`_struct_to_dict()` Helper Function**:
   - Directly accesses protobuf `Struct` and `Value` fields
   - Handles all value types (null, number, string, bool, struct, list)
   - Recursively processes nested structures
- ~1.5-2x faster than `json_format.MessageToDict` for metadata
conversion

2. **List Pre-allocation**:
- `parse_query_response`: Pre-allocates `matches` list with `[None] *
len(matches_proto)`
   - `parse_list_namespaces_response`: Pre-allocates `namespaces` list
   - ~6.5% performance improvement over append-based construction

3. **Direct SparseValues Creation**:
- Replaced `parse_sparse_values(dict)` with direct
`SparseValues(indices=..., values=...)` creation
   - ~410x faster (avoids dict creation and conversion overhead)

## Testing

- All existing unit tests pass (224 tests in `tests/unit_grpc`)
- Comprehensive pytest benchmark tests added for all optimized
functions:
- `test_fetch_response_optimization.py`: Tests for fetch response with
varying metadata sizes
- `test_query_response_optimization.py`: Tests for query response with
varying match counts, dimensions, metadata sizes, and sparse vectors
- `test_other_parse_methods.py`: Tests for all other parse methods
(fetch_by_metadata, list_namespaces, stats, upsert, update,
namespace_description)
- Mypy type checking passes with and without grpc extras (with types
extras)
- No breaking changes - output format remains identical

## Related

This addresses the performance issue reported in PR #537, implementing a
similar optimization approach but adapted for the current codebase
structure. All parse methods have been optimized with comprehensive
performance testing to verify improvements.

Author: jhamon

pinecone/grpc/index_grpc.py

@@ -3,7 +3,6 @@
 import logging
 from typing import List, Any, Iterable, cast, Literal, Iterator, TYPE_CHECKING
 
-from google.protobuf import json_format
 
 from pinecone.utils.tqdm import tqdm
 from pinecone.utils import require_kwargs
@@ -15,6 +14,7 @@
     parse_fetch_response,
     parse_fetch_by_metadata_response,
     parse_query_response,
+    query_response_to_dict,
     parse_stats_response,
     parse_upsert_response,
     parse_update_response,
@@ -41,6 +41,7 @@
 from pinecone.core.grpc.protos.db_data_2025_10_pb2 import (
     Vector as GRPCVector,
     QueryVector as GRPCQueryVector,
+    QueryResponse as ProtoQueryResponse,
     UpsertRequest,
     DeleteRequest,
     QueryRequest,
@@ -501,13 +502,13 @@ def _query(
         include_metadata: bool | None = None,
         sparse_vector: (SparseValues | GRPCSparseValues | SparseVectorTypedDict) | None = None,
         **kwargs,
-    ) -> tuple[dict[str, Any], dict[str, str] | None]:
+    ) -> tuple[ProtoQueryResponse, dict[str, str] | None]:
         """
-        Low-level query method that returns raw JSON dict and initial metadata without parsing.
+        Low-level query method that returns protobuf Message and initial metadata without parsing.
         Used internally by query() and query_namespaces() for performance.
 
         Returns:
-            Tuple of (json_dict, initial_metadata). initial_metadata may be None.
+            Tuple of (protobuf_message, initial_metadata). initial_metadata may be None.
         """
         if vector is not None and id is not None:
             raise ValueError("Cannot specify both `id` and `vector`")
@@ -535,7 +536,7 @@ def _query(
 
         timeout = kwargs.pop("timeout", None)
         response, initial_metadata = self.runner.run(self.stub.Query, request, timeout=timeout)
-        return json_format.MessageToDict(response), initial_metadata
+        return response, initial_metadata
 
     def query(
         self,
@@ -626,8 +627,8 @@ def query(
                 future, result_transformer=parse_query_response, timeout=timeout
             )
         else:
-            # For sync requests, use _query to get raw dict and metadata, then parse it
-            json_response, initial_metadata = self._query(
+            # For sync requests, use _query to get protobuf Message and metadata, then parse it
+            response, initial_metadata = self._query(
                 vector=vector,
                 id=id,
                 namespace=namespace,
@@ -640,7 +641,7 @@ def query(
                 **kwargs,
             )
             return parse_query_response(
-                json_response, _check_type=False, initial_metadata=initial_metadata
+                response, _check_type=False, initial_metadata=initial_metadata
             )
 
     def query_namespaces(
@@ -681,8 +682,9 @@ def query_namespaces(
 
         only_futures = cast(Iterable[Future], futures)
         for response in as_completed(only_futures):
-            json_response, _ = response.result()  # Ignore initial_metadata for query_namespaces
-            # Pass raw dict directly to aggregator - no parsing needed
+            proto_response, _ = response.result()  # Ignore initial_metadata for query_namespaces
+            # Convert protobuf Message to dict format for aggregator using optimized helper
+            json_response = query_response_to_dict(proto_response)
             aggregator.add_results(json_response)
 
         final_results = aggregator.get_results()
@@ -946,8 +948,7 @@ def describe_index_stats(
 
         request = DescribeIndexStatsRequest(**args_dict)
         response, _ = self.runner.run(self.stub.DescribeIndexStats, request, timeout=timeout)
-        json_response = json_format.MessageToDict(response)
-        return parse_stats_response(json_response)
+        return parse_stats_response(response)
 
     @require_kwargs
     def create_namespace(

pinecone/grpc/resources/vector_grpc.py

@@ -3,7 +3,6 @@
 import logging
 from typing import Any, Iterable, cast, Literal
 
-from google.protobuf import json_format
 
 from pinecone.utils.tqdm import tqdm
 from concurrent.futures import as_completed, Future
@@ -13,6 +12,7 @@
     parse_fetch_response,
     parse_fetch_by_metadata_response,
     parse_query_response,
+    query_response_to_dict,
     parse_stats_response,
     parse_upsert_response,
     parse_update_response,
@@ -32,6 +32,7 @@
 from pinecone.db_control.models.list_response import ListResponse as SimpleListResponse, Pagination
 from pinecone.core.grpc.protos.db_data_2025_10_pb2 import (
     Vector as GRPCVector,
+    QueryResponse as ProtoQueryResponse,
     UpsertRequest,
     DeleteRequest,
     QueryRequest,
@@ -444,13 +445,13 @@ def _query(
         include_metadata: bool | None = None,
         sparse_vector: (SparseValues | GRPCSparseValues | SparseVectorTypedDict) | None = None,
         **kwargs,
-    ) -> tuple[dict[str, Any], dict[str, str] | None]:
+    ) -> tuple[ProtoQueryResponse, dict[str, str] | None]:
         """
-        Low-level query method that returns raw JSON dict and initial metadata without parsing.
+        Low-level query method that returns protobuf Message and initial metadata without parsing.
         Used internally by query() and query_namespaces() for performance.
 
         Returns:
-            Tuple of (json_dict, initial_metadata). initial_metadata may be None.
+            Tuple of (protobuf_message, initial_metadata). initial_metadata may be None.
         """
         if vector is not None and id is not None:
             raise ValueError("Cannot specify both `id` and `vector`")
@@ -478,7 +479,7 @@ def _query(
 
         timeout = kwargs.pop("timeout", None)
         response, initial_metadata = self._runner.run(self._stub.Query, request, timeout=timeout)
-        return json_format.MessageToDict(response), initial_metadata
+        return response, initial_metadata
 
     def query(
         self,
@@ -569,8 +570,8 @@ def query(
                 future, result_transformer=parse_query_response, timeout=timeout
             )
         else:
-            # For sync requests, use _query to get raw dict and metadata, then parse it
-            json_response, initial_metadata = self._query(
+            # For sync requests, use _query to get protobuf Message and metadata, then parse it
+            response, initial_metadata = self._query(
                 vector=vector,
                 id=id,
                 namespace=namespace,
@@ -583,7 +584,7 @@ def query(
                 **kwargs,
             )
             return parse_query_response(
-                json_response, _check_type=False, initial_metadata=initial_metadata
+                response, _check_type=False, initial_metadata=initial_metadata
             )
 
     def query_namespaces(
@@ -658,8 +659,9 @@ def query_namespaces(
 
         only_futures = cast(Iterable[Future], futures)
         for response in as_completed(only_futures):
-            json_response, _ = response.result()  # Ignore initial_metadata for query_namespaces
-            # Pass raw dict directly to aggregator - no parsing needed
+            proto_response, _ = response.result()  # Ignore initial_metadata for query_namespaces
+            # Convert protobuf Message to dict format for aggregator using optimized helper
+            json_response = query_response_to_dict(proto_response)
             aggregator.add_results(json_response)
 
         final_results = aggregator.get_results()
@@ -853,5 +855,4 @@ def describe_index_stats(
 
         request = DescribeIndexStatsRequest(**args_dict)
         response, _ = self._runner.run(self._stub.DescribeIndexStats, request, timeout=timeout)
-        json_response = json_format.MessageToDict(response)
-        return parse_stats_response(json_response)
+        return parse_stats_response(response)