fix: pinecone index creation (#395)

* fix pinecone opinionated writes, fix potential weaviate name formatting bug, add identity to default index name for opinionated writes, add reference test script for validation, change step name due to potentially confusing log, change method name due to potentially confusing naming

* tidy, cv

* remove kwargs pass

Author: ahmetmeleq

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 0.5.8
+
+### Fixes
+
+* **Fix on pinecone index creation functionality**
+
 ## 0.5.7
 
 ### Fixes

test_e2e/reference_test_scripts/README.md

@@ -0,0 +1 @@
+You can run these to validate your development effort end-to-end. They are not intended to run in CI.

test_e2e/reference_test_scripts/pinecone.sh

@@ -0,0 +1,130 @@
+#!/usr/bin/env bash
+
+set -e
+
+DEST_PATH=$(dirname "$(realpath "$0")")
+SCRIPT_DIR=$(dirname "$DEST_PATH")
+cd "$SCRIPT_DIR"/.. || exit 1
+OUTPUT_FOLDER_NAME=s3-pinecone-dest
+OUTPUT_DIR=$SCRIPT_DIR/structured-output/$OUTPUT_FOLDER_NAME
+WORK_DIR=$SCRIPT_DIR/workdir/$OUTPUT_FOLDER_NAME
+max_processes=${MAX_PROCESSES:=$(python3 -c "import os; print(os.cpu_count())")}
+
+if [ -z "$PINECONE_API_KEY" ]; then
+  echo "Skipping Pinecone ingest test because PINECONE_API_KEY env var is not set."
+  exit 0
+fi
+
+RANDOM_SUFFIX=$((RANDOM % 100000 + 1))
+
+# Set the variables with default values if they're not set in the environment
+PINECONE_INDEX=${PINECONE_INDEX:-"ingest-test-$RANDOM_SUFFIX"}
+PINECONE_HOST_POSTFIX=${PINECONE_HOST_POSTFIX:-"4627-b74a"}
+PINECONE_PROJECT_ID=${PINECONE_PROJECT_ID:-"art8iaj"}
+
+# shellcheck disable=SC1091
+source "$SCRIPT_DIR"/cleanup.sh
+function cleanup {
+
+  # Get response code to check if index exists
+  response_code=$(curl \
+    -s -o /dev/null \
+    -w "%{http_code}" \
+    --request GET \
+    --url "https://api.pinecone.io/indexes/$PINECONE_INDEX" \
+    --header 'accept: application/json' \
+    --header "Api-Key: $PINECONE_API_KEY")
+
+  # Cleanup (delete) index if it exists
+  if [ "$response_code" == "200" ]; then
+    echo ""
+    echo "deleting index $PINECONE_INDEX"
+    curl --request DELETE \
+      "https://api.pinecone.io/indexes/$PINECONE_INDEX" \
+      --header "Api-Key: $PINECONE_API_KEY" \
+      --header 'content-type: application/json'
+
+  else
+    echo "There was an error during index deletion for index $PINECONE_INDEX, with response code: $response_code. It might be that index $PINECONE_INDEX does not exist, so there is nothing to delete."
+  fi
+
+  # Local file cleanup
+  cleanup_dir "$WORK_DIR"
+  cleanup_dir "$OUTPUT_DIR"
+}
+
+trap cleanup EXIT
+
+echo "Creating index $PINECONE_INDEX"
+response_code=$(curl \
+  -s -o /dev/null \
+  -w "%{http_code}" \
+  --request POST \
+  --url "https://api.pinecone.io/indexes" \
+  --header "accept: application/json" \
+  --header "content-type: application/json" \
+  --header "Api-Key: $PINECONE_API_KEY" \
+  --data '
+{
+  "name": "'"$PINECONE_INDEX"'",
+  "dimension": 384,
+  "metric": "cosine",
+  "spec": {
+    "serverless": {
+      "cloud": "aws",
+      "region": "us-east-1"
+    }
+  }
+}
+')
+
+if [ "$response_code" -lt 400 ]; then
+  echo "Index creation success: $response_code"
+else
+  echo "Index creation failure: $response_code"
+  exit 1
+fi
+
+PYTHONPATH=. ./unstructured_ingest/main.py \
+  local \
+  --num-processes "$max_processes" \
+  --output-dir "$OUTPUT_DIR" \
+  --strategy fast \
+  --verbose \
+  --reprocess \
+  --input-path example-docs/book-war-and-peace-1225p.txt \
+  --work-dir "$WORK_DIR" \
+  --chunking-strategy by_title \
+  --chunk-combine-text-under-n-chars 150 --chunk-new-after-n-chars 1500 --chunk-max-characters 2500 --chunk-multipage-sections \
+  --embedding-provider "huggingface" \
+  pinecone \
+  --api-key "$PINECONE_API_KEY" \
+  --index-name "$PINECONE_INDEX" \
+  --batch-size 80
+
+# It can take some time for the index to catch up with the content that was written, this check between 10s sleeps
+# to give it that time process the writes. Will timeout after checking for a minute.
+num_of_vectors_remote=0
+attempt=1
+sleep_amount=8
+while [ "$num_of_vectors_remote" -eq 0 ] && [ "$attempt" -lt 4 ]; do
+  echo "attempt $attempt: sleeping $sleep_amount seconds to let index finish catching up after writes"
+  sleep $sleep_amount
+
+  num_of_vectors_remote=$(curl --request POST \
+    -s \
+    --url "https://$PINECONE_INDEX-$PINECONE_PROJECT_ID.svc.aped-$PINECONE_HOST_POSTFIX.pinecone.io/describe_index_stats" \
+    --header "accept: application/json" \
+    --header "content-type: application/json" \
+    --header "Api-Key: $PINECONE_API_KEY" | jq -r '.totalVectorCount')
+
+  echo "vector count in Pinecone: $num_of_vectors_remote"
+  attempt=$((attempt + 1))
+done
+
+EXPECTED=1825
+
+if [ "$num_of_vectors_remote" -ne $EXPECTED ]; then
+  echo "Number of vectors in Pinecone are $num_of_vectors_remote when the expected number is $EXPECTED. Test failed."
+  exit 1
+fi

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "0.5.7"  # pragma: no cover
+__version__ = "0.5.8"  # pragma: no cover

unstructured_ingest/connector/pinecone.py

@@ -63,14 +63,14 @@ def to_dict(self, **kwargs):
     @property
     def pinecone_index(self):
         if self._index is None:
-            self._index = self.create_index()
+            self._index = self.get_index()
         return self._index
 
     def initialize(self):
         pass
 
     @requires_dependencies(["pinecone"], extras="pinecone")
-    def create_index(self) -> "PineconeIndex":
+    def get_index(self) -> "PineconeIndex":
         from pinecone import Pinecone
         from unstructured import __version__ as unstructured_version
 
@@ -83,6 +83,16 @@ def create_index(self) -> "PineconeIndex":
         logger.debug(f"connected to index: {pc.describe_index(self.connector_config.index_name)}")
         return index
 
+    @requires_dependencies(["pinecone"], extras="pinecone")
+    def create_index(self) -> "PineconeIndex":
+        logger.warning(
+            "create_index (a misleading name as of now) will be deprecated soon. "
+            + "Use get_index instead. This is due to unstructured supporting actual "
+            + "index creation/provisioning now. "
+            + "(Support for v2 connectors only. you are currently using a v1 connector.)"
+        )
+        return self.get_index()
+
     @DestinationConnectionError.wrap
     def check_connection(self):
         _ = self.pinecone_index

unstructured_ingest/v2/interfaces/uploader.py

@@ -38,7 +38,9 @@ def is_batch(self) -> bool:
     def run_batch(self, contents: list[UploadContent], **kwargs: Any) -> None:
         raise NotImplementedError()
 
-    def create_destination(self, destination_name: str = "elements", **kwargs: Any) -> bool:
+    def create_destination(
+        self, destination_name: str = "unstructuredautocreated", **kwargs: Any
+    ) -> bool:
         # Update the uploader config if needed with a new destination that gets created.
         # Return a flag on if anything was created or not.
         return False
@@ -61,6 +63,6 @@ async def run_data_async(self, data: list[dict], file_data: FileData, **kwargs:
 @dataclass
 class VectorDBUploader(Uploader, ABC):
     def create_destination(
-        self, vector_length: int, destination_name: str = "elements", **kwargs: Any
+        self, vector_length: int, destination_name: str = "unstructuredautocreated", **kwargs: Any
     ) -> bool:
         return False

unstructured_ingest/v2/pipeline/steps/index.py

@@ -11,7 +11,7 @@
 
 IndexerT = TypeVar("IndexerT", bound=Indexer)
 
-STEP_ID = "index"
+STEP_ID = "indexer"
 
 
 @dataclass

unstructured_ingest/v2/processes/connectors/pinecone.py

@@ -208,7 +208,7 @@ def format_destination_name(self, destination_name: str) -> str:
     def create_destination(
         self,
         vector_length: int,
-        destination_name: str = "elements",
+        destination_name: str = "unstructuredautocreated",
         destination_type: Literal["pod", "serverless"] = "serverless",
         serverless_cloud: str = "aws",
         serverless_region: str = "us-west-2",
@@ -219,7 +219,7 @@ def create_destination(
     ) -> bool:
         from pinecone import PodSpec, ServerlessSpec
 
-        index_name = destination_name or self.connection_config.index_name
+        index_name = self.connection_config.index_name or destination_name
         index_name = self.format_destination_name(index_name)
         self.connection_config.index_name = index_name
 
@@ -228,13 +228,11 @@ def create_destination(
             logger.info(f"creating pinecone index {index_name}")
 
             pc = self.connection_config.get_client()
-
             if destination_type == "serverless":
                 pc.create_index(
-                    name=destination_name,
+                    name=index_name,
                     dimension=vector_length,
                     spec=ServerlessSpec(cloud=serverless_cloud, region=serverless_region),
-                    **kwargs,
                 )
 
                 return True
@@ -244,7 +242,6 @@ def create_destination(
                     name=destination_name,
                     dimension=vector_length,
                     spec=PodSpec(environment=pod_environment, pod_type=pod_type, pods=pod_count),
-                    **kwargs,
                 )
 
                 return True

unstructured_ingest/v2/processes/connectors/weaviate/weaviate.py

@@ -241,10 +241,13 @@ def format_destination_name(self, destination_name: str) -> str:
         return formatted.capitalize()
 
     def create_destination(
-        self, destination_name: str = "elements", vector_length: Optional[int] = None, **kwargs: Any
+        self,
+        destination_name: str = "unstructuredautocreated",
+        vector_length: Optional[int] = None,
+        **kwargs: Any,
     ) -> bool:
-        destination_name = self.format_destination_name(destination_name)
         collection_name = self.upload_config.collection or destination_name
+        collection_name = self.format_destination_name(collection_name)
         self.upload_config.collection = collection_name
 
         connectors_dir = Path(__file__).parents[1]
@@ -254,9 +257,7 @@ def create_destination(
         collection_config["class"] = collection_name
 
         if not self._collection_exists():
-            logger.info(
-                f"creating default weaviate collection '{collection_name}' with default configs"
-            )
+            logger.info(f"creating weaviate collection '{collection_name}' with default configs")
             with self.connection_config.get_client() as weaviate_client:
                 weaviate_client.collections.create_from_dict(config=collection_config)
                 return True