def partial_update_documents(self, partial_documents: List[Dict], marqo_index) \
            -> MarqoUpdateDocumentsResponse:
        """Partially update documents in the given index by marqo_index object.

        The partial_documents without _id will error out and the error will be returned in the response without
        error out the entire batch.

        If there exists duplicate _id in the partial_documents, the last document will be used.

        If the document does not exist, this document will error out and the error will be returned in the response.

        Args:
            partial_documents: A list of documents to partially update received in the request
            marqo_index: The index object to partially update documents in

        Raises:
            UnsupportedFeatureError: If the index is unstructured

        Return:
            MarqoUpdateDocumentsResponse containing the response of the partial update operation
        """
        if marqo_index.type is IndexType.Unstructured:
            raise UnsupportedFeatureError("Partial document update is not supported for unstructured indexes. "
                                          "Please use add_documents with use_existing_tensor=True instead")
        elif marqo_index.type is IndexType.Structured:
            pass
        elif marqo_index.type is IndexType.SemiStructured:
            if not marqo_index.index_supports_partial_updates: # Partial updates for semi-structured indexes are only supported for Marqo version >= 2.16.0
                raise UnsupportedFeatureError(
                    f"The 'update_documents' endpoint for an unstructured index "
                    f"is only supported for indexes created with Marqo version 2.16.0 or "
                    f"later. This index was created with Marqo {marqo_index.version}."
                )
        else:
            raise ValueError(f"Invalid index type: {marqo_index.type}")

        start_time = timer()
        vespa_index = vespa_index_factory(marqo_index)
        vespa_documents: List[VespaDocument] = []
        unsuccessful_docs: List[Tuple[int, MarqoUpdateDocumentsItem]] = []

        # Remove duplicated documents based on _id
        partial_documents, documents_that_contain_maps = self.process_documents(partial_documents,
                                                                                         unsuccessful_docs, is_index_semi_structured=marqo_index.type is IndexType.SemiStructured)
        documents_that_contain_maps_but_dont_exist_in_vespa = set() # Set to keep track of documents that contain maps but don't exist in Vespa. This will remain unpopulated for structured indexes.
        existing_vespa_documents = {}

        # This block will only execute for SemiStructured Indexes and in case the request has documents that contain maps.
        # 1. Retrieve those documents, which contain maps in the update request, from Vespa.
        # 2. If there's any documents that dont' exist in Vespa, we will append them to unsuccessful_docs
        if marqo_index.type is IndexType.SemiStructured and documents_that_contain_maps: # Only retrieve the document back if the partial update request contains maps and the index is semi-structured
            get_batch_response = self.vespa_client.get_batch(ids = list(documents_that_contain_maps.keys()), fields = [
                VESPA_FIELD_ID, INT_FIELDS, FLOAT_FIELDS, VESPA_DOC_FIELD_TYPES, VESPA_DOC_VERSION_UUID], schema = marqo_index.schema_name)
            responses = get_batch_response.responses
            for resp in responses:
                if resp.document:
                    existing_vespa_documents[resp.document.fields[VESPA_FIELD_ID]] = resp.document.dict()
                else: 
                    id = self.extract_document_id_from_vespa_id(resp) # Extract the document id from the Vespa response. Vespa response will contain the document id even though the document was not found.
                    unsuccessful_docs.append((documents_that_contain_maps.get(id), MarqoUpdateDocumentsItem(id = id,
                                                                                                         status = int(api_exceptions.BadRequestError.status_code),
                                                                                                         error = "Marqo vector store couldn't update the document. Please see: " + update_documents_response() + " for more details")))
                    documents_that_contain_maps_but_dont_exist_in_vespa.add(id)

        for index, doc in enumerate(partial_documents):
            if (marqo_index.type is IndexType.SemiStructured # Only have this check in place for SemiStructured indexes
                    and documents_that_contain_maps_but_dont_exist_in_vespa
                    and doc.get(MARQO_DOC_ID, '') in documents_that_contain_maps_but_dont_exist_in_vespa):
                continue
            try:
                vespa_document = VespaDocument(**vespa_index.to_vespa_partial_document(doc, existing_vespa_documents.get(doc.get(MARQO_DOC_ID, ''), None)))
                vespa_documents.append(vespa_document)
            except ParsingError as e:
                unsuccessful_docs.append(
                    (index, MarqoUpdateDocumentsItem(id=doc.get(MARQO_DOC_ID, ''), error=e.message,
                                                     status=int(api_exceptions.InvalidArgError.status_code))))

        with RequestMetricsStore.for_request().time("partial_update.vespa._bulk"):
            vespa_res: UpdateDocumentsBatchResponse = (
                self.vespa_client.update_documents_batch(vespa_documents,
                                                         marqo_index.schema_name,
                                                         vespa_id_field=vespa_index.get_vespa_id_field()))

        with RequestMetricsStore.for_request().time("partial_update.postprocess"):
            result = self._translate_update_document_response(vespa_res, unsuccessful_docs,
                                                              marqo_index.name, start_time)

        return result
Partially update documents in the given index by marqo_index object.

        The partial_documents without _id will error out and the error will be returned in the response without
        error out the entire batch.

        If there exists duplicate _id in the partial_documents, the last document will be used.

        If the document does not exist, this document will error out and the error will be returned in the response.

        Args:
            partial_documents: A list of documents to partially update received in the request
            marqo_index: The index object to partially update documents in

        Raises:
            UnsupportedFeatureError: If the index is unstructured

        Return:
            MarqoUpdateDocumentsResponse containing the response of the partial update operation
        
src/marqo/core/document/document.py