/*
   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License, version 2.0,
   as published by the Free Software Foundation.

   This program is also distributed with certain software (including
   but not limited to OpenSSL) that is licensed under separate terms,
   as designated in a particular file or component or in included license
   documentation.  The authors of MySQL hereby grant you an additional
   permission to link the program and your derivative works with the
   separately licensed software that they have included with MySQL.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License, version 2.0, for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
*/

#include <algorithm>
#include <iomanip>
#include <list>
#include <memory>
#include <mutex>
#include <queue>
#include <regex>
#include <set>
#include <shared_mutex>
#include <map>
#include <string>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include <vector>

#if defined(__GNUC__) || defined(__clang__)
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wsuggest-override"
#endif

#ifdef _WIN32
#define PLUGIN_EXPORT extern "C" __declspec(dllexport)
#else
#define PLUGIN_EXPORT extern "C"
#endif

#include <ctime>

#ifdef _WIN32
#include <io.h>
#include <string.h>
#define myvector_unlink(path) _unlink(path)
#define myvector_strcasecmp(s1, s2) _stricmp((s1), (s2))
#else
#include <strings.h>
#include <time.h>
#include <unistd.h>
#define myvector_unlink(path) unlink(path)
#define myvector_strcasecmp(s1, s2) strcasecmp((s1), (s2))
#endif

#include "mysql.h"
#include "mysql/plugin.h"
#include "mysql/udf_registration_types.h"
#include "mysql_version.h"
#include "myvector.h"
#include "myvector_errors.h"

#include "myvector_log.h"

// using namespace std; - Removed for code quality
using std::atomic;
using std::current_exception;
using std::endl;
using std::exception_ptr;
using std::getline;
using std::ifstream;
using std::list;
using std::lock_guard;
using std::map;
using std::min;
using std::mutex;
using std::pair;
using std::priority_queue;
using std::regex;
using std::regex_search;
using std::reverse;
using std::set;
using std::shared_lock;
using std::shared_mutex;
using std::string;
using std::stringstream;
using std::thread;
using std::to_string;
using std::unique_lock;
using std::unordered_map;
using std::unordered_set;
using std::vector;

#include <mysql/components/component_implementation.h>
#include <mysql/components/my_service.h>
#include <mysql/components/services/mysql_string.h>
#include <mysql/components/services/udf_metadata.h>

#include "hnswdisk.h"
#include "hnswlib.h"
#include "my_checksum.h"
#include "myvectorutils.h"

#ifndef MYVECTOR_COMPONENT_BUILD
extern REQUIRES_SERVICE_PLACEHOLDER(mysql_udf_metadata);
extern REQUIRES_SERVICE_PLACEHOLDER(mysql_string_converter);
extern REQUIRES_SERVICE_PLACEHOLDER(mysql_string_factory);

#include <mysql/service_plugin_registry.h>

extern my_service<SERVICE_TYPE(mysql_udf_metadata)>* h_udf_metadata_service;
#endif

#ifndef MYVECTOR_COMPONENT_BUILD
#include "my_sys.h"
#include "mysqld_error.h"
#endif

/* Fail the whole statement from a UDF's row function, with a message. Setting
 * *error alone is not enough: MySQL then returns NULL for this row and every
 * later row of the statement, without saying why (#170). The component build
 * uses its own copies of these UDFs (component_src/myvector_udf_service.cc). */
#ifndef MYVECTOR_COMPONENT_BUILD
#define MYVECTOR_UDF_FAIL(fn, ...)                                             \
    do {                                                                       \
        char udf_fail_msg_[256];                                               \
        snprintf(udf_fail_msg_, sizeof(udf_fail_msg_), __VA_ARGS__);           \
        my_error(ER_UDF_ERROR, MYF(0), fn, udf_fail_msg_);                     \
    } while (0)
#else
#define MYVECTOR_UDF_FAIL(fn, ...) \
    do {                           \
    } while (0)
#endif

#define SET_UDF_ERROR_AND_RETURN(...)                                          \
    {                                                                          \
        MYVEC_LOG_ERROR(__VA_ARGS__);                                          \
        *error = 1;                                                            \
        return (result);                                                       \
    }

class AbstractVectorIndex;

/* Each serialized vector has 4 bytes of metadata.
 * Byte 1 - MyVector vector format version
 * Byte 2 - Native datatype of vector elements. e.g FP32, FP16 etc
 * Byte 3 - Unused
 * Byte 4 - Unused
 */

static const unsigned int MYVECTOR_VERSION_V1 = 0x01;

static const unsigned int MYVECTOR_VECTOR_FP32 = 0x01;

static const unsigned int MYVECTOR_VECTOR_FP16 = 0x02;

static const unsigned int MYVECTOR_VECTOR_BV = 0x04;

#define MYVECTOR_V1_FP32_METADATA                                              \
    ((MYVECTOR_VECTOR_FP32 << 8) | MYVECTOR_VERSION_V1)

#define MYVECTOR_V1_BV_METADATA                                                \
    ((MYVECTOR_VECTOR_BV << 8) | MYVECTOR_VERSION_V1)

/* Maximum length of string that can be passed to myvector_construct(). */

static const unsigned int MYVECTOR_CONSTRUCT_MAX_LEN = 128000;

/* Maximum buffer space for return value of myvector_display().
   e.g 3072 dimensions x 23 characters for each dimension = 70kb string
 */
static const unsigned int MYVECTOR_DISPLAY_MAX_LEN = 128000;

/* Default precision for float output in myvector_display() */
static const unsigned int MYVECTOR_DISPLAY_DEF_PREC = 7;

/* Overhead per MYVECTOR column value - 4 bytes for checksum and
   4 bytes for metadata. Overhead disabled if we are using MySQL's VECTOR
 */
#if MYSQL_VERSION_ID >= 90000
static const unsigned int MYVECTOR_COLUMN_EXTRA_LEN = 0;
#else
static const unsigned int MYVECTOR_COLUMN_EXTRA_LEN = 8;
#endif

/* Default number of neighbours to return by myvector_ann_set() */
static const unsigned int MYVECTOR_DEFAULT_ANN_RETURN_COUNT = 10;

/* Filtered search: up to this many allowed rows, HNSW indexes compute exact
 * distances over the allowed rows instead of walking the graph.
 */
static const size_t MYVECTOR_FILTER_EXACT_THRESHOLD = 10000;

/* Max number of neighbours that can be retrieved in single myvector_ann_set()
 */
static const unsigned int MYVECTOR_MAX_ANN_RETURN_COUNT = 10000;

/* Buffer for the myvector_ann_set() result: up to MYVECTOR_MAX_ANN_RETURN_COUNT
 * keys of up to 20 digits, a comma after each, the brackets and a NUL.
 */
static const unsigned int MYVECTOR_ANN_SET_MAX_LEN =
    MYVECTOR_MAX_ANN_RETURN_COUNT * 21 + 3;

/* Basic check for validity of index last update timestamp > '01-01-2024' */
static const unsigned long MYVECTOR_MIN_VALID_UPDATE_TS = 1704047400;

/* Parallel threads are started after 100K vectors + keys have been batch'ed up
 */
static const unsigned int HNSW_PARALLEL_BUILD_UNIT_SIZE = 100000;

extern char* myvector_index_dir;
extern long myvector_feature_level;

char* latin1 = const_cast<char*>("latin1");

const set<string> MYVECTOR_INDEX_TYPES{"KNN", "HNSW", "HNSW_BV"};

/* Thread-local storage for distances from last ANN search.
 * Using a static thread_local object avoids manual memory management
 * and prevents memory leaks. The map is automatically constructed
 * on first use per thread and destroyed when the thread exits.
 */
static thread_local unordered_map<KeyTypeInteger, double> tls_distances_map;
static thread_local unordered_map<KeyTypeInteger, double>* tls_distances =
    &tls_distances_map;

/* isValidIndexType - KNN, HNSW or HNSW_BV, in any case */
inline bool isValidIndexType(string indextype) {
    transform(indextype.begin(), indextype.end(), indextype.begin(),
              [](unsigned char c) { return static_cast<char>(toupper(c)); });
    return (MYVECTOR_INDEX_TYPES.find(indextype) != MYVECTOR_INDEX_TYPES.end());
}

inline int MyVectorStorageLength(int dim) {
    return (dim * sizeof(FP32)) + MYVECTOR_COLUMN_EXTRA_LEN;
}

inline int MyVectorDimFromStorageLength(int length) {
    return (length - MYVECTOR_COLUMN_EXTRA_LEN) / sizeof(FP32);
}

inline int MyVectorBVStorageLength(int dim) {
    return (dim / BITS_PER_BYTE) + MYVECTOR_COLUMN_EXTRA_LEN;
}

inline int MyVectorBVDimFromStorageLength(int length) {
    return (length - MYVECTOR_COLUMN_EXTRA_LEN) * BITS_PER_BYTE;
}

/* Compute L2/Eucliean squared distance via optimized function from hnswlib */
double computeL2Distance(const FP32* __restrict v1,
                         const FP32* __restrict v2,
                         int dim) {
    hnswlib::L2Space sp(dim);
    hnswlib::DISTFUNC<FP32> distfn;
    double dist = 0.0;
    size_t sdim = dim;

    if (v1 && v2 && dim) {
        distfn = sp.get_dist_func();
        dist = distfn(v1, v2, &sdim);
    }

    return dist;
}

/* Compute InnerProduct distance via optimized function from hnswlib */
double computeIPDistance(const FP32* __restrict v1,
                         const FP32* __restrict v2,
                         int dim) {
    hnswlib::InnerProductSpace sp(dim);
    hnswlib::DISTFUNC<FP32> distfn;
    double dist = 0.0;
    size_t sdim = dim;

    if (v1 && v2 && dim) {
        distfn = sp.get_dist_func();
        dist = distfn(v1, v2, &sdim);
    }

    return dist;
}

/* Compute Cosine distance using standard formula */
double computeCosineDistance(const FP32* __restrict v1,
                             const FP32* __restrict v2,
                             int dim) {
    double dist = 0.0, v1v2 = 0.0, normv1 = 0.0, normv2 = 0.0;

    for (int i = 0; i < dim; i++) {
        v1v2 += (v1[i] * v2[i]);
        normv1 += (v1[i] * v1[i]);
        normv2 += (v2[i] * v2[i]);
    }

    double t = (sqrt(normv1 * normv2));
    if (t)
        dist = (double)v1v2 / t;

    return (1 - dist);
}

MyVectorDistanceFn myvector_distance_fn(const char* name, size_t len) {
    std::string m(name ? name : "", name ? len : 0);
    std::transform(m.begin(), m.end(), m.begin(),
                   [](unsigned char c) { return static_cast<char>(std::tolower(c)); });
    if (m == "l2" || m == "euclidean")
        return computeL2Distance;
    if (m == "cosine")
        return computeCosineDistance;
    if (m == "ip")
        return computeIPDistance;
    return nullptr;
}

float computeCosineDistanceFn(const void* __restrict v1,
                              const void* __restrict v2,
                              const void* __restrict qty_ptr) {
    size_t qty = *((const size_t*)qty_ptr);  /// dimension of the vector
    return (float)computeCosineDistance(
        (const FP32*)v1, (const FP32*)v2, (int)qty);
}

class AngularDistanceSpace : public hnswlib::SpaceInterface<float> {
    hnswlib::DISTFUNC<float> fstdistfunc_;
    size_t data_size_;
    size_t dim_;

public:
    AngularDistanceSpace(size_t dim) {
        fstdistfunc_ = computeCosineDistanceFn;
        dim_ = dim;
        data_size_ = dim * sizeof(float);
    }

    size_t get_data_size() { return data_size_; }

    hnswlib::DISTFUNC<float> get_dist_func() { return fstdistfunc_; }

    void* get_dist_func_param() { return &dim_; }

    ~AngularDistanceSpace() {}
};

/* HammingDistanceFn - Calculate Hamming distance between 2 bit vectors.
 * Prototype is in HNSWLIB style */
float HammingDistanceFn(const void* __restrict pVect1,
                        const void* __restrict pVect2,
                        const void* __restrict qty_ptr) {
    size_t qty = *((const size_t*)qty_ptr);  /// dimension of the vector
    float dist = 0;
    unsigned long ldist = 0;
    const unsigned long* a = (const unsigned long*)pVect1;
    const unsigned long* b = (const unsigned long*)pVect2;

    /* Hamming Distance between 2 byte sequences - Number of bit positions
     * matching/different in both the sequences. In the plugin, we calculate
     * the diff'ing bit positions and that is the distance. Thus smaller
     * distance implies the vectors are 'nearer'/'similar' to each other.
     */
    size_t iter = (qty / (sizeof(unsigned long) * BITS_PER_BYTE));
    for (size_t i = 0; i < iter; i++) {
        unsigned long res = (*a ^ *b);
        a++;
        b++;

        ldist += __builtin_popcountll(res);
    }
    dist = ldist;
    return dist;
}

/* HammingBinaryVectorSpace : Implement hnswlib's SpaceInterface for
 * Hamming Distance based HNSW index for Binary Vectors.
 */
class HammingBinaryVectorSpace : public hnswlib::SpaceInterface<float> {
    hnswlib::DISTFUNC<float> fstdistfunc_;
    size_t data_size_;
    size_t dim_;

public:
    HammingBinaryVectorSpace(size_t dim) {
        fstdistfunc_ = HammingDistanceFn;
        dim_ = dim;
        data_size_ = (dim / BITS_PER_BYTE);  /// 1 bit per dimension - assumes
                                             /// dim is multiple of 8
    }

    size_t get_data_size() { return data_size_; }

    hnswlib::DISTFUNC<float> get_dist_func() { return fstdistfunc_; }

    void* get_dist_func_param() { return &dim_; }

    ~HammingBinaryVectorSpace() {}
};

/* KNNIndex - A vector index type that implements brute-force KNN search in
 * the MyVector plugin. This index type could possibly be faster than SQL
 * performing ORDER BY myvector_distance(...) [as long as all vectors fit
 * in memory].
 */
class KNNIndex : public AbstractVectorIndex {
public:
    KNNIndex(const string& name, const string& options);

    ~KNNIndex() {}

    /* Next 4 methods are no-op in the KNN in-memory index */
    bool saveIndex(const string& path, const string& option = "");

    bool saveIndexIncr(const string& path, const string& option = "");

    bool loadIndex(const string& path);

    bool dropIndex(const string& path);

    bool initIndex();

    bool closeIndex();

    string getName() { return m_name; }
    string getType() { return "KNN"; }

    bool isCosineMetric() const override {
        return (m_distfn == computeCosineDistance);
    }

    string getStatus();

    bool searchVectorNN(VectorPtr qvec,
                        int dim,
                        vector<KeyTypeInteger>& keys,
                        int n,
                        const unordered_set<KeyTypeInteger>* allowed = nullptr,
                        int ef_search = 0);
    bool insertVector(VectorPtr vec, int dim, KeyTypeInteger id);
    bool deleteVector(KeyTypeInteger id) override;

    bool supportsIncrUpdates() { return true; }

    bool supportsPersist() { return false; }

    bool supportsConcurrentUpdates() { return false; }  /// no mutexing!

    bool supportsIncrRefresh() { return true; }

    bool isReady() { return true; }

    bool isDirty() { return false; }  /// No persistence

    int getDimension() { return m_dim; }

    bool startParallelBuild(int nthreads) { return false; }

    void setUpdateTs(unsigned long ts) { m_updateTs = ts; }

    unsigned long getUpdateTs() { return m_updateTs; }

    unsigned long getRowCount() { return m_n_rows; }

    void getLastUpdateCoordinates(string& binlogFile, size_t& binlogPos) {
        std::lock_guard<std::mutex> l(m_coordMutex);
        binlogFile = m_binlogFile;
        binlogPos  = m_binlogPosition;
    }

    void setLastUpdateCoordinates(const string& binlogFile,
                                  const size_t& binlogPos) {
        std::lock_guard<std::mutex> l(m_coordMutex);
        m_binlogFile     = binlogFile;
        m_binlogPosition = binlogPos;
    }

private:
    string m_name;
    string m_options;
    int m_dim;
    unsigned long m_updateTs;
    MyVectorOptions m_optionsMap;

    /* Resolved (matched, canonical-case) distance metric name -- what the
     * index actually uses, as opposed to m_optionsMap.getOption("dist")
     * which is the raw, as-typed option value. getStatus() must report
     * this one; see issue #119 ("MYVECTOR_INDEX_STATUS prints the raw
     * option ... which hides the mismatch"). */
    string m_dist;

    /* Last binlog position applied. Until the index is built or loaded, it is
     * the "never" sentinel, so the binlog workers skip events for it: the index
     * is in the collection from open() on, before initIndex() runs (#187). */
    std::mutex m_coordMutex;
    string m_binlogFile = "zzzzzz.bin";
    size_t m_binlogPosition{99999999999};

    mutable std::shared_mutex search_insert_mutex_;

    atomic<unsigned long> m_n_rows{0};
    atomic<unsigned long> m_n_searches{0};

    /* The in-memory data store for the vectors */
    vector<pair<vector<FP32>, KeyTypeInteger>> m_vectors;

    double (*m_distfn)(const FP32* v1, const FP32* v2, int dim);
};

KNNIndex::KNNIndex(const string& name, const string& options)
    : m_name(name), m_options(options), m_optionsMap(options), m_updateTs(0) {
    m_dim = m_optionsMap.getIntOption("dim", 0);

    m_distfn = computeL2Distance;
    m_dist = "L2";
    string distOpt = m_optionsMap.getOption("dist");
    if (distOpt.size()) {
        if (!myvector_strcasecmp(distOpt.c_str(), "Cosine")) {
            m_distfn = computeCosineDistance;
            m_dist = "Cosine";
        } else if (!myvector_strcasecmp(distOpt.c_str(), "IP")) {
            m_distfn = computeIPDistance;
            m_dist = "IP";
        } else if (myvector_strcasecmp(distOpt.c_str(), "L2"))
            MYVEC_LOG_ERROR(
                "MyVector unknown dist option '%s' for index %s, using L2",
                distOpt.c_str(), name.c_str());
    } else {
        m_optionsMap.setOption("dist", "L2");
    }
    setUpdateTs(0);
}

/* Brute-force, exact search KNN implemented using in-memory vector<> and
 * priority queue. Potentially faster than SELECT ... ORDER BY
 * myvector_distance()
 */
bool KNNIndex::searchVectorNN(VectorPtr qvec,
                              int dim,
                              vector<KeyTypeInteger>& keys,
                              int n,
                              const unordered_set<KeyTypeInteger>* allowed,
                              int /* ef_search: exact search, not used */) {
    std::shared_lock lock(search_insert_mutex_);

    priority_queue<pair<FP32, KeyTypeInteger>> pq;
    keys.clear();

    /* Use priority queue to find out 'n' neighbours with least distance */
    for (auto& row : m_vectors) {
        if (allowed && !allowed->count(row.second))
            continue;
        vector<FP32>& a = row.first;
        double dist = m_distfn((FP32*)qvec, a.data(), m_dim);

        if (pq.size() < n)
            pq.push({dist, row.second});
        else {
            auto top = pq.top();
            if (dist < top.first) {
                pq.pop();
                pq.push({dist, row.second});
            }
        }
    } /* for */

    while (pq.size()) {
        auto r = pq.top();
        pq.pop();
        keys.push_back(r.second);
        (*tls_distances)[r.second] = r.first;  /// pkid -> distance
    }

    reverse(keys.begin(), keys.end());  /// nearest to farthest

    m_n_searches++;
    return true;
}

/* insertVector - just stash the vector into in-memory vector<> collection */
bool KNNIndex::insertVector(VectorPtr vec, int dim, KeyTypeInteger id) {
    std::unique_lock lock(search_insert_mutex_);

    FP32* fvec = static_cast<FP32*>(vec);
    vector<FP32> row(fvec, fvec + m_dim);
    m_vectors.push_back({row, id});  /// simple index - multithread safe

    m_n_rows++;

    return true;
}

/* deleteVector - drop every entry with this key (online DELETE/UPDATE) */
bool KNNIndex::deleteVector(KeyTypeInteger id) {
    std::unique_lock lock(search_insert_mutex_);
    size_t before = m_vectors.size();
    m_vectors.erase(std::remove_if(m_vectors.begin(), m_vectors.end(),
                                   [id](const pair<vector<FP32>, KeyTypeInteger>& e) {
                                       return e.second == id;
                                   }),
                    m_vectors.end());
    size_t removed = before - m_vectors.size();
    m_n_rows -= removed;
    return removed > 0;
}

bool KNNIndex::saveIndex(const string&, const string&) {
    MYVEC_LOG_WARN("KNN Memory Index (%s) - Save Index to disk is no-op",
                   m_name.c_str());

    return true;
}

bool KNNIndex::saveIndexIncr(const string&, const string&) {
    MYVEC_LOG_WARN("KNN Memory Index (%s) - Save Index Incr to disk is no-op",
                   m_name.c_str());

    return true;
}

bool KNNIndex::dropIndex(const string&) {
    MYVEC_LOG_WARN("KNN Memory Index (%s) - Drop Index is no-op",
                   m_name.c_str());

    return true;
}

bool KNNIndex::loadIndex(const string&) {
    MYVEC_LOG_WARN("KNN Memory Index (%s) - Load Index is no op",
                   m_name.c_str());
    return true;
}

bool KNNIndex::initIndex() {
    MYVEC_LOG_DEBUG("KNN Memory Index (%s) - initIndex()", m_name.c_str());

    m_vectors.clear();
    m_n_rows = 0;
    m_n_searches = 0;
    // Sentinel: any real binlog file name is lexicographically less than
    // "zzzzzz.bin", so isAfter() returns false until setLastUpdateCoordinates
    // is called with the BUILD position.
    setLastUpdateCoordinates("zzzzzz.bin", 99999999999);

    return true;
}

bool KNNIndex::closeIndex() { return true; }

string KNNIndex::getStatus() {
    std::stringstream ss;

    ss << endl;
    ss << "Vector Index : " << m_name << endl;
    ss << "Type : KNN" << endl;
    ss << "Dimension : " << m_dim << endl;
    ss << "Distance : " << m_dist << endl;
    ss << "Rows Inserted : " << m_n_rows << endl;
    ss << "Searches : " << m_n_searches << endl;

    return ss.str();
}

class HNSWMemoryIndex : public AbstractVectorIndex {
public:
    HNSWMemoryIndex(const string& name, const string& options);

    ~HNSWMemoryIndex();

    bool supportsIncrUpdates() { return m_incrUpdates; }
    bool supportsIncrRefresh() { return m_incrRefresh; }
    bool isDirty() { return m_isDirty; }

    bool saveIndex(const string& path, const string& option);

    bool saveIndexIncr(const string& path, const string& option);

    bool loadIndex(const string& path);

    bool dropIndex(const string& path);

    bool initIndex();

    bool closeIndex();

    string getName() { return m_name; }

    string getType() { return m_type; }

    bool isCosineMetric() const override {
        return (m_dist == "Cosine" || m_dist == "CosineNorm" || m_dist == "Angular");
    }

    string getStatus();

    bool searchVectorNN(VectorPtr qvec,
                        int dim,
                        vector<KeyTypeInteger>& keys,
                        int n,
                        const unordered_set<KeyTypeInteger>* allowed = nullptr,
                        int ef_search = 0);

    bool insertVector(VectorPtr vec, int dim, KeyTypeInteger id);
    bool deleteVector(KeyTypeInteger id) override;

    int getDimension() { return m_dim; }

    void setUpdateTs(unsigned long ts) { m_updateTs = ts; }

    unsigned long getUpdateTs() { return m_updateTs; }

    unsigned long getRowCount() { return m_n_rows; }

    bool startParallelBuild(int nthreads);

    virtual hnswlib::SpaceInterface<float>* getSpace(size_t dim);

    int getEfConstruction() { return m_ef_construction; }
    int getM() { return m_M; }
    int getSize() { return m_size; }

    void getLastUpdateCoordinates(string& binlogFile, size_t& binlogPos);
    void setLastUpdateCoordinates(const string& binlogFile,
                                  const size_t& binlogPos);
    void getCheckPointString(string& ckstr);

private:
    string m_name;
    string m_type;
    string m_options;
    MyVectorOptions m_optionsMap;
    unsigned long m_updateTs;

    int m_dim;
    int m_ef_construction;
    int m_ef_search;
    int m_M;
    int m_size;

    string m_dist;

    hnswlib::AlgorithmInterface<FP32>* m_alg_hnsw = nullptr;
    hnswlib::SpaceInterface<float>* m_space = nullptr;

    atomic<unsigned long> m_n_rows{0};
    atomic<unsigned long> m_n_searches{0};

    bool m_isDirty;

    bool m_incrUpdates;
    bool m_incrRefresh;

    /* Temporary store for multi-threaded, parallel index build */
    vector<char> m_batch;
    vector<KeyTypeInteger> m_batchkeys;
    bool m_isParallelBuild{false};

    bool flushBatchParallel();
    bool flushBatchSerial();

    int m_threads;

    /// last update coordinates. Until the index is built or loaded they are the
    /// "never" sentinel, so the binlog workers skip it: open() puts the index in
    /// the collection before initIndex() creates m_alg_hnsw (#187).
    std::mutex m_coordMutex;
    string m_binlogFile = "zzzzzz.bin";
    size_t m_binlogPosition = 99999999999;
};

HNSWMemoryIndex::HNSWMemoryIndex(const string& name, const string& options)
    : m_name(name), m_options(options), m_optionsMap(options), m_updateTs(0) {
    bool valid = true;
    m_dim = m_optionsMap.getIntOption("dim", 0, &valid);
    m_size = m_optionsMap.getIntOption("size", 0, &valid);
    m_ef_construction = m_optionsMap.getIntOption("ef", 100, &valid);
    m_ef_search = m_ef_construction;
    m_M = m_optionsMap.getIntOption("M", 16, &valid);
    m_type = m_optionsMap.getOption("type");  // Supports HNSW and HNSW_BV
    /* getSpace() matches the type exactly. VectorIndexCollection::open() accepts it
     * in any case, so a lower-case "hnsw" used to reach here unmatched, getSpace()
     * returned nullptr and the first use of the null space crashed mysqld. */
    transform(m_type.begin(), m_type.end(), m_type.begin(), [](unsigned char c) {
        return static_cast<char>(toupper(c));
    });
    m_incrUpdates = m_optionsMap.getOption("online") == "Y";
    m_incrRefresh = m_optionsMap.getOption("track").length() > 0;

    m_dist = "L2";

    {
        string distOpt = m_optionsMap.getOption("dist");
        if (!myvector_strcasecmp(distOpt.c_str(), "Cosine"))
            m_dist = "Cosine";
        else if (!myvector_strcasecmp(distOpt.c_str(), "CosineNorm"))
            m_dist = "CosineNorm";
        else if (!myvector_strcasecmp(distOpt.c_str(), "Angular"))
            m_dist = "Angular";
        else if (distOpt.size() && myvector_strcasecmp(distOpt.c_str(), "L2"))
            MYVEC_LOG_ERROR(
                "MyVector unknown dist option '%s' for index %s, using L2",
                distOpt.c_str(), name.c_str());
    }

    if (m_optionsMap.getOption("ef_search").length())
        m_ef_search = m_optionsMap.getIntOption("ef_search", m_ef_construction);

    MYVEC_LOG_DEBUG("hnsw index params %s %s  %d %d %d %d %d",
                name.c_str(),
                m_type.c_str(),
                m_dim,
                m_size,
                m_ef_construction,
                m_ef_search,
                m_M);
}

HNSWMemoryIndex::~HNSWMemoryIndex() {
    if (m_alg_hnsw)
        delete m_alg_hnsw;
    if (m_space)
        delete m_space;
}

bool HNSWMemoryIndex::initIndex() {
    MYVEC_LOG_DEBUG("hnsw initIndexO %p %s %d %d %d %d %d",
                this,
                m_name.c_str(),
                m_dim,
                m_size,
                m_ef_construction,
                m_ef_search,
                m_M);
    m_space = getSpace(m_dim);

    m_alg_hnsw = new hnswlib::HierarchicalDiskNSW<FP32>(
        m_space, m_size, m_M, m_ef_construction);

    (dynamic_cast<hnswlib::HierarchicalDiskNSW<FP32>*>(m_alg_hnsw))
        ->setEf(m_ef_search);

    m_n_rows = 0;
    m_n_searches = 0;

    setLastUpdateCoordinates("zzzzzz.bin", 99999999999);
    setUpdateTs(0);

    return true;
}

void HNSWMemoryIndex::getCheckPointString(string& ckstr) {
    stringstream ss;

    if (supportsIncrUpdates()) {
        string binlogFile;
        size_t binlogPos = 0;

        getLastUpdateCoordinates(binlogFile, binlogPos);
        ss << "Checkpoint:binlog:" << binlogFile << ":" << binlogPos;
    } else {
        ss << "Checkpoint:timestamp:" << getUpdateTs();
    }
    ckstr = ss.str();
}

bool HNSWMemoryIndex::saveIndex(const string& path, const string& option) {
#ifdef TODO
    if (!m_isDirty) {
        MYVEC_LOG_WARN("HNSW index %s is not updated, save not required",
                       m_name.c_str());
        return true;
    }
#endif

    // lockExclusive();

    if (m_isParallelBuild) {
        flushBatchSerial();  // last batch, maybe small
        // The pending batch is now in the in-memory graph; this is a distinct,
        // already-completed step from the on-disk write below, so it must not be
        // repeated on a later call just because that write happens to fail.
        m_isParallelBuild = false;
    }

    MYVEC_LOG_DEBUG(
        "HNSWemoryIndex::saveIndex %s %s.", path.c_str(), option.c_str());

    string filename = MyVectorIndexBase(path, m_name) + ".hnsw.index";

    string checkPointStr;
    getCheckPointString(checkPointStr);

    if (!m_alg_hnsw) {
        MYVEC_LOG_ERROR("HNSWMemoryIndex::saveIndex (%s) : null HNSW object.",
                    m_name.c_str());
        return false;
    }

    hnswlib::HierarchicalDiskNSW<FP32>* alg_hnsw =
        dynamic_cast<hnswlib::HierarchicalDiskNSW<FP32>*>(m_alg_hnsw);
    alg_hnsw->setCheckPointId(checkPointStr);

    /* hnswlib reports I/O errors (unwritable directory, disk full, ...) by throwing
     * std::runtime_error. Letting one escape into mysqld aborts the whole server
     * (std::terminate), so turn it into a failed save that the caller can report. */
    try {
        if (option == "build") {
            // hnswlib method for full write/rewrite. Expect 10GB to take 10 secs.
            alg_hnsw->saveIndex(filename);
        } else {
            // "refresh" or "checkpoint" - special MyVector incremental persistence.
            alg_hnsw->doCheckPoint(filename);
        }
    } catch (const std::exception& e) {
        MYVEC_LOG_ERROR("HNSWMemoryIndex::saveIndex (%s) failed writing %s: %s",
                        m_name.c_str(),
                        filename.c_str(),
                        e.what());
        return false;
    }

    m_isDirty = false;

    return true;
}

bool HNSWMemoryIndex::saveIndexIncr(const string& path, const string& option) {
    return true;
}

bool HNSWMemoryIndex::loadIndex(const string& path) {
    if (m_alg_hnsw)
        delete m_alg_hnsw;
    if (m_space)
        delete m_space;

    m_alg_hnsw = nullptr;
    m_space = nullptr;

    m_space = getSpace(m_dim);

    string indexfile = MyVectorIndexBase(path, m_name) + ".hnsw.index";

    MYVEC_LOG_DEBUG(
        "Loading HNSW index %s from %s", m_name.c_str(), indexfile.c_str());

    /* hnswlib throws std::runtime_error for errors */
    m_alg_hnsw = nullptr;
    try {
        m_alg_hnsw = new hnswlib::HierarchicalDiskNSW<FP32>(m_space, indexfile);
    } catch (std::runtime_error& e) {
        MYVEC_LOG_WARN("Error loading hnsw index (%s) from file : %s",
                      m_name.c_str(),
                      e.what());
    }

    if (!m_alg_hnsw) { /* no disk files found */
        initIndex();
    } else {
        string binlogFile;
        size_t binlogPosition = 0;
        size_t ts = 0;

        string ckid =
            dynamic_cast<hnswlib::HierarchicalDiskNSW<FP32>*>(m_alg_hnsw)
                ->getCheckPointId();

        if (ckid.find("Checkpoint:timestamp") != string::npos) {
            ts = atol(ckid.substr(ckid.rfind(":") + 1).c_str());
            MYVEC_LOG_DEBUG("load index checkpoint ts = %lu.", ts);
            setUpdateTs(ts);
        } else if (ckid.find("Checkpoint:binlog") != string::npos) {
            // ckptid=Checkpoint:binlog:binlog.000516:6761
            size_t p1 = ckid.rfind(":");
            binlogPosition = atol(ckid.substr(p1 + 1).c_str());
            size_t p2 = ckid.rfind(":", p1 - 1);
            binlogFile = ckid.substr(p2 + 1, (p1 - (p2 + 1)));
            setLastUpdateCoordinates(binlogFile, binlogPosition);
        }
    }

    MYVEC_LOG_DEBUG(
        "debug HNSW index %s from %s", m_name.c_str(), indexfile.c_str());
    dynamic_cast<hnswlib::HierarchicalDiskNSW<FP32>*>(m_alg_hnsw)->debug();
    return true;
}

bool HNSWMemoryIndex::dropIndex(const string& path) {
    /* Force drop index - delete files and free memory */
    const string base = MyVectorIndexBase(path, m_name);
    string indexfile = base + ".hnsw.index";
    myvector_unlink(indexfile.c_str());
    string linksfile = base + ".hnsw.index.links";
    myvector_unlink(linksfile.c_str());
    string linksdatafile = base + ".hnsw.index.links.data";
    myvector_unlink(linksdatafile.c_str());
    string statusfile = base + ".hnsw.index.status";
    myvector_unlink(statusfile.c_str());

    if (m_alg_hnsw)
        delete m_alg_hnsw;
    m_alg_hnsw = nullptr;

    if (m_space)
        delete m_space;
    m_space = nullptr;

    return true;
}

bool HNSWMemoryIndex::closeIndex() { return true; }

hnswlib::SpaceInterface<float>* HNSWMemoryIndex::getSpace(size_t dim) {
    if (m_type == "HNSW" && m_dist == "L2")
        return new hnswlib::L2Space(m_dim);
    else if (m_type == "HNSW" && m_dist == "CosineNorm")
        return new hnswlib::InnerProductSpace(m_dim);
    else if (m_type == "HNSW" && m_dist == "Cosine")
        return new AngularDistanceSpace(m_dim);
    else if (m_type == "HNSW" && m_dist == "Angular")
        return new AngularDistanceSpace(m_dim);
    else if (m_type == "HNSW_BV")
        return new HammingBinaryVectorSpace(m_dim);

    return nullptr;
}

string HNSWMemoryIndex::getStatus() {
    std::stringstream ss;

    ss << endl;
    ss << "Vector Index : " << m_name << endl;
    ss << "Type : " << m_type << endl;
    ss << "Dimension : " << m_dim << endl;
    /* HNSW_BV always searches with Hamming distance (see getSpace()),
     * independent of m_dist/the "dist" option -- report that, not the
     * generic (and here misleading, since it's never actually used)
     * m_dist default of "L2". */
    ss << "Distance : " << (m_type == "HNSW_BV" ? "Hamming" : m_dist) << endl;
    ss << "Max. Capacity : " << m_size << endl;
    ss << "M = " << m_M << endl;

    if (m_alg_hnsw) {
        ss << "Element Data Size : "
           << (dynamic_cast<hnswlib::HierarchicalDiskNSW<FP32>*>(m_alg_hnsw))
                  ->size_data_per_element_
           << endl;
        auto* disk = dynamic_cast<hnswlib::HierarchicalDiskNSW<FP32>*>(m_alg_hnsw);
        ss << "Current Rows : "
           << (disk->cur_element_count - disk->getDeletedCount())
           << endl;
        ss << "Searches : " << m_n_searches << endl;
    }

    return ss.str();
}

/* KeySetFilter - hnswlib filter that accepts only labels in a key set */
class KeySetFilter : public hnswlib::BaseFilterFunctor {
public:
    explicit KeySetFilter(const unordered_set<KeyTypeInteger>& keys)
        : m_keys(keys) {}
    bool operator()(hnswlib::labeltype id) override {
        return m_keys.count(id) != 0;
    }

private:
    const unordered_set<KeyTypeInteger>& m_keys;
};

bool HNSWMemoryIndex::searchVectorNN(VectorPtr qvec,
                                     int dim,
                                     vector<KeyTypeInteger>& keys,
                                     int n,
                                     const unordered_set<KeyTypeInteger>* allowed,
                                     int ef_search) {
    priority_queue<pair<FP32, hnswlib::labeltype>> result;

    auto* disk_hnsw =
        dynamic_cast<hnswlib::HierarchicalDiskNSW<FP32>*>(m_alg_hnsw);
    /* HNSW_BV stores dim/8 bytes per vector, so it cannot be read back as
     * FP32 values: it always takes the graph path below.
     */
    if (allowed && disk_hnsw && m_type == "HNSW" &&
        allowed->size() <= MYVECTOR_FILTER_EXACT_THRESHOLD) {
        /* Few allowed rows: an exact scan over just those rows is cheap, and
         * a graph walk that must skip most nodes loses recall.
         */
        hnswlib::DISTFUNC<float> distfn = m_space->get_dist_func();
        void* distparam = m_space->get_dist_func_param();
        for (KeyTypeInteger key : *allowed) {
            vector<FP32> v;
            try {
                v = disk_hnsw->getDataByLabel<FP32>(key);
            } catch (const std::runtime_error&) {
                continue;  /// key not in the index (or deleted)
            }
            FP32 dist = distfn(qvec, v.data(), distparam);
            if (result.size() < (size_t)n)
                result.push({dist, key});
            else if (dist < result.top().first) {
                result.pop();
                result.push({dist, key});
            }
        }
    } else {
        /* ef_search applies to this call only; the index's ef_ is not changed,
         * because other queries read it concurrently (#165). */
        static const unordered_set<KeyTypeInteger> kNoKeys;  // unused if !allowed
        KeySetFilter filter(allowed ? *allowed : kNoKeys);
        hnswlib::BaseFilterFunctor* isIdAllowed = allowed ? &filter : nullptr;
        if (ef_search > 0 && disk_hnsw)
            result = disk_hnsw->searchKnnEf(qvec, n, (size_t)ef_search, isIdAllowed);
        else
            result = m_alg_hnsw->searchKnn(qvec, n, isIdAllowed);
    }

    keys.clear();
    tls_distances->clear();
    while (!result.empty()) {
        keys.push_back(result.top().second);
        (*tls_distances)[result.top().second] =
            result.top().first;  /// pkid -> distance
        result.pop();
    }

    reverse(keys.begin(), keys.end());  // nearest to farthest
    m_n_searches++;
    return true;
}

void HNSWMemoryIndex::getLastUpdateCoordinates(string& binlogFile,
                                               size_t& binlogPosition) {
    std::lock_guard<std::mutex> l(m_coordMutex);
    binlogFile = m_binlogFile;
    binlogPosition = m_binlogPosition;
}

void HNSWMemoryIndex::setLastUpdateCoordinates(const string& binlogFile,
                                               const size_t& binlogPosition) {
    std::lock_guard<std::mutex> l(m_coordMutex);
    m_binlogFile = binlogFile;
    m_binlogPosition = binlogPosition;
    MYVEC_LOG_DEBUG(
        "setLastUpdateCoordinates %s %lu", binlogFile.c_str(), binlogPosition);
}

// Parallel HNSW index load (hnswlib:example_search_mt.cpp)
template <class Function>
inline void ParallelFor(size_t start,
                        size_t end,
                        size_t numThreads,
                        Function fn) {
    MYVEC_LOG_DEBUG("Entered ParallelFor %lu %lu t=%lu", start, end, numThreads);
    if (numThreads == 0) {
        numThreads = thread::hardware_concurrency();
    }

    if (numThreads == 1) {
        for (size_t id = start; id < end; id++) {
            fn(id, 0);
        }
    } else {
        vector<thread> threads;
        atomic<size_t> current(start);

        // keep track of exceptions in threads
        // https://stackoverflow.com/a/32428427/1713196
        exception_ptr lastException = nullptr;
        mutex lastExceptMutex;

        for (size_t threadId = 0; threadId < numThreads; ++threadId) {
            threads.push_back(thread([&, threadId] {
                while (true) {
                    size_t id = current.fetch_add(1);

                    if (id >= end) {
                        break;
                    }

                    try {
                        fn(id, threadId);
                    } catch (...) {
                        unique_lock<mutex> lastExcepLock(lastExceptMutex);
                        lastException = current_exception();
                        /*
                         * This will work even when current is the largest value
                         * that size_t can fit, because fetch_add returns the
                         * previous value before the increment (what will result
                         * in overflow and produce 0 instead of current + 1).
                         */
                        current = end;
                        break;
                    }
                }
            }));
        }
        for (auto& thread : threads) {
            thread.join();
        }
        if (lastException) {
            rethrow_exception(lastException);
        }
    }
}

bool HNSWMemoryIndex::startParallelBuild(int nthreads) {
    m_batch.clear();
    m_batchkeys.clear();
    m_isParallelBuild = true;
    m_threads = nthreads;
    return true;
}

bool HNSWMemoryIndex::flushBatchSerial() {
    MYVEC_LOG_DEBUG("flushBatchSerial %lu", m_batchkeys.size());
    for (unsigned int i = 0; i < m_batchkeys.size(); i++) {
        m_alg_hnsw->addPoint((void*)&(m_batch[i * m_space->get_data_size()]),
                             m_batchkeys[i]);
    }
    // Mirror flushBatchParallel(): once the pending rows are in the in-memory graph,
    // this batch is consumed regardless of whether the caller's later on-disk save
    // succeeds. Leaving it unset caused saveIndex() to re-add the same stale rows on
    // every subsequent save after a failed one (they were never cleared here).
    m_batch.clear();
    m_batchkeys.clear();
    return true;
}

bool HNSWMemoryIndex::flushBatchParallel() {
    MYVEC_LOG_DEBUG("Entered flushBatchParallel for (%s), nthreads = %d, sz = %u",
                m_name.c_str(),
                m_threads,
                (unsigned int)m_batchkeys.size());
    /* Add data to index - HNSW multi-threaded example from
     * hnswlib:example_search_mt.cpp.
     */
    ParallelFor(
        0, m_batchkeys.size(), m_threads, [&](size_t row, size_t threadId) {
            m_alg_hnsw->addPoint(
                (void*)&(m_batch[row * m_space->get_data_size()]),
                m_batchkeys[row]);
        });

    m_batch.clear();
    m_batchkeys.clear();

    return true;
}

bool HNSWMemoryIndex::insertVector(VectorPtr vec, int dim, KeyTypeInteger id) {
    FP32* fvec = static_cast<FP32*>(vec);
    if (m_isParallelBuild) {
        // m_batch.insert(m_batch.end(), fvec, fvec + m_dim);
        m_batch.insert(
            m_batch.end(), (char*)vec, ((char*)vec + m_space->get_data_size()));
        m_batchkeys.push_back(id);

        if (m_batchkeys.size() == HNSW_PARALLEL_BUILD_UNIT_SIZE) {
            flushBatchParallel();
        }
    } else {
        if (!m_alg_hnsw)
            return false;  /// not built or loaded yet (#187)
        m_alg_hnsw->addPoint(fvec, id);
    }

    m_n_rows++;  // atomic
    m_isDirty = true;
    return true;
}

/* deleteVector - mark the key deleted (online DELETE/UPDATE). The node stays in
 * the graph; searches skip it, and the mark is written at the next checkpoint.
 * A later insertVector() of the same key clears the mark and updates the vector.
 */
bool HNSWMemoryIndex::deleteVector(KeyTypeInteger id) {
    auto* disk = dynamic_cast<hnswlib::HierarchicalDiskNSW<FP32>*>(m_alg_hnsw);
    if (!disk || m_isParallelBuild)
        return false;
    try {
        disk->markDelete(id);
    } catch (const std::runtime_error&) {
        return false;  /// key not in the index, or already deleted
    }
    if (m_n_rows > 0)
        m_n_rows--;
    m_isDirty = true;
    return true;
}

AbstractVectorIndex* VectorIndexCollection::open(const string& name,
                                                 const string& options,
                                                 const string& useraction) {
    lock_guard<mutex> l(m_mutex);  // exclusive

    MYVEC_LOG_DEBUG("Opening new index %s %s %s",
                name.c_str(),
                options.c_str(),
                useraction.c_str());

    AbstractVectorIndex* hnewindex = nullptr;

    MyVectorOptions vopt(options);
    string itype = vopt.getOption("type");
    transform(itype.begin(), itype.end(), itype.begin(),
              [](unsigned char c) { return static_cast<char>(toupper(c)); });

    /* A missing or unknown type is an error, not a silent KNN index: a typo
     * such as type=hnws, or a comment the parser could not read (#158), used
     * to build brute-force KNN and report SUCCESS.
     */
    if (!isValidIndexType(itype)) {
        MYVEC_LOG_ERROR("MyVector %s index type '%s' for %s, options = %s",
                        itype.empty() ? "missing" : "unknown",
                        itype.c_str(),
                        name.c_str(),
                        options.c_str());
        return nullptr;
    }

    if (itype == "HNSW" || itype == "HNSW_BV") {
        hnewindex = new HNSWMemoryIndex(name, options);
    } else {
        hnewindex = new KNNIndex(name, options);
    }

    m_indexes[name] = hnewindex;
    hnewindex->lockShared(); /* acquire lock before returning, matching get() */

    return hnewindex;
}

AbstractVectorIndex* VectorIndexCollection::get(const string& name) {
    lock_guard<mutex> l(m_mutex);
    AbstractVectorIndex* hindex = nullptr;
    if (m_indexes.find(name) != m_indexes.end()) {
        hindex = m_indexes[name];
        hindex->lockShared(); /* acquire lock before returning from here */
    } else {
        MYVEC_LOG_ERROR("VectorIndexCollection::get() index not found %s",
                        name.c_str());
    }

    return hindex;
}

bool VectorIndexCollection::close(AbstractVectorIndex* hindex) {
    lock_guard<mutex> l(m_mutex);

    hindex->unlockShared(); /* taken when opening the index */

    hindex->lockExclusive(); /* wait for all readers to drain */
    hindex->closeIndex();
    m_indexes.erase(m_indexes.find(hindex->getName()));
    hindex->unlockExclusive();

    delete hindex; /* no other thread can hold shared lock */
    return true;
}

/* FindEarliestBinlogFile() - Find the oldest binlog file from all
 * the "online" vector indexes checkpoint info.
 */
string VectorIndexCollection::FindEarliestBinlogFile() {
    string ret = "";
    for (auto entry : m_indexes) {
        string binlogfile;
        size_t binlogpos;
        if (entry.second->supportsIncrUpdates()) {
            entry.second->getLastUpdateCoordinates(binlogfile, binlogpos);
            if (ret == "")
                ret = binlogfile;
            else if (binlogfile < ret)
                ret = binlogfile;
        }
    }
    if (ret == "zzzzzz.bin")
        ret = "";
    MYVEC_LOG_DEBUG("FindEarliestBinlogFile : %s.", ret.c_str());
    return ret;
}

/* FindEarliestCheckpoint() - The oldest binlog position up to which an
 * "online" vector index is saved on disk. Indexes that were never built
 * (sentinel "zzzzzz.bin") are skipped. Returns false if there is none.
 */
bool VectorIndexCollection::FindEarliestCheckpoint(string& binlogfile,
                                                   size_t& binlogpos) {
    lock_guard<mutex> l(m_mutex);
    bool found = false;
    for (auto& entry : m_indexes) {
        if (!entry.second->supportsIncrUpdates())
            continue;
        string file;
        size_t pos = 0;
        entry.second->getLastUpdateCoordinates(file, pos);
        if (file.empty() || file == "zzzzzz.bin")
            continue;
        if (!found || file < binlogfile ||
            (file == binlogfile && pos < binlogpos)) {
            binlogfile = file;
            binlogpos = pos;
            found = true;
        }
    }
    return found;
}

VectorIndexCollection g_indexes;

/* The MYVECTOR* Annotations supported by this plugin */
const string MYVECTOR_COLUMN_A = "MYVECTOR(";
const string MYVECTOR_IS_ANN_A = "MYVECTOR_IS_ANN(";
const string MYVECTOR_SEARCH_A = "MYVECTOR_SEARCH";
const string MYVECTOR_DEFAULT_INDEX_TYPE = "type=KNN";

const string MYVECTOR_IS_ANN_USAGE =
    "MYVECTOR_IS_ANN('<vector col>','<id col>','<search_vec>'[,'<options>'])";
const string MYVECTOR_SEARCH_USAGE =
    "MYVECTOR_SEARCH(baseTable,idColumn,vectorColumn,queryTable[,options])";

/* MySQL Column COMMENT max. length is 1024.e.g comment with all fields set :
 * MYVECTOR Column
 * |type=HNSW,dim=1536,size=1000000,M=64,ef=100,track=updatets,threads=8,dist=L2
 */
const size_t MYVECTOR_MAX_COLUMN_INFO_LEN = 128;

/* For v1, let us restrict to 4096. OpenAI has 3072 dimension embeddings
 * now in model :  text-embedding-3-large. Technically, there is no limitation
 * in the VARBINARY datatype. MySQL's VECTOR datatype supports max of 16383.
 */
#ifdef MYVECTOR_COMPONENT_BUILD
extern ulong myvector_max_vector_dim;  // defined in myvector_component_config.cc
#else
ulong myvector_max_vector_dim = 4096;
#endif

/* findUnquotedMyVectorColumn() - find the first *unquoted* occurrence of
 * MYVECTOR_COLUMN_A ("MYVECTOR(") in a SQL statement: scans left to right,
 * tracking single-quote ('...'), double-quote ("...") and backtick (`...`)
 * delimited regions -- doubling the quote char ('' / "" / ``) is the escape
 * form inside all three, and a backslash also escapes the next char inside
 * '...'/"..." per MySQL's default sql_mode -- and only matches outside all
 * of them.
 *
 * Without this, a literal "MYVECTOR(" typed inside a quoted string, e.g.
 * the documented `wordvec VARBINARY(200) COMMENT 'MYVECTOR(type=HNSW,...)'`
 * column form (README.md), is mistaken for the inline MYVECTOR(...) DDL
 * annotation and rewritten into invalid SQL (ERROR 1064). See issue #130.
 */
static size_t findUnquotedMyVectorColumn(const string& s) {
    char inQuote = '\0';  // '\0' = not inside a quoted region, else the opening quote char
    for (size_t i = 0; i < s.size(); i++) {
        char c = s[i];
        if (inQuote) {
            if (c == '\\' && inQuote != '`' && i + 1 < s.size()) {
                i++;  // backslash-escaped char inside '...' / "..." -- skip it too
                continue;
            }
            if (c == inQuote) {
                if (i + 1 < s.size() && s[i + 1] == inQuote) {
                    i++;  // doubled quote char ('' / "" / ``) -- still inside the region
                    continue;
                }
                inQuote = '\0';  // end of quoted region
            }
            continue;
        }
        if (c == '\'' || c == '"' || c == '`') {
            inQuote = c;
            continue;
        }
        if (s.compare(i, MYVECTOR_COLUMN_A.length(), MYVECTOR_COLUMN_A) == 0)
            return i;
    }
    return string::npos;
}

/* rewriteMyVectorColumnDef() - rewrite the MYVECTOR(...) annotation in
 * CREATE TABLE & ALTER TABLE.
 */
bool rewriteMyVectorColumnDef(const string& query, string& newQuery,
                              string& error_msg) {
    // support multiple MYVECTOR(...) columns
    size_t pos;
    bool error = false;

    newQuery = query;
    while ((pos = findUnquotedMyVectorColumn(newQuery)) != string::npos) {
        size_t spos = pos + MYVECTOR_COLUMN_A.length();
        size_t epos = newQuery.find_first_of(')', pos);

        if (epos == string::npos) {
            error_msg = "MYVECTOR column terminating ) not found";
            MYVEC_LOG_ERROR("%s.", error_msg.c_str());
            error = true;
            break;
        }

        string colinfo = newQuery.substr(spos, (epos - spos));
        if (colinfo.length() > MYVECTOR_MAX_COLUMN_INFO_LEN) {
            error_msg = "MYVECTOR column info too long";
            MYVEC_LOG_ERROR("%s, length = %zu.", error_msg.c_str(),
                            colinfo.length());
            error = true;
            break;
        }

        MyVectorOptions vo(colinfo);

        if (!vo.isValid()) {
            error_msg = "MYVECTOR column options parse error";
            MYVEC_LOG_ERROR("%s, options=%s.", error_msg.c_str(),
                            colinfo.c_str());
            error = true;
            break;
        }

        string vtype = vo.getOption("type");
        if (vtype == "") {
            colinfo = MYVECTOR_DEFAULT_INDEX_TYPE + "," + colinfo;
            vtype = "KNN";
            vo.setOption("type", vtype);
        } else {
            transform(vtype.begin(), vtype.end(), vtype.begin(),
                      [](unsigned char c) { return static_cast<char>(toupper(c)); });
            if (!isValidIndexType(vtype)) {
                error_msg = "MYVECTOR column type invalid: " + vo.getOption("type") +
                            " (use KNN, HNSW or HNSW_BV)";
                MYVEC_LOG_ERROR("%s.", error_msg.c_str());
                error = true;
                break;
            }
        }

        if (vo.getOption("dim") == "") {
            error_msg = "MYVECTOR column dimension not defined";
            MYVEC_LOG_ERROR("%s.", error_msg.c_str());
            error = true;
            break;
        }

        bool addTrackingColumn = false;
        string trackingColumn;
        if (vo.getOption("track").length()) {
            addTrackingColumn = true;
            trackingColumn = vo.getOption("track");
        }

        bool dimValid = true;
        int dim = vo.getIntOption("dim", 0, &dimValid);

        if (!dimValid || dim <= 1 || (ulong)dim > myvector_max_vector_dim) {
            error_msg =
                "MYVECTOR column dimension incorrect " + to_string(dim);
            MYVEC_LOG_ERROR("%s.", error_msg.c_str());
            error = true;
            break;
        }

        /* dist is matched case-insensitively by the index implementations
         * (see HNSWMemoryIndex / KNNIndex constructors); reject anything
         * they wouldn't recognize here instead of letting it silently fall
         * back to L2 at index-build/open time. */
        string distOpt = vo.getOption("dist");
        if (distOpt.length()) {
            string distUpper = distOpt;
            transform(distUpper.begin(), distUpper.end(), distUpper.begin(),
                      [](unsigned char c) { return static_cast<char>(toupper(c)); });
            if (distUpper != "L2" && distUpper != "EUCLIDEAN" &&
                distUpper != "COSINE" && distUpper != "COSINENORM" &&
                distUpper != "ANGULAR" && distUpper != "IP") {
                error_msg = "MYVECTOR column dist option invalid: " + distOpt;
                MYVEC_LOG_ERROR("%s.", error_msg.c_str());
                error = true;
                break;
            }
        }

        size_t varblength = 0;
        if (vtype != "HNSW_BV")
            varblength = MyVectorStorageLength(dim);
        else
            varblength = MyVectorBVStorageLength(dim);  // binary vector

        string newColumn = "";
#if MYSQL_VERSION_ID >= 90000
        newColumn = "VECTOR(" + to_string(dim) +
                    ") COMMENT 'MYVECTOR Column |" + colinfo + "'";
#else
        newColumn = "VARBINARY(" + to_string(varblength) +
                    ") COMMENT 'MYVECTOR Column |" + colinfo + "'";
#endif

        if (addTrackingColumn) {
            newColumn =
                newColumn + ", " + trackingColumn +
                " TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE "
                "CURRENT_TIMESTAMP";
        }
        newQuery =
            newQuery.substr(0, pos) + newColumn + newQuery.substr(epos + 1);
    }  // while

    MYVEC_LOG_INFO("MYVECTOR column rewrite \n%s.", newQuery.c_str());
    return error;
}

/* rewriteMyVectorIsANN() - rewrite the "WHERE MYVECTOR_IS_ANN(...)" annotation
 */
bool rewriteMyVectorIsANN(const string& query, string& newQuery) {
    size_t pos;
    bool error = false;

    newQuery = query;
    while ((pos = newQuery.find(MYVECTOR_IS_ANN_A)) != string::npos) {
        size_t spos = pos + MYVECTOR_IS_ANN_A.length();
        size_t epos = spos + 1;
        int ob = 1;
        /* We can have nested () in MYVECTOR_IS_ANN
         * e.g MYVECTOR_IS_ANN(a,b,myvector_construct(...))
         */
        while (epos < newQuery.length()) {
            if (newQuery[epos] == '(')
                ob++;
            if (newQuery[epos] == ')') {
                ob--;
                if (!ob)
                    break;  /// done
            }
            epos++;
        }

        if (ob) {
            error = true;
            break;
        }

        string strparams = newQuery.substr(spos, (epos - spos));

        vector<string> annparams;
        split(strparams, annparams);

        if (annparams.size() < 3) {
            error = true;
            break;
        }

        // If the 4th top-level arg (options) is a bare integer k, convert it to
        // an 'nn=k' options string. myvector_ann_set expects a string arg; MySQL
        // sets lengths[n]=0 for integers, causing the options to be silently
        // skipped and JSON_TABLE to fail. Top-level commas are found outside
        // parens/brackets/quotes so that vector expressions like
        // myvector_construct('[1,2,3]') and a 5th filter argument such as
        // (SELECT JSON_ARRAYAGG(id) FROM t WHERE c IN (1,2)) work correctly.
        {
            vector<size_t> top_commas;
            int depth = 0;
            bool in_sq = false, in_dq = false;
            for (size_t ci = 0; ci < strparams.size(); ++ci) {
                char ch = strparams[ci];
                if (!in_sq && !in_dq) {
                    if (ch == '(' || ch == '[') ++depth;
                    else if (ch == ')' || ch == ']') --depth;
                    else if (ch == '\'') in_sq = true;
                    else if (ch == '"') in_dq = true;
                    else if (ch == ',' && depth == 0) top_commas.push_back(ci);
                } else if (in_sq && ch == '\'') in_sq = false;
                else if (in_dq && ch == '"') in_dq = false;
            }
            if (top_commas.size() == 3 || top_commas.size() == 4) {
                size_t abeg = top_commas[2] + 1;
                size_t aend = (top_commas.size() == 4) ? top_commas[3]
                                                       : strparams.size();
                string arg = strparams.substr(abeg, aend - abeg);
                size_t s = arg.find_first_not_of(" \t\r\n");
                size_t e = arg.find_last_not_of(" \t\r\n");
                if (s != string::npos) {
                    arg = arg.substr(s, e - s + 1);
                    if (arg.find_first_not_of("0123456789") == string::npos) {
                        strparams = strparams.substr(0, abeg) + " 'nn=" + arg +
                                    "'" + strparams.substr(aend);
                    }
                }
            }
        }

        string idcolexpr = annparams[1];
        if (idcolexpr.length() < 2) {
            error = true;
            break;
        }
        idcolexpr = idcolexpr.substr(
            1, idcolexpr.length() - 2);  // remove the single quote

        stringstream ss;
        ss << "( " << idcolexpr << " IN "
           << "(select `myvecid` from "
           << "(select myvector_ann_set(" << strparams << ") `_mvjson`) `_mvsrc`,"
           << "JSON_TABLE(`_mvsrc`.`_mvjson`, " << '"' << "$[*]" << '"'
           << " COLUMNS(`myvecid` BIGINT PATH \"$\")) `myvector_ann`) )";

        newQuery =
            newQuery.substr(0, pos) + ss.str() + newQuery.substr(epos + 1);
    }  // while

    MYVEC_LOG_INFO("MYVECTOR_IS_ANN query rewrite \n%s.", newQuery.c_str());
    return error;
}

/* rewriteMyVectorSearch - rewrite the MYVECTOR_SEARCH[...] annotation */
bool rewriteMyVectorSearch(const string& query, string& newQuery) {
    bool error = false;
    size_t pos;

    newQuery = query;
    string delim;
    /* No nested [] in MYVECTOR_SEARCH[...] */
    while ((pos = newQuery.find(MYVECTOR_SEARCH_A)) != string::npos) {
        size_t spos = pos + MYVECTOR_SEARCH_A.length();
        size_t epos;
        if (newQuery[spos] == '[')
            delim = "]";
        else if (newQuery[spos] == '{')
            delim = "}";
        else {
            error = true;
            break;
        }
        spos += delim.length();
        if ((epos = newQuery.find(delim, spos)) == string::npos) {
            error = true;
            break;
        }

        string strparams = newQuery.substr(spos, (epos - spos));

        vector<string> annparams;
        split(strparams, annparams);

        if (annparams.size() < 4 || annparams.size() > 5) {
            MYVEC_LOG_ERROR("Incorrect MYVECTOR_SEARCH syntax : %s\nExample usage : %s",
                            strparams.c_str(),
                            MYVECTOR_SEARCH_USAGE.c_str());
            error = true;
            break;
        }

        /*  select article5 from MYVECTOR_SEARCH[test.t1, id, test.t1.v1, query,
         * n=5]; */

        string basetable = annparams[0];
        string idcol = annparams[1];
        string vecindex = annparams[2];
        string queryt = annparams[3];

        string annopt = "";
        if (annparams.size() > 4)
            annopt = annparams[4];

        stringstream ss;
        /// The query table must have a vector column named 'searchvec'
        ss << basetable << " where " << idcol << " in (select myvecid from "
           << queryt << " b, json_table(myvector_ann_set('" << vecindex << "','"
           << idcol << "', searchvec, '" << annopt << "') , " << '"' << "$[*]"
           << '"' << " COLUMNS(`myvecid` BIGINT PATH " << '"' << "$" << '"'
           << ")) `myvector_ann`)";

        newQuery = newQuery.substr(0, pos) + ss.str() +
                   newQuery.substr(epos + delim.length());
    }

    return error;
}

/* myvector_query_rewrite() - Entrypoint of pre-parse query rewrite by this
 * plugin. This routine looks for CREATE TABLE, ALTER TABLE, SELECT and
 * EXPLAIN and presence of MYVECTOR and proceeds to perform  the query
 * transformation.
 */
bool myvector_query_rewrite(const string& query, string* rewritten_query) {
    /* Check if SQL is CREATE/ALTER/SELECT/EXPLAIN */
    if (query.length() == 0 || !strchr("CcAaSsEe", query[0]))
        return false;

    /* quick top-level check and exit if no MYVECTOR* pattern found in query */
    if (!strstr(query.c_str(), "MYVECTOR"))
        return false;

    static const regex create_table("^CREATE\\s+TABLE",
                                    regex::icase | regex::nosubs);
    static const regex alter_table("^ALTER\\s+TABLE",
                                   regex::icase | regex::nosubs);
    static const regex select_stmt("^SELECT\\s+", regex::icase | regex::nosubs);
    static const regex explain_stmt("^EXPLAIN\\s+",
                                    regex::icase | regex::nosubs);

    string newQuery = "";
    *rewritten_query = query;

    if ((regex_search(query, select_stmt) ||
         regex_search(query, explain_stmt))) {
        if (strstr(query.c_str(), MYVECTOR_IS_ANN_A.c_str())) {
            if (rewriteMyVectorIsANN(query, newQuery)) {
                newQuery = "";
            }
        } else if (strstr(query.c_str(), MYVECTOR_SEARCH_A.c_str())) {
            if (rewriteMyVectorSearch(query, newQuery)) {
                newQuery = "";
            }
        }
    } else if ((regex_search(query, create_table) ||
                regex_search(query, alter_table)) &&
               (strstr(query.c_str(), MYVECTOR_COLUMN_A.c_str()))) {
        string error_msg;
        if (rewriteMyVectorColumnDef(query, newQuery, error_msg)) {
            // Escape any single quotes before embedding in SIGNAL statement.
            string safe_msg;
            safe_msg.reserve(error_msg.size());
            for (char c : error_msg)
                safe_msg += (c == '\'') ? "''" : string(1, c);
            newQuery =
                "SIGNAL SQLSTATE 'HY000' SET MESSAGE_TEXT = '" + safe_msg + "'";
        }
    }

    if (newQuery.length())
        *rewritten_query = newQuery;

    return (*rewritten_query != query);
}

PLUGIN_EXPORT bool myvector_ann_set_init(UDF_INIT* initid,
                                         UDF_ARGS* args,
                                         char* message) {
    initid->ptr = nullptr;
    if (args->arg_count < 3 || args->arg_count > 5) {
        strcpy(message, ER_MYVECTOR_INCORRECT_ARGUMENTS);
        return true;  // error
    }

    /* Filter key list (5th arg) is read as text, whatever its SQL type */
    if (args->arg_count == 5)
        args->arg_type[4] = STRING_RESULT;

    char* col = args->args[0];
    AbstractVectorIndex* vi = g_indexes.get(col);
    if (!vi) {
        snprintf(message, MYSQL_ERRMSG_SIZE, ER_MYVECTOR_INDEX_NOT_FOUND " (%s)", col);
        return true;  // error
    }
    SharedLockGuard l(vi);

    /* Room for MYVECTOR_MAX_ANN_RETURN_COUNT keys of any size */
    initid->max_length = MYVECTOR_ANN_SET_MAX_LEN;
    initid->ptr = (char*)malloc(initid->max_length);
#ifndef MYVECTOR_COMPONENT_BUILD
    (*h_udf_metadata_service)->result_set(initid, "charset", latin1);
#endif

    /* tls_distances is now a static thread_local object, no allocation needed
     */
    tls_distances->clear();

    return false;
}

PLUGIN_EXPORT void myvector_ann_set_deinit(UDF_INIT* initid) {
    if (initid && initid->ptr)
        free(initid->ptr);
    /* tls_distances is a static thread_local object, no deallocation needed.
     * Clear it to free memory used by the map entries.
     */
    tls_distances->clear();
}

PLUGIN_EXPORT char* myvector_ann_set(UDF_INIT* initid,
                                     UDF_ARGS* args,
                                     char* result,
                                     unsigned long* length,
                                     unsigned char* is_null,
                                     unsigned char* error) {
    char* col = args->args[0];
    char* idcol = args->args[1];
    FP32* searchvec = (FP32*)args->args[2];
    const char* searchoptions = nullptr;

    if (!col || !idcol || !searchvec) {
        *error = 1;
        *is_null = 1;
        *length = 0;
        return initid->ptr;
    }

    if (args->arg_count >= 4)
        searchoptions = args->args[3];

    /* Optional 5th arg: keys of the rows that may be returned, e.g.
     * (SELECT JSON_ARRAYAGG(id) FROM t WHERE ...). NULL (no matching rows)
     * returns an empty set.
     */
    unordered_set<KeyTypeInteger> allowed;
    bool filtered = (args->arg_count == 5);
    if (filtered && args->args[4] &&
        !parseKeyList(args->args[4], args->lengths[4], allowed)) {
        *error = 1;
        *is_null = 1;
        *length = 0;
        return initid->ptr;
    }

    int nn = MYVECTOR_DEFAULT_ANN_RETURN_COUNT;
    int ef_search = 0;
    if (searchoptions && args->lengths[3]) {
        MyVectorOptions vo(searchoptions);

        /* How many neighbours to return? */
        nn = vo.getIntOption("nn", MYVECTOR_DEFAULT_ANN_RETURN_COUNT);
        if (nn <= 0)
            nn = MYVECTOR_DEFAULT_ANN_RETURN_COUNT;

        nn = min((const unsigned int)nn, MYVECTOR_MAX_ANN_RETURN_COUNT);

        ef_search = vo.getIntOption("ef_search", 0);
    }

    AbstractVectorIndex* vi = g_indexes.get(col);
    if (!vi) {
        *error = 1;
        *is_null = 1;
        *length = 0;
        return initid->ptr;
    }
    /* get() already took the shared lock; every later return on this vi must
     * go through something that releases it, so construct the guard before
     * any check that can return -- a bare `return` here would leak the lock
     * forever (no other path decrements it), hanging a future DROP/rebuild's
     * lockExclusive() wait for readers to drain (review of audit run-1,
     * Slice 3: PR #234). */
    SharedLockGuard l(vi);

    /* args->lengths[2] is the query vector's true byte length; nothing above
     * checks it against what the index's own dimension requires before the
     * fixed-stride reads below (OOB read otherwise -- audit run-1). Mirrors
     * myvector_distance's dim1==dim2 check. */
    unsigned long expected_bytes = MyVectorExpectedVectorBytes(vi);
    if (args->lengths[2] < expected_bytes) {
        MYVECTOR_UDF_FAIL("myvector_ann_set",
                          "query vector is %lu bytes, index '%s' requires %lu.",
                          args->lengths[2], col, expected_bytes);
        *error = 1;
        *is_null = 1;
        *length = 0;
        return initid->ptr;
    }

    stringstream ss;
    if (searchvec) {
        vector<KeyTypeInteger> result;
        /* ef_search is passed per call: storing it on the shared index raced
         * with concurrent searches and stuck for later queries (#165). */
        if (!filtered || !allowed.empty())
            vi->searchVectorNN(searchvec, vi->getDimension(), result, nn,
                               filtered ? &allowed : nullptr, ef_search);

        /* simple JSON list of neighbour rows Pkid */
        ss << "[";
        for (int i = 0; i < result.size(); i++) {
            if (i)
                ss << ",";
            ss << result[i];
        }
        ss << "]";
    } else {
        *is_null = 1;
        *error = 1;
        ss << "[NULL]";
    }

    *length = ss.str().length();
    result = initid->ptr;
    strcpy(result, ss.str().c_str());
    return result;
}

/* SQFloatVectorToBinaryVector - Simple scalar quantization to convert
 * a sequence of native floats to a binary vector. If the float value is
 * greater than 0, then corresponding bit is set to 1 in the binary vector.
 */
int SQFloatVectorToBinaryVector(FP32* fvec, unsigned long* ivec, int dim) {
    memset(ivec, 0, (dim / BITS_PER_BYTE));  // 3rd param is bytes

    unsigned long elem = 0;
    unsigned long idx = 0;

    for (int i = 0; i < dim; i++) {
        elem = elem << 1;
        if (fvec[i] > 0) {
            elem = elem | 1;
        }

        if (((i + 1) % 64) == 0) {  // 8 bytes packed (i.e 64 dims in 1 ulong)
            ivec[idx] = elem;
            elem = 0;
            idx++;
        }
    }
    /// TODO : if dim is not a mulitple of 64
    return (idx * sizeof(unsigned long));  // number of bytes
}

/* Forward declaration for o=bv path */
char* myvector_construct_bv(const std::string& srctype,
                            char* src,
                            char* dst,
                            unsigned long srclen,
                            unsigned long* length,
                            unsigned char* is_null,
                            unsigned char* error);

/* Helper: perform myvector_construct conversion. Used for both constant-arg
 * caching (in init) and per-row conversion. Returns true on success.
 */
static bool myvector_construct_do_convert(char* ptr,
                                          unsigned long ptrlen,
                                          const char* opt,
                                          unsigned long optlen,
                                          char* retvec,
                                          unsigned long* retlen,
                                          unsigned char* is_null,
                                          unsigned char* error) {
    /* retvec always points sizeof(size_t) bytes into the MYVECTOR_CONSTRUCT_MAX_LEN
     * buffer allocated by myvector_construct_init (both call sites use that same
     * allocation) -- trailer is the metadata+checksum this function itself appends
     * below, so every write site must stay within capacity, not MYVECTOR_CONSTRUCT_MAX_LEN. */
#if MYSQL_VERSION_ID < 90000
    const unsigned long trailer = sizeof(unsigned int) + sizeof(ha_checksum);
#else
    const unsigned long trailer = 0;
#endif
    const unsigned long capacity = MYVECTOR_CONSTRUCT_MAX_LEN - trailer;

    bool skipConvert = false;
    if (!opt || !optlen)
        opt = "i=string,o=float";
    else {
        MyVectorOptions vo(string(opt, optlen));
        if (vo.getOption("i") == "float" && vo.getOption("o") == "float")
            skipConvert = true;
        if (vo.getOption("o") == "bv") {
            myvector_construct_bv(vo.getOption("i"),
                                 ptr,
                                 retvec,
                                 ptrlen,
                                 retlen,
                                 is_null,
                                 error);
            return (*error == 0);
        }
    }

    unsigned long len = 0;
    if (skipConvert) {
        if ((ptrlen % sizeof(FP32)) != 0) {
            *error = 1;
            return false;
        }
        if (ptrlen > capacity) {
            MYVECTOR_UDF_FAIL("myvector_construct",
                              "input vector too large: %lu bytes, max %lu.",
                              ptrlen, capacity);
            *error = 1;
            return false;
        }
        memcpy(retvec, ptr, ptrlen);
        len = ptrlen;
    } else {
        char* start = nullptr;
        char endch;
        if ((start = strchr(ptr, '[')))
            endch = ']';
        else if ((start = strchr(ptr, '{')))
            endch = '}';
        else if ((start = strchr(ptr, '(')))
            endch = ')';
        else {
            start = ptr;
            endch = '\0';
        }
        if (endch)
            start++;
        char* p = start;
        while (*p && *p != endch) {
            while (*p && (*p == ' ' || *p == ','))
                p++;
            char* p1 = p;
            while (*p && *p != ' ' && *p != ',' && *p != endch)
                p++;
            char buff[64];
            size_t toklen = (size_t)(p - p1);
            /* A legitimate float literal never needs anywhere close to 63
             * characters; reject instead of silently truncating and parsing
             * whatever garbage is left -- atof() on a truncated giant digit
             * string can overflow to +-Inf with *error left at 0 (success),
             * corrupting later distance ranking instead of failing loudly
             * (review of audit run-1, Slice 3: PR #234). */
            if (toklen >= sizeof(buff)) {
                MYVECTOR_UDF_FAIL("myvector_construct",
                                  "numeric token too long (max %zu characters).",
                                  sizeof(buff) - 1);
                *error = 1;
                return false;
            }
            strncpy(buff, p1, toklen);
            buff[toklen] = 0;
            if (len + sizeof(FP32) > capacity) {
                MYVECTOR_UDF_FAIL("myvector_construct",
                                  "input vector too large: max %lu bytes.",
                                  capacity);
                *error = 1;
                return false;
            }
            FP32 fval = atof(buff);
            memcpy(&retvec[len], &fval, sizeof(FP32));
            len += sizeof(FP32);
        }
    }

#if MYSQL_VERSION_ID < 90000
    unsigned int metadata = MYVECTOR_V1_FP32_METADATA;
    memcpy(&retvec[len], &metadata, sizeof(metadata));
    len += sizeof(metadata);
    ha_checksum cksum = my_checksum(0, (const unsigned char*)retvec, len);
    memcpy(&retvec[len], &cksum, sizeof(cksum));
    len += sizeof(cksum);
#endif
    *retlen = len;
    return true;
}

char* myvector_construct_bv(const std::string& srctype,
                            char* src,
                            char* dst,
                            unsigned long srclen,
                            unsigned long* length,
                            unsigned char* is_null,
                            unsigned char* error) {
    /* dst always points sizeof(size_t) bytes into the same
     * MYVECTOR_CONSTRUCT_MAX_LEN-capacity buffer myvector_construct_do_convert
     * writes into -- see its own capacity comment. */
#if MYSQL_VERSION_ID < 90000
    const unsigned long trailer = sizeof(unsigned int) + sizeof(ha_checksum);
#else
    const unsigned long trailer = 0;
#endif
    const unsigned long capacity = MYVECTOR_CONSTRUCT_MAX_LEN - trailer;
    unsigned long retlen = 0;

    if (srctype == "bv") {
        if (srclen > capacity) {
            MYVECTOR_UDF_FAIL("myvector_construct",
                              "input vector too large: %lu bytes, max %lu.",
                              srclen, capacity);
            *error = 1;
            return dst;
        }
        memcpy(
            dst, src, srclen);  // src is bytes representing the binary vector
        retlen = srclen;
    } else if (srctype == "float") {
        int dim = MyVectorDimFromStorageLength(srclen);
        /* dim <= 0 (e.g. srclen too short) must be rejected outright, not
         * treated as "0 bytes needed": SQFloatVectorToBinaryVector's own
         * memset(ivec, 0, dim / BITS_PER_BYTE) divides a signed dim by an
         * unsigned constant, so a negative dim silently becomes a ~4-billion
         * byte count instead of a negative one, overflowing dst far worse
         * than any oversized-but-positive dim this check already catches
         * (review of audit run-1, Slice 3: PR #234). */
        if (dim <= 0) {
            MYVECTOR_UDF_FAIL("myvector_construct",
                              "input vector too short to derive a dimension.");
            *error = 1;
            return dst;
        }
        unsigned long predicted_bytes =
            (((unsigned long)dim + 63) / 64) * sizeof(unsigned long);
        if (predicted_bytes > capacity) {
            MYVECTOR_UDF_FAIL("myvector_construct",
                              "input vector too large: dim %d, max %lu bytes.",
                              dim, capacity);
            *error = 1;
            return dst;
        }
        retlen =
            SQFloatVectorToBinaryVector((FP32*)src, (unsigned long*)dst, dim);
    } else if (srctype == "string") {
        char *start = nullptr, *ptr = nullptr;
        char endch;

        if ((start = strchr(src, '[')))
            endch = ']';
        else if ((start = strchr(src, '{')))
            endch = '}';
        else if ((start = strchr(src, '(')))
            endch = ')';
        else {
            start = src;
            endch = '\0';
        }
        if (endch)
            start++;

        ptr = start;

        while (*ptr && *ptr != endch) {
            while (*ptr && (*ptr == ' ' || *ptr == ','))
                ptr++;
            char* p1 = ptr;
            while (*ptr && *ptr != ' ' && *ptr != ',' && *ptr != endch)
                ptr++;
            char buff[64];
            size_t toklen = (size_t)(ptr - p1);
            /* A legitimate 0-255 byte-value token never needs anywhere close
             * to 63 characters; reject instead of silently truncating and
             * parsing whatever garbage is left -- atoi() on a truncated giant
             * digit string is UB on overflow (commonly, but not reliably,
             * saturating) with *error left at 0 (success) either way (review
             * of audit run-1, Slice 3: PR #234). */
            if (toklen >= sizeof(buff)) {
                MYVECTOR_UDF_FAIL("myvector_construct",
                                  "byte-value token too long (max %zu characters).",
                                  sizeof(buff) - 1);
                *error = 1;
                return dst;
            }

            strncpy(buff, p1, toklen);
            buff[toklen] = 0;

            if (retlen >= capacity) {
                MYVECTOR_UDF_FAIL("myvector_construct",
                                  "input vector too large: max %lu bytes.",
                                  capacity);
                *error = 1;
                return dst;
            }
            dst[retlen] = (unsigned char)(atoi(buff));

            retlen += sizeof(unsigned char);
        }  // while
    }  // else

#if MYSQL_VERSION_ID < 90000
    unsigned int metadata = MYVECTOR_V1_BV_METADATA;
    memcpy(&dst[retlen], &metadata, sizeof(metadata));
    retlen += sizeof(metadata);

    ha_checksum cksum = my_checksum(0, (const unsigned char*)dst, retlen);
    memcpy(&dst[retlen], &cksum, sizeof(cksum));
    retlen += sizeof(cksum);
#endif
    *length = retlen;
    return dst;
}

PLUGIN_EXPORT bool myvector_construct_init(UDF_INIT* initid,
                                           UDF_ARGS* args,
                                           char* message) {
    if (args->arg_count < 1 || args->arg_count > 2) {
        strcpy(message, ER_MYVECTOR_INCORRECT_ARGUMENTS);
        return true;  // error
    }
    initid->max_length = MYVECTOR_CONSTRUCT_MAX_LEN;
    size_t alloc_size = sizeof(size_t) + MYVECTOR_CONSTRUCT_MAX_LEN;
    char* buf = (char*)malloc(alloc_size);
    if (!buf) {
        strcpy(message, "myvector_construct: malloc failed");
        return true;
    }
    *(size_t*)buf = 0;  // 0 = no cache

    /* Issue #79: cache only when ALL args are constant. If arg_count >= 2 and
     * args[1] is non-constant (NULL in init), we must not cache, else we would
     * ignore per-row options and return incorrect results. */
    bool can_cache = (args->args[0] != nullptr && args->lengths[0] > 0);
    if (args->arg_count >= 2 && args->args[1] == nullptr)
        can_cache = false;  // options vary per row
    if (can_cache) {
        const char* opt =
            (args->arg_count >= 2 && args->args[1]) ? args->args[1] : nullptr;
        unsigned long optlen =
            (args->arg_count >= 2 && args->args[1]) ? args->lengths[1] : 0;
        unsigned char is_null = 0, err = 0;
        unsigned long result_len = 0;
        if (myvector_construct_do_convert(args->args[0],
                                          args->lengths[0],
                                          opt,
                                          optlen,
                                          buf + sizeof(size_t),
                                          &result_len,
                                          &is_null,
                                          &err) &&
            !err) {
            *(size_t*)buf = result_len;
        }
    }
    initid->ptr = buf;
    return false;
}

/* SQL function : MYVECTOR_CONSTRUCT(embedding_string)
 *   Input   : Embedding vector string e.g
 *             "[-0.006929283495992422 -0.005336422007530928 ...]"
 *   Returns : Serialized "myvector" to store in VARBINARY column.
 *             Each floating point value in the embedding vector string is
 *             converted to a 4-byte "FP32" (IEEE754). At the end, 2 4-byte
 *             integers are added :-
 *             1) 4-byte Metadata
 *             2) 4-byte Checksum (computed from all the FP32s plus Metadata)
 *
 *             If input string is NULL, then NULL is returned.
 */

PLUGIN_EXPORT char* myvector_construct(UDF_INIT* initid,
                                       UDF_ARGS* args,
                                       char* result,
                                       unsigned long* length,
                                       unsigned char* is_null,
                                       unsigned char* error) {
    size_t* cache_len = (size_t*)initid->ptr;
    if (*cache_len > 0) {
        *length = *cache_len;
        return initid->ptr + sizeof(size_t);
    }

    char* ptr = args->args[0];
    if (!ptr) {
        *is_null = 1;
        return initid->ptr + sizeof(size_t);
    }

    const char* opt = (args->arg_count >= 2 && args->args[1]) ? args->args[1] : nullptr;
    unsigned long optlen =
        (args->arg_count >= 2 && args->args[1]) ? args->lengths[1] : 0;

    /* o=bv branches to myvector_construct_bv - not handled by helper */
    if (opt && optlen) {
        MyVectorOptions vo(string(opt, optlen));
        if (vo.getOption("o") == "bv")
            return myvector_construct_bv(vo.getOption("i"),
                                         ptr,
                                         initid->ptr + sizeof(size_t),
                                         args->lengths[0],
                                         length,
                                         is_null,
                                         error);
    }

    unsigned long retlen = 0;
    if (!myvector_construct_do_convert(ptr,
                                        args->lengths[0],
                                        opt,
                                        optlen,
                                        initid->ptr + sizeof(size_t),
                                        &retlen,
                                        is_null,
                                        error))
        return initid->ptr + sizeof(size_t);

    *length = retlen;
    return initid->ptr + sizeof(size_t);
}

PLUGIN_EXPORT void myvector_construct_deinit(UDF_INIT* initid) {
    if (initid && initid->ptr)
        free(initid->ptr);
}

PLUGIN_EXPORT bool myvector_display_init(UDF_INIT* initid,
                                         UDF_ARGS* args,
                                         char* message) {
    if (args->arg_count == 0 || args->arg_count > 2) {
        strcpy(message, ER_MYVECTOR_INCORRECT_ARGUMENTS);
        return true;  // error
    }
    initid->max_length = MYVECTOR_DISPLAY_MAX_LEN;
    initid->ptr = (char*)malloc(MYVECTOR_DISPLAY_MAX_LEN);
#ifndef MYVECTOR_COMPONENT_BUILD
    (*h_udf_metadata_service)->result_set(initid, "charset", latin1);
#endif
    return false;
}

PLUGIN_EXPORT char* myvector_display(UDF_INIT* initid,
                                     UDF_ARGS* args,
                                     char* result,
                                     unsigned long* length,
                                     char* is_null,
                                     char* error) {
    unsigned char* bvec = (unsigned char*)args->args[0];
    FP32* fvec = (FP32*)args->args[0];
    if (!bvec || !args->lengths[0]) {
        *is_null = 1;  // NULL in, NULL out, for this row only (#170)
        return result;
    }

    /* We assume MySQL passes the VARBINARY value aligned correctly i.e aligned
     * at highest 8 bytes. Need to verify that when supporting double, FP16 etc.
     */

    int precision = 0;
    if (args->arg_count > 1 && args->args[1] && args->lengths[1]) {
        precision = atoi(args->args[1]);
    }
    if (!precision)
        precision = MYVECTOR_DISPLAY_DEF_PREC;

    stringstream ostr;

#if MYSQL_VERSION_ID < 90000
    ha_checksum cksum1;
    char* raw = args->args[0];
    /* Checksum is at the end */
    memcpy((char*)&cksum1,
           &(raw[args->lengths[0] - sizeof(ha_checksum)]),
           sizeof(ha_checksum));
    ha_checksum cksum2 = my_checksum(
        0, (const unsigned char*)raw, args->lengths[0] - sizeof(ha_checksum));
    if (cksum1 != cksum2) {
        *error = 1;
        return "<invalid vector>";
    }
#endif

    int dim = 0;
#if MYSQL_VERSION_ID < 90000
    unsigned int metadata = 0;
    memcpy((char*)&metadata,
           &bvec[args->lengths[0] - MYVECTOR_COLUMN_EXTRA_LEN],
           sizeof(metadata));

    if (metadata == MYVECTOR_V1_FP32_METADATA) {
        bvec = nullptr;
        dim = MyVectorDimFromStorageLength(args->lengths[0]);
    } else if (metadata == MYVECTOR_V1_BV_METADATA) {
        fvec = nullptr;
        dim = MyVectorBVDimFromStorageLength(args->lengths[0]);
        dim = dim / 8; /* bit-packet */
    } else {           /* 'old' v0 vectors */
        bvec = nullptr;
        dim = (args->lengths[0]) / sizeof(FP32);
    }
#else
    dim = MyVectorDimFromStorageLength(args->lengths[0]);
#endif

    ostr << "[";
    ostr << std::setprecision(precision);
    for (int i = 0; i < dim; i++) {
        if (i)
            ostr << ", ";
        if (fvec) {
            ostr << *fvec;
            fvec++;
        } else {
            ostr << (unsigned int)*bvec;
            bvec++;
        }
    }
    ostr << "]";

    result = initid->ptr;
    strcpy(result, ostr.str().c_str());
    *length = ostr.str().length();

    return result;
}

PLUGIN_EXPORT void myvector_display_deinit(UDF_INIT* initid) {
    if (initid && initid->ptr)
        free(initid->ptr);
}

/* UDF MYVECTOR_DISTANCE() Implementation */
PLUGIN_EXPORT bool myvector_distance_init(UDF_INIT* initid,
                                          UDF_ARGS* args,
                                          char* message) {
    if (args->arg_count < 2 || args->arg_count > 3) {
        strcpy(message, ER_MYVECTOR_INCORRECT_ARGUMENTS);
        return true;  /// error
    }
    initid->maybe_null = true;
    initid->ptr = nullptr;
    /* A constant metric is checked once, here, and its function cached: an
     * unknown name fails the statement before any row is read (#170). */
    if (args->arg_count == 3 && args->args[2]) {
        MyVectorDistanceFn fn = myvector_distance_fn(args->args[2], args->lengths[2]);
        if (!fn) {
            snprintf(message, MYSQL_ERRMSG_SIZE,
                     "myvector_distance(): unknown distance metric '%.*s'; "
                     "use L2, EUCLIDEAN, Cosine or IP",
                     (int)std::min<unsigned long>(args->lengths[2], 64), args->args[2]);
            return true;
        }
        initid->ptr = reinterpret_cast<char*>(fn);
    }
    return false;
}

PLUGIN_EXPORT bool myvector_construct_binaryvector_init(UDF_INIT* initid,
                                                        UDF_ARGS* args,
                                                        char* message) {
    if (args->arg_count != 1) {
        strcpy(message,
               "Incorrect arguments, usage : "
               "myvector_construct_binary_vector(vec_col_expr)");
        return true;  // error
    }
    initid->max_length = 1024;
    initid->ptr = (char*)malloc(initid->max_length);
    return false;
}

PLUGIN_EXPORT bool myvector_hamming_distance_init(UDF_INIT* initid,
                                                  UDF_ARGS* args,
                                                  char* message) {
    return false;
}

PLUGIN_EXPORT double myvector_hamming_distance(UDF_INIT*,
                                               UDF_ARGS* args,
                                               char* is_null,
                                               char*) {
    double dist = 0.0;
    unsigned long* v1 = (unsigned long*)(args->args[0]);
    unsigned long* v2 = (unsigned long*)(args->args[1]);

    size_t dim = args->lengths[0] * 8;

    return HammingDistanceFn(v1, v2, &dim);
}

PLUGIN_EXPORT double myvector_distance(UDF_INIT* initid,
                                       UDF_ARGS* args,
                                       char* is_null,
                                       char* error) {
    FP32* v1 = (FP32*)(args->args[0]);
    FP32* v2 = (FP32*)(args->args[1]);

    /* NULL in, NULL out, for this row only. Setting *error here made every
     * later row NULL too (#170). */
    if (!v1 || !v2 || !args->lengths[0] || !args->lengths[1]) {
        *is_null = 1;
        return 0.0;
    }

    int dim1 = MyVectorDimFromStorageLength(args->lengths[0]);
    int dim2 = MyVectorDimFromStorageLength(args->lengths[1]);
    /* Different dimensions are an error, not a distance over the shorter
     * length: that could return 0, a "perfect match" (#171). */
    if (dim1 != dim2 || dim1 <= 0) {
        MYVECTOR_UDF_FAIL("myvector_distance",
                          "vectors have different dimensions (%d and %d)", dim1, dim2);
        *error = 1;
        return 0.0;
    }

    MyVectorDistanceFn distfn = computeL2Distance;  // default
    if (initid->ptr) {
        distfn = reinterpret_cast<MyVectorDistanceFn>(initid->ptr);
    } else if (args->arg_count == 3) {
        if (!args->args[2]) {
            *is_null = 1;  // NULL metric: NULL distance
            return 0.0;
        }
        distfn = myvector_distance_fn(args->args[2], args->lengths[2]);
        if (!distfn) {
            MYVECTOR_UDF_FAIL("myvector_distance",
                              "unknown distance metric '%.*s'; use L2, EUCLIDEAN, Cosine or IP",
                              (int)std::min<unsigned long>(args->lengths[2], 64), args->args[2]);
            *error = 1;
            return 0.0;
        }
    }
    return distfn(v1, v2, dim1);
}

PLUGIN_EXPORT void myvector_distance_deinit(UDF_INIT* initid) {}

/* myvector_search_open_udf returns index status and messages that can be
 * longer than MySQL's 255-byte UDF result buffer (the status grew with the
 * online skip counts, #205), so it writes to its own MYVECTOR_BUFF_SIZE buffer,
 * the size of the VARCHAR(1024) the SQL procedures read it into. */
PLUGIN_EXPORT bool myvector_search_open_udf_init(UDF_INIT* initid,
                                                 UDF_ARGS* args,
                                                 char* message) {
    if (args->arg_count != 5) {
        strcpy(message, "Incorrect arguments to MyVector internal UDF.");
        return true;
    }
    initid->ptr = static_cast<char*>(malloc(MYVECTOR_BUFF_SIZE));
    if (!initid->ptr) {
        strcpy(message, "MyVector: out of memory.");
        return true;
    }
    initid->max_length = MYVECTOR_BUFF_SIZE - 1;
    return false;
}

void BuildMyVectorIndexSQL(const char* db,
                           const char* table,
                           const char* idcol,
                           const char* veccol,
                           const char* action,
                           const char* trackingColumn,
                           AbstractVectorIndex* vi,
                           char* errorbuf);

bool ValidateVectorIndexRequest(const char* vecid,
                                const char* pkidcol,
                                string& comment,
                                string& error);

void myvector_open_index_impl(char* vecid,
                              char* details,
                              char* pkidcol,
                              char* action,
                              char* extra,
                              char* result,
                              bool validateAgainstSchema) {
    bool existing = true;

    /* myvector_search_open_udf is SQL-registered like any ordinary function;
     * nothing prevents a caller from invoking it directly instead of through
     * MYVECTOR_INDEX_INTERNAL (sql/myvectorplugin.sql), which would otherwise
     * be the only thing confirming `vecid` names a real MYVECTOR column and
     * that `details`/`pkidcol` describe it truthfully. Re-derive the same
     * facts here, from information_schema, for every externally-triggered
     * call (validateAgainstSchema is false only for the plugin's own
     * startup scan in OpenAllOnlineVectorIndexes, which already read this
     * same comment from information_schema moments earlier). `details` is
     * replaced with the column's real comment on success: a caller's own
     * "details" string is never trusted past this point.
     */
    string validatedComment;
    const char* effective_details = details;
    if (validateAgainstSchema) {
        string validationError;
        if (!ValidateVectorIndexRequest(
                vecid, pkidcol, validatedComment, validationError)) {
            snprintf(result, 255, "ERROR: %.200s", validationError.c_str());
            return;
        }
        effective_details = validatedComment.c_str();
    }

    /* Admin operations usecases :-

     1. Load rows into base table -> call myvector("build"). If VectorIndex is
     already populated, it will be discarded and rebuilt. Persist index after
     build!
     2. Patch existing rows, delete a few rows, insert new rows -> call
     myvector("build")
     3. Any trouble with existing index  -> call myvector("drop")
     4. After reboot/restart VM -> call myvector("load") (or myvector("build"))
     5. If tracking column technique -> call myvector("refresh") - index should
     be loaded first. refresh will not persist. After reboot/restart, call
     myvector("load") followed by myvector("refresh")
     6. For explicit persist  -> call myvector("save"), needed after "refresh"
    */

    /* A missing or unknown type in the column comment is an error, not a KNN
     * index. build, refresh and load read the comment, so check it even when
     * the index is already open; status, drop and save do not, so an index
     * whose comment was later broken can still be dropped.
     */
    string itype = MyVectorOptions(effective_details).getOption("type");
    bool readsComment = !strcmp(action, "build") || !strcmp(action, "refresh") ||
                        !strcmp(action, "load");
    AbstractVectorIndex* vi = nullptr;
    if (!readsComment || isValidIndexType(itype))
        vi = g_indexes.get(vecid);
    if (!vi) {
        if (isValidIndexType(itype))
            vi = g_indexes.open(vecid, effective_details, action);
        if (!vi) {
            /* result is at least 255 bytes (MySQL's UDF string buffer) */
            if (itype.empty())
                snprintf(result,
                         255,
                         "ERROR: missing index type for %.80s."
                         " Use type=KNN, HNSW or HNSW_BV",
                         vecid);
            else if (!isValidIndexType(itype))
                snprintf(result,
                         255,
                         "ERROR: unknown index type '%.40s' for %.80s."
                         " Use type=KNN, HNSW or HNSW_BV",
                         itype.c_str(),
                         vecid);
            else
                strcpy(result, "Failed to open index");
            return;
        }
        existing = false;
    }
    SharedLockGuard l(vi);

    string trackingColumn = "";
    int nthreads = 0;

    MyVectorOptions vo(effective_details);
    if (vo.getOption("track").length()) {
        trackingColumn = vo.getOption("track");
    }
    if (!strcmp(action, "refresh") && !trackingColumn.length()) {
        strcpy(result, "Tracking column not found for incremental refresh.");
        return;
    }
    if (vo.getOption("threads").length()) {  // override
        nthreads = vo.getIntOption("threads", myvector_index_bg_threads);
    } else {
        nthreads = myvector_index_bg_threads;
    }

    if (!strcmp(action, "save")) {
        if (!vi->saveIndex(myvector_index_dir))
            strcpy(result,
                   "ERROR: index could not be saved to disk"
                   " (see the server log)");
    } else if (!strcmp(action, "status")) {
        string s = vi->getStatus();
        string idx(vecid);  // db.table.column; skips are counted per db.table
        s += myvector_online_skip_status(idx.substr(0, idx.rfind('.')));
        /* result holds MYVECTOR_BUFF_SIZE bytes: the UDF's own buffer, or the
         * caller's in the component's startup load */
        snprintf(result, MYVECTOR_BUFF_SIZE, "%s", s.c_str());
    } else if (!strcmp(action, "drop")) {
        vi->dropIndex(myvector_index_dir);
        l.release();
        g_indexes.close(vi);
        vi = nullptr;
    }

    else if (!strcmp(action, "load")) {
        MYVEC_LOG_DEBUG("Loading index %s.", vecid);
        vi->loadIndex(myvector_index_dir);  // will handle 'reload' also
    } else if (!strcmp(action, "build")) {
        vi->dropIndex(myvector_index_dir);

        vi->initIndex();  // start new

        if (nthreads >= 2)
            vi->startParallelBuild(nthreads);
    } else if (!strcmp(action, "refresh")) {
        if (nthreads >= 2)
            vi->startParallelBuild(nthreads);
    }

    /* A rebuilt or reloaded index: a skip of a reason already warned about is
     * worth a new warning (#205). */
    if (!strcmp(action, "load") || !strcmp(action, "build")) {
        string idx(vecid);
        myvector_online_skip_rearm(idx.substr(0, idx.rfind('.')));
    }

    if (!strcmp(action, "build") || !strcmp(action, "refresh")) {
        char errorbuf[1024];
        char *db, *table, *veccol;
        db = vecid;
        table = strchr(db, '.');
        *table = 0;
        table++;
        veccol = strchr(table, '.');
        *veccol = 0;
        veccol++;
        BuildMyVectorIndexSQL(db,
                              table,
                              pkidcol,
                              veccol,
                              action,
                              trackingColumn.c_str(),
                              vi,
                              errorbuf);
        strcpy(result, errorbuf);
        if (!strcmp(action, "refresh")) {
            unsigned long lastts = vi->getUpdateTs();
            char timebuf[64];
            strcat(result, ", Vector Index refreshed at : ");
#ifdef _WIN32
            time_t t = (time_t)lastts;
            struct tm tm_buf;
            gmtime_s(&tm_buf, &t);
            strftime(timebuf,
                     sizeof(timebuf),
                     "%a %b %d %H:%M:%S %Y\n",
                     &tm_buf);
#else
            time_t t = (time_t)lastts;
            struct tm tm_buf;
            gmtime_r(&t, &tm_buf);
            strftime(timebuf, sizeof(timebuf), "%a %b %d %H:%M:%S %Y\n", &tm_buf);
#endif
            strcat(result, timebuf);
        }

        if (!vi->saveIndex(myvector_index_dir, action))
            strcpy(result,
                   "ERROR: index built but could not be saved to disk"
                   " (see the server log)");
    }
    return;
}

PLUGIN_EXPORT char* myvector_search_open_udf(UDF_INIT* initid,
                                             UDF_ARGS* args,
                                             char* /* MySQL's 255-byte buffer: too small */,
                                             unsigned long* length,
                                             unsigned char* is_null,
                                             unsigned char*) {
    char* vecid = args->args[0];
    char* details = args->args[1];
    char* pkidcol = args->args[2];
    char* action = args->args[3];
    char* extra = args->args[4];

    MYVEC_LOG_INFO("myvector_search_open() params %s %s %s %s %s",
                   vecid,
                   details,
                   pkidcol,
                   action,
                   extra);

    char* result = initid->ptr;  // MYVECTOR_BUFF_SIZE bytes
    strcpy(result, "SUCCESS");

    myvector_open_index_impl(
        vecid, details, pkidcol, action, extra, result, true);

    *length = strlen(result);
    return result;
}

PLUGIN_EXPORT void myvector_search_open_udf_deinit(UDF_INIT* initid) {
    free(initid->ptr);
    initid->ptr = nullptr;
}

PLUGIN_EXPORT bool myvector_search_save_udf_init(UDF_INIT* initid,
                                                 UDF_ARGS* args,
                                                 char* message) {
    if (args->arg_count != 5) {
        strcpy(message, "Incorrect arguments to MyVector internal UDF.");
        return true;
    }
    return false;
}

PLUGIN_EXPORT char* myvector_search_save_udf(UDF_INIT*,
                                             UDF_ARGS* args,
                                             char* result,
                                             unsigned long* length,
                                             unsigned char* is_null,
                                             unsigned char*) {
    char* vecid = args->args[0];
    char* details = args->args[1];
    char* pkidcol = args->args[2];
    char* action = args->args[3];
    char* extra = args->args[4];

    if (!g_indexes.get(vecid)) {
        MYVEC_LOG_ERROR("Index %s is not opened for build/refresh.", vecid);
        strcpy(result, "FAILED");
        *length = 6;  // "FAILED" is 6 characters; 7 would read one byte past the NUL
        return result;
    }

    AbstractVectorIndex* vi = g_indexes.get(vecid);
    if (!vi->saveIndex(myvector_index_dir, action)) {
        strcpy(result, "FAILED");
        *length = 6;
        return result;
    }
    strcpy(result, "SUCCESS");
    *length = 7;
    return result;
}

PLUGIN_EXPORT void myvector_search_save_udf_deinit() {}

PLUGIN_EXPORT bool myvector_search_add_row_udf_init(UDF_INIT* initid,
                                                    UDF_ARGS* args,
                                                    char* message) {
    char* vecid = args->args[0];
    char* details = args->args[1];

    /* Check if index is open and 'cache' it */
    if (!g_indexes.get(vecid)) {
        MYVEC_LOG_ERROR("Index %s is not opened for update.", vecid);
        return true;
    }

    initid->ptr = (char*)(g_indexes.get(vecid));
    return false;
}

static bool isZeroVector(const FP32* vec, int dim) {
    double norm = 0.0;
    for (int i = 0; i < dim; i++)
        norm += (double)vec[i] * vec[i];
    return (norm == 0.0);
}

/* UDF : myvector_search_add_row_udf() */
PLUGIN_EXPORT long long myvector_search_add_row_udf(UDF_INIT* initid,
                                                    UDF_ARGS* args,
                                                    char* is_null,
                                                    char* error) {
    char* vecid = args->args[0];
    char* details = args->args[1];

    long long pkid = *((long long*)args->args[2]);
    char* vecval = args->args[3];
    int dims = MyVectorDimFromStorageLength(args->lengths[3]);

    AbstractVectorIndex* vi = (AbstractVectorIndex*)(initid->ptr);
    if (vi) {
        if (args->lengths[3] < MyVectorExpectedVectorBytes(vi)) {
            MYVEC_LOG_ERROR("Vector too short for index (%lu bytes, need %lu)"
                            " (pkid=%lld).", args->lengths[3],
                            MyVectorExpectedVectorBytes(vi), pkid);
            *is_null = 1;
            *error = 1;
            return 0;
        }
        if (vi->isCosineMetric() && isZeroVector((const FP32*)vecval, dims)) {
            MYVEC_LOG_ERROR("Zero-magnitude vector rejected for cosine index"
                            " (pkid=%lld).", pkid);
            *is_null = 1;
            *error = 1;
            return 0;
        }
        vi->insertVector((FP32*)vecval, dims, pkid);
    } else {
        *error = 1;
        return 0;
    }

    return 1; /* 1 row added to index */
}

/* UDF : myvector_search_add_row_udf() */
PLUGIN_EXPORT void myvector_search_add_row_udf_deinit(UDF_INIT* initid) {
    /* build/rebuild/refresh complete - persist the index if it is capable */
    AbstractVectorIndex* vi = (AbstractVectorIndex*)(initid->ptr);

    if (vi) {
        MYVEC_LOG_INFO("Not Saving index %s to disk", vi->getName().c_str());

        // vi->saveIndex(myvector_index_dir);

        MYVEC_LOG_INFO("Not Saving index %s to disk completed",
                       vi->getName().c_str());
    }

    return;
}

PLUGIN_EXPORT bool myvector_is_valid_init(UDF_INIT* initid,
                                          UDF_ARGS* args,
                                          char* message) {
    if (!initid || args->arg_count != 2) {
        strcpy(message,
               "Incorrect arguments to myvector_is_valid(), "
               "Usage : myvector_is_valid(<vector>,<dim>)");
        return true;
    }
    return false;
}

/* myvector_is_valid : Verify vector checksum and dimension */
PLUGIN_EXPORT long long myvector_is_valid(UDF_INIT*,
                                          UDF_ARGS* args,
                                          unsigned char*,
                                          unsigned char*) {
    long long dim = *((long long*)args->args[1]);

    /* A NULL vector OR a zero-dimension vector is not valid */
    if (!args->args[0] ||
        args->lengths[0] < (MYVECTOR_COLUMN_EXTRA_LEN + sizeof(FP32))) {
        return 0;
    }

    int vdim = MyVectorDimFromStorageLength(args->lengths[0]);

    if (vdim != dim) {
        return 0;
    }

    ha_checksum cksum1;
    char* raw = args->args[0];

    /* Checksum is at the end */
    memcpy((char*)&cksum1,
           &(raw[args->lengths[0] - sizeof(ha_checksum)]),
           sizeof(ha_checksum));
    ha_checksum cksum2 = my_checksum(
        0, (const unsigned char*)raw, args->lengths[0] - sizeof(ha_checksum));
    if (cksum1 != cksum2) {
        MYVEC_LOG_DEBUG(
            "myvector_is_valid checksum failure (%u != %u)", cksum1, cksum2);
        return 0;
    }

    return 1;  // success
}

PLUGIN_EXPORT void myvector_is_valid_deinit(UDF_INIT*) {}

PLUGIN_EXPORT bool myvector_row_distance_init(UDF_INIT* initid,
                                              UDF_ARGS* args,
                                              char* message) {
    if (!initid || args->arg_count != 1) {
        strcpy(message,
               "Incorrect arguments to myvector_row_distance(), "
               "Usage : myvector_row_distance(idval)");
        return true;
    }
    initid->const_item = 0;
    initid->decimals = 20; /* For some reason, this is explicitly needed here */
    return false;
}

PLUGIN_EXPORT double myvector_row_distance(UDF_INIT*,
                                           UDF_ARGS* args,
                                           char*,
                                           char*) {
    double dist = 99999999999.99;
    KeyTypeInteger idval = *((KeyTypeInteger*)args->args[0]);

    if (tls_distances->size()) {
        if (tls_distances->find(idval) != tls_distances->end())
            dist = (*tls_distances)[idval];
    }

    return dist;
}
PLUGIN_EXPORT void myvector_row_distance_deinit(UDF_INIT*) {}

bool isAfter(const string& binlogfile2,
             const size_t binlogpos2,
             const string& binlogfile1,
             const size_t binlogpos1) {
    return ((binlogfile2 == binlogfile1 && binlogpos2 > binlogpos1) ||
            (binlogfile2 > binlogfile1));
}

void myvector_table_op(const string& dbname,
                       const string& tbname,
                       const string& cname,
                       KeyTypeInteger pkid,
                       vector<unsigned char>& vec,
                       const string& binlogfile,
                       const size_t& binlogpos,
                       bool isDelete) {
    string vecid = dbname + "." + tbname + "." + cname;
    AbstractVectorIndex* vi = g_indexes.get(vecid);

    if (vi) {
        SharedLockGuard l(vi);
        string binlogfileold;
        size_t binlogposold;

        vi->getLastUpdateCoordinates(binlogfileold, binlogposold);
        if (isAfter(binlogfile, binlogpos, binlogfileold, binlogposold)) {
            if (isDelete) {
                vi->deleteVector(pkid);  /// false if the key was not indexed
            } else if (vec.size() < MyVectorExpectedVectorBytes(vi)) {
                /* A plain INSERT/UPDATE with ordinary privilege can write any
                 * byte length to the tracked column; insertVector() has no way
                 * to see vec's real size from the raw pointer it receives, so
                 * this is the only place that can catch an undersized vector
                 * before it causes a fixed-stride OOB read (audit run-1). */
                MYVEC_LOG_WARN("Skipping undersized vector (%zu bytes, need %lu)"
                               " for pkid=%zu, row is in table but not in index.",
                               vec.size(), MyVectorExpectedVectorBytes(vi),
                               (size_t)pkid);
            } else if (vi->isCosineMetric() &&
                isZeroVector(reinterpret_cast<const FP32*>(vec.data()),
                             vi->getDimension())) {
                MYVEC_LOG_WARN("Skipping zero-magnitude vector for cosine index"
                               " (pkid=%zu), row is in table but not in index.",
                               (size_t)pkid);
            } else {
                vi->insertVector(vec.data(), vi->getDimension(), pkid);
            }
        } else {
            MYVEC_LOG_DEBUG("Skipping index update (%s %lu) < (%s %lu).",
                        binlogfile.c_str(),
                        binlogpos,
                        binlogfileold.c_str(),
                        binlogposold);
        }
    }
}

/* Row events skipped by the binlog listener (#205): counted per table and
 * reason, and each (table, reason) logged once, so a busy table cannot flood
 * the error log. Counts live in memory and reset when the server restarts. */
namespace {
struct OnlineSkipStats {
    std::map<OnlineSkipReason, unsigned long long> counts;  // since server start
    std::set<OnlineSkipReason> warned;  // since the index was last built or loaded
};
std::mutex g_online_skip_mutex;
std::map<string, OnlineSkipStats> g_online_skips;  // key: db.table

const char* onlineSkipReasonText(OnlineSkipReason why) {
    switch (why) {
        case OnlineSkipReason::kKeyType:
            return "key column is not an integer";
        case OnlineSkipReason::kRowImage:
            return "binlog_row_image is not FULL";
        case OnlineSkipReason::kColumnCount:
            return "column count does not match the table";
        case OnlineSkipReason::kColumnType:
            return "unreadable column type or truncated event";
    }
    return "unknown";
}
}  // namespace

void myvector_online_skip(const string& db,
                          const string& table,
                          OnlineSkipReason why,
                          const string& detail) {
    bool first;
    {
        std::lock_guard<std::mutex> l(g_online_skip_mutex);
        OnlineSkipStats& st = g_online_skips[db + "." + table];
        st.counts[why]++;
        first = st.warned.insert(why).second;
    }
    if (first)
        MYVEC_LOG_WARN("Online updates for %s.%s: skipping row events (%s%s%s). "
                       "Their changes do not reach the vector index; "
                       "MYVECTOR_INDEX_STATUS counts them.",
                       db.c_str(), table.c_str(), onlineSkipReasonText(why),
                       detail.empty() ? "" : ": ", detail.c_str());
}

void myvector_online_skip_rearm(const string& dbtable) {
    std::lock_guard<std::mutex> l(g_online_skip_mutex);
    auto it = g_online_skips.find(dbtable);
    if (it != g_online_skips.end())
        it->second.warned.clear();
}

string myvector_online_skip_status(const string& dbtable) {
    std::lock_guard<std::mutex> l(g_online_skip_mutex);
    auto it = g_online_skips.find(dbtable);
    if (it == g_online_skips.end())
        return "";
    unsigned long long total = 0;
    std::stringstream reasons;
    for (const auto& rc : it->second.counts) {
        reasons << (total ? ", " : "") << onlineSkipReasonText(rc.first) << ": "
                << rc.second;
        total += rc.second;
    }
    std::stringstream ss;
    ss << "Online events skipped : " << total << " (" << reasons.str() << ")" << endl;
    return ss.str();
}

/* myvector_checkpoint_index() - Incrementally persist a vector index. Check
 * hnswdisk.i for implementation details. This routine is called from the
 * binlog event listener thread at every binlog file rotation. The frequency
 * can be changed in future if needed.
 */
void myvector_checkpoint_index(const string& dbtable,
                               const string& veccol,
                               const string& binlogFile,
                               size_t binlogPos) {
    string vecid = dbtable + "." + veccol;
    AbstractVectorIndex* vi = g_indexes.get(vecid);

    if (vi) {
        SharedLockGuard l(vi);
        string binlogfileold;
        size_t binlogposold;

        vi->getLastUpdateCoordinates(binlogfileold, binlogposold);
        MYVEC_LOG_DEBUG("Checkpoint index %s at (%s %lu)\n",
                    vecid.c_str(),
                    binlogFile.c_str(),
                    binlogPos);
        if (isAfter(binlogFile, binlogPos, binlogfileold, binlogposold)) {
            vi->setLastUpdateCoordinates(binlogFile, binlogPos);
            if (!vi->saveIndex(myvector_index_dir, "checkpoint")) {
                // The tracked position was already advanced above so the checkpoint ID
                // embedded in this save attempt was correct; on failure roll it back so
                // isAfter() and the next checkpoint retry from the last position that
                // was actually persisted, instead of silently skipping it forever.
                MYVEC_LOG_ERROR(
                    "Checkpoint save failed for %s at (%s %lu); binlog position not "
                    "persisted, will retry",
                    vecid.c_str(), binlogFile.c_str(), binlogPos);
                vi->setLastUpdateCoordinates(binlogfileold, binlogposold);
            }
        }
    }
}

string myvector_find_earliest_binlog_file() {
    return g_indexes.FindEarliestBinlogFile();
}

bool myvector_find_earliest_checkpoint(string& binlogfile, size_t& binlogpos) {
    return g_indexes.FindEarliestCheckpoint(binlogfile, binlogpos);
}
/* end of myvector.cc */

#if defined(__GNUC__) || defined(__clang__)
#pragma GCC diagnostic pop
#endif