def _create_embedding_index()

in datastore/providers/analyticdb_datastore.py [0:0]

• 12 lines of code
• 3 McCabe index (conditional complexity)

    def _create_embedding_index(self, cur: psycopg2.extensions.cursor):
        cur.execute(
            f"""
            SELECT * FROM pg_indexes WHERE tablename='{self.collection_name}';
            """
        )
        index_exists = any(
            index[2] == f"{self.collection_name}_embedding_idx"
            for index in cur.fetchall()
        )
        if not index_exists:
            cur.execute(
                f"""
                CREATE INDEX {self.collection_name}_embedding_idx
                ON {self.collection_name}
                USING ann(embedding)
                WITH (
                    distancemeasure=L2,
                    dim=OUTPUT_DIM,
                    pq_segments=64,
                    hnsw_m=100,
                    pq_centers=2048
                );
                """
            )