Iscp integration, hnsw update improvement

[cpegeric]

User description

What type of PR is this?

• API-change
• BUG
• Improvement
• Documentation
• Feature
• Test and CI
• Code Refactoring

Which issue(s) this PR fixes:

issue ##21835

What this PR does / why we need it:

1. hnsw update improvement - hnsw update will call hnsw APIs instead of running SQL. With HNSW APIs, we can
   load the models once and perform all batches from DataRetriever and finally save model files into database.
   This changes save a lot of time for upload/download the model every time there is a 8192 vector block.
2. Common interface SqlProcess created to allow vector index and fulltext APIs in both frontend (process.Process) and background mode (when process.Process is absent)
3. iscp integration

---

PR Type

Enhancement, Tests

---

Description

• ISCP Integration: Implemented comprehensive Index Sync Consumer Producer (ISCP) system for asynchronous index management with CDC (Change Data Capture) support
• HNSW Update Improvements: Refactored HNSW synchronization from function-based to struct-based architecture with HnswSync for better modularity and async operations
• SQL Process Abstraction: Introduced SqlProcess abstraction layer replacing process.Process across vector index operations for background SQL execution
• Async Index Support: Added full async support for HNSW, IVF, and fulltext indexes with CDC task creation and management
• DDL Integration: Enhanced DDL operations (CREATE, ALTER, DROP) with ISCP job registration and cleanup for vector and fulltext indexes
• Performance Optimizations: Replaced JSON library with sonic for better performance and added thread management to HNSW models
• Comprehensive Testing: Added extensive test coverage for async index operations including HNSW f64, IVF, and fulltext indexes
• Transaction Management: Refactored consumers to use direct transaction management with proper timeout configurations

---

Diagram Walkthrough

flowchart LR
  A["Old Process Interface"] --> B["SqlProcess Abstraction"]
  C["Function-based HNSW Sync"] --> D["HnswSync Struct"]
  E["Synchronous Indexes"] --> F["ISCP Async Framework"]
  F --> G["CDC Task Management"]
  F --> H["Background SQL Execution"]
  D --> I["JSON-based CDC Data"]
  B --> J["Vector Index Operations"]
  G --> K["HNSW/IVF/Fulltext Indexes"]

Loading

File Walkthrough

	Relevant files
Code refactoring	2 files sync.go Refactor HNSW CDC sync to struct-based async architecture pkg/vectorindex/hnsw/sync.go • Refactored CdcSync function into a struct-based approach with HnswSync struct • Separated CDC operations into Update, Save, and RunOnce methods for better modularity • Added DownloadAll method for pre-downloading index models • Changed function signatures to use sqlexec.SqlProcess instead of process.Process +119/-74 mock_consumer.go Refactor ISCP consumer for direct transaction management  pkg/iscp/mock_consumer.go • Updated internal SQL consumer to use direct transaction management • Added engine and transaction client dependencies for background operations • Modified consume methods to work with client.TxnOperator instead of executor.TxnExecutor • Enhanced error handling and transaction lifecycle management +53/-29
Tests	23 files sync_test.go Update HNSW sync tests for new struct-based API                    pkg/vectorindex/hnsw/sync_test.go • Updated all test functions to use new HnswSync struct API • Changed mock functions to accept sqlexec.SqlProcess instead of process.Process • Added new test for continuous update operations with small capacity • Modified test setup to create SqlProcess instances +136/-27 index_consumer_test.go Enhance ISCP consumer tests with IVF index support              pkg/iscp/index_consumer_test.go • Added support for IVF index testing with new table definitions and consumer info • Replaced mock SQL executor with stub-based approach using ExecWithResult • Added comprehensive tests for IVF snapshot and tail operations • Updated HNSW consumer tests to use new SQL writer approach +158/-106 cache_test.go Update vector index cache tests for SQL process abstraction pkg/vectorindex/cache/cache_test.go • Updated all cache test methods to use sqlexec.SqlProcess instead of process.Process • Modified mock search implementations to accept new SQL process parameter • Added SqlProcess creation in all test functions +24/-17  model_test.go HNSW model tests updated for SqlProcess interface                pkg/vectorindex/hnsw/model_test.go • Updated test functions to use sqlexec.SqlProcess instead of process.Process • Modified mock functions to accept SqlProcess parameter • Updated all LoadMetadata, LoadIndex, and LoadIndexFromBuffer calls to use new interface +22/-18  iscp_util_test.go ISCP utility functions test suite implementation                  pkg/sql/compile/iscp_util_test.go • Added comprehensive test suite for ISCP utility functions • Implemented mock functions for testing error scenarios in job registration/unregistration • Added tests for index CDC validation, task creation, and deletion operations +301/-0  search_test.go IVF-flat search tests updated for SqlProcess interface      pkg/vectorindex/ivfflat/search_test.go • Updated test mock functions to use sqlexec.SqlProcess instead of process.Process • Modified test setup to create SqlProcess wrapper around process.Process • Updated all search function calls to use new interface +12/-7    search_test.go HNSW search tests updated for SqlProcess interface              pkg/vectorindex/hnsw/search_test.go • Updated mock functions to use sqlexec.SqlProcess parameter • Modified test setup to create SqlProcess wrapper • Updated cache search calls to use new interface +15/-10  build_test.go HNSW build tests updated for SqlProcess interface                pkg/vectorindex/hnsw/build_test.go • Updated test functions to use sqlexec.SqlProcess wrapper • Modified NewHnswBuild calls to use new interface +5/-2      ivf_search_test.go IVF search table function tests updated                                    pkg/sql/colexec/table_function/ivf_search_test.go • Updated mock functions and test interfaces to use sqlexec.SqlProcess • Modified mock search implementation to work with new interface +6/-5      build_dml_util_test.go DML utility async index tests added                                            pkg/sql/plan/build_dml_util_test.go • Added test cases for async fulltext index functions • Implemented tests for invalid JSON and async flag validation +49/-0    fulltext_test.go Fulltext table function tests updated                                        pkg/sql/colexec/table_function/fulltext_test.go • Updated mock functions to use sqlexec.SqlProcess parameter • Modified test setup to work with new interface +5/-3      hnsw_search_test.go HNSW search table function tests updated                                  pkg/sql/colexec/table_function/hnsw_search_test.go • Updated mock search implementation to use sqlexec.SqlProcess • Modified test interfaces for new SqlProcess parameter +3/-2      hnsw_create_test.go HNSW create table function tests updated                                  pkg/sql/colexec/table_function/hnsw_create_test.go • Updated mock function to use sqlexec.SqlProcess parameter • Modified test setup for new interface +3/-2      sqlexec_test.go SQL execution tests updated for SqlProcess                              pkg/vectorindex/sqlexec/sqlexec_test.go • Updated tests to use NewSqlProcess wrapper around process • Modified RunTxn calls to use SqlProcess interface +5/-2      types_test.go Vector index types test updated for capacity parameter      pkg/vectorindex/types_test.go • Updated test to pass capacity parameter to NewVectorIndexCdc +1/-1      vector_hnsw_f64_async.sql Async HNSW vector index integration tests                                test/distributed/cases/vector/vector_hnsw_f64_async.sql • Added comprehensive test cases for async HNSW index functionality • Includes tests for insert, update, delete operations with CDC synchronization • Tests vector similarity search after async index updates +96/-0    fulltext_async.result Async fulltext index test results                                                test/distributed/cases/fulltext/fulltext_async.result • Added expected results for async fulltext index tests • Shows successful async index creation and search functionality +23/-0    fulltext_async.sql Async fulltext index integration tests                                      test/distributed/cases/fulltext/fulltext_async.sql • Added test cases for async fulltext index functionality • Tests index creation, data insertion, search, and table operations +25/-0    vector_ivf_async.result New test results for asynchronous IVF vector indexing        test/distributed/cases/vector/vector_ivf_async.result • Added new test result file for asynchronous IVF (Inverted File) vector index functionality • Contains test results for creating IVF indexes with ASYNC keyword and vector similarity queries • Includes tests for experimental_ivf_index setting, table creation, index creation, and L2 distance queries • Tests both small datasets and larger datasets loaded from CSV files with sleep delays for async operations +61/-0    vector_hnsw.result Updated HNSW test results for async index operations          test/distributed/cases/vector/vector_hnsw.result • Added sleep(30) call and corresponding result to allow time for asynchronous index building • Updated query results to show proper vector similarity search results after async index completion • Added drop table vector_index_01 statement and minor precision changes in cosine distance values +10/-2    vector_ivf_async.sql New test cases for asynchronous IVF vector indexing            test/distributed/cases/vector/vector_ivf_async.sql • New test file for asynchronous IVF vector index functionality • Tests experimental_ivf_index setting, table creation with vector columns, and async index creation • Includes vector similarity queries using L2_DISTANCE function with sleep delays for async operations • Tests both small manual datasets and larger CSV file imports with async index building +60/-0    vector_hnsw_f64_async.result New test results for asynchronous HNSW f64 vector indexing test/distributed/cases/vector/vector_hnsw_f64_async.result • Added new test result file for asynchronous HNSW vector indexing with 64-bit float vectors • Contains test results for CRUD operations (insert, update, delete) with async HNSW indexes • Includes vector similarity search results using L2_DISTANCE with sleep delays for async operations • Tests both small datasets and larger CSV file imports with async index building +66/-0    vector_hnsw.sql Updated HNSW test cases for async index operations              test/distributed/cases/vector/vector_hnsw.sql • Added sleep(30) call to allow time for asynchronous index building before queries • Updated comment from "no result found" to "async index update so result found" • Added drop table vector_index_01 statement for proper test cleanup +6/-1
Enhancement	26 files alter.go Integrate ISCP job management in table alteration workflow pkg/sql/compile/alter.go • Added ISCP job management during table alteration operations • Implemented index CDC task creation for unaffected indexes during copy operations • Enhanced clone logic to handle index table information with unique and algorithm details • Added proper cleanup of ISCP jobs for temporary tables +103/-47 model.go Enhance HNSW model with thread management and SQL process support pkg/vectorindex/hnsw/model.go • Added NThread field to HnswModel struct for thread management • Refactored index initialization into separate initIndex method • Enhanced error handling in index creation and loading operations • Updated method signatures to use sqlexec.SqlProcess instead of process.Process +70/-26  sqlexec.go Add SQL execution abstraction for background operations    pkg/vectorindex/sqlexec/sqlexec.go • Introduced SqlContext and SqlProcess abstractions for background SQL execution • Added support for running SQL operations without frontend process context • Implemented RunTxnWithSqlContext for background transaction management • Enhanced SQL execution functions to work with both frontend and background contexts +242/-62 ddl.go Integrate ISCP job management across DDL operations            pkg/sql/compile/ddl.go • Added ISCP job registration and management for various DDL operations • Implemented index CDC task creation for vector and fulltext indexes • Added cleanup of ISCP jobs during database and table drops • Enhanced IVF index handling with async support and ISCP integration +95/-9    index_consumer.go ISCP index consumer refactoring with async HNSW support    pkg/iscp/index_consumer.go • Refactored IndexConsumer to use engine.Engine and client.TxnClient instead of executor.SQLExecutor • Added separate execution paths for HNSW and other index types with runHnsw and runIndex functions • Implemented transaction management using sqlexec.RunTxnWithSqlContext with different timeout configurations • Added support for JSON-based CDC data processing for HNSW indexes using sonic.Unmarshal +171/-55 search.go IVF-flat search interface updated to SqlProcess                    pkg/vectorindex/ivfflat/search.go • Replaced process.Process with sqlexec.SqlProcess throughout the search implementation • Updated context handling to use sqlproc.GetContext() instead of proc.Ctx • Modified function signatures for LoadIndex, Search, findCentroids, and searchEntries methods +22/-19  build_dml_util.go DML utility functions enhanced with async index support    pkg/sql/plan/build_dml_util.go • Added async index checks using catalog.IsIndexAsync() to skip synchronous processing • Updated comment formatting for better readability in delete plans documentation • Added early returns for async indexes in multiple index building functions +65/-18  search.go HNSW search interface updated to SqlProcess                            pkg/vectorindex/hnsw/search.go • Replaced process.Process with sqlexec.SqlProcess in search interface • Updated LoadMetadata, LoadIndex, and Search method signatures • Modified context handling to use sqlproc.GetContext() +9/-10    cache.go Vector index cache updated for SqlProcess interface            pkg/vectorindex/cache/cache.go • Updated VectorIndexSearchIf interface to use sqlexec.SqlProcess instead of process.Process • Modified Load and Search method signatures throughout the cache implementation • Updated all cache operations to work with new SqlProcess interface +10/-10  ddl_index_algo.go DDL index algorithms enhanced with async support                  pkg/sql/compile/ddl_index_algo.go • Added async index support for fulltext and HNSW indexes • Implemented CDC task creation for async indexes instead of direct SQL execution • Added conditional logic to handle both sync and async index creation workflows +60/-7    index_sqlwriter.go HNSW SQL writer refactored for JSON-based CDC                        pkg/iscp/index_sqlwriter.go • Modified HNSW SQL writer to return JSON data instead of SQL statements • Added NewSync method to create HnswSync instances • Updated CDC writer capacity configuration +15/-4    ivf_search.go IVF search table function updated for SqlProcess                  pkg/sql/colexec/table_function/ivf_search.go • Updated IVF search table function to use sqlexec.NewSqlProcess(proc) • Modified getVersion and cache search calls to use SqlProcess interface +3/-2      build.go HNSW build interface updated to SqlProcess                              pkg/vectorindex/hnsw/build.go • Updated NewHnswBuild and related methods to use sqlexec.SqlProcess • Modified context handling to use sqlproc.GetContext() +7/-6      hnsw_search.go HNSW search table function updated for SqlProcess                pkg/sql/colexec/table_function/hnsw_search.go • Updated HNSW search table function to use sqlexec.NewSqlProcess(proc) • Modified cache search call to use SqlProcess interface +2/-1      data_retriever.go Data retriever watermark update refactored                              pkg/iscp/data_retriever.go • Updated UpdateWatermark method to use direct SQL execution with transaction • Replaced executor interface with context, cnUUID, and txn parameters • Added proper timeout and system account context handling +18/-4    types.go Vector index types updated with sonic JSON library              pkg/vectorindex/types.go • Replaced json package with sonic for better performance • Updated NewVectorIndexCdc to accept capacity parameter • Modified JSON marshaling to use sonic library +4/-4      hnsw_create.go HNSW create table function updated for SqlProcess                pkg/sql/colexec/table_function/hnsw_create.go • Updated HNSW creation functions to use sqlexec.NewSqlProcess(proc) • Modified NewHnswBuild calls to use SqlProcess interface +3/-3      consumer.go Consumer factory updated with engine and transaction client pkg/iscp/consumer.go • Updated NewConsumer function to accept engine.Engine and client.TxnClient parameters • Modified consumer creation calls to pass additional parameters +9/-3      sql.go IVF-flat SQL utilities updated for SqlProcess                        pkg/vectorindex/ivfflat/sql.go • Updated GetVersion function to use sqlexec.SqlProcess instead of process.Process • Modified context handling and error messages +4/-4      types.go ISCP types interface updated for direct transaction handling pkg/iscp/types.go • Updated DataRetriever interface to change UpdateWatermark method signature • Removed executor dependency in favor of direct context and transaction parameters +1/-2      metadata_scan.go Metadata scan updated for SqlProcess interface                      pkg/sql/colexec/table_function/metadata_scan.go • Updated metadata scan to use sqlexec.NewSqlProcess(proc) for SQL execution • Modified RunSql call to use SqlProcess interface +4/-2      fulltext.go Fulltext table function updated for SqlProcess                      pkg/sql/colexec/table_function/fulltext.go • Updated fulltext functions to use sqlexec.NewSqlProcess(proc) • Modified SQL execution calls to use SqlProcess interface +2/-2      ivf_create.go IVF create table function updated for SqlProcess                  pkg/sql/colexec/table_function/ivf_create.go • Updated IVF creation functions to use sqlexec.NewSqlProcess(proc) • Modified version retrieval and SQL execution calls +2/-2      iteration.go ISCP iteration updated for new consumer interface                pkg/iscp/iteration.go • Updated NewConsumer call to include cnEngine and cnTxnClient parameters +1/-1      watermark_updater.go Watermark updater function made mockable                                  pkg/iscp/watermark_updater.go • Made ExecWithResult function a variable for easier testing/mocking +1/-1      util.go Compile utility function error handling improved                  pkg/sql/compile/util.go • Updated genInsertIndexTableSqlForFullTextIndex to return error alongside SQL strings • Added error handling to function signature +2/-2
Miscellaneous	2 files function_id_test.go Remove HNSW CDC update function ID                                              pkg/sql/plan/function/function_id_test.go • Removed HNSW_CDC_UPDATE function ID from predefined function IDs • Updated FUNCTION_END_NUMBER to reflect the removal +1/-3      function_id.go HNSW CDC update function ID removed                                            pkg/sql/plan/function/function_id.go • Removed HNSW_CDC_UPDATE function ID constant • Decremented FUNCTION_END_NUMBER accordingly +1/-7
Feature	1 files iscp_util.go ISCP utility functions for index CDC management                    pkg/sql/compile/iscp_util.go • Implemented ISCP (Index Sync Consumer Producer) utility functions for CDC task management • Added functions for creating, deleting, and validating index CDC tasks • Implemented job registration/unregistration with proper error handling • Added support for different sinker types based on index algorithms +208/-0
Additional files	5 files func_hnsw.go +0/-106  func_hnsw_test.go +0/-191  list_builtIn.go +0/-21    vector_hnsw_async.result +66/-0    vector_hnsw_async.sql +96/-0

---

[qodo-merge-pro]

PR Compliance Guide 🔍

Below is a summary of compliance checks for this PR:

Security Compliance
🟢	No security concerns identified No security vulnerabilities detected by AI analysis. Human verification advised for critical code.
Ticket Compliance
🟡	🎫 #21835 🟢 Provide a mechanism to update HNSW indexes via CDC on INSERT, DELETE, UPDATE. Establish a general async CDC-based framework applicable to vector indexes (HNSW, IVF) and fulltext. Integrate CDC-driven index maintenance with DDL operations (CREATE, ALTER, DROP), including job registration and cleanup. Support asynchronous/background execution context for index updates separate from foreground SQL processing. Include tests validating async update flows for HNSW, IVF, and fulltext indices. Ensure transactionally safe application of CDC updates to index metadata/tables. ⚪ End-to-end verification in a real cluster that CDC events are produced/consumed correctly across failures and restarts. Performance validation of async updates under high write throughput and large indexes. Backward compatibility and migration behavior for existing indexes without CDC.
Codebase Duplication Compliance
⚪	Codebase context is not defined Follow the guide to enable codebase context checks.
Custom Compliance
⚪	No custom compliance provided Follow the guide to enable custom compliance check.

Compliance status legend

🟢 - Fully Compliant
🟡 - Partial Compliant
🔴 - Not Compliant
⚪ - Requires Further Human Verification
🏷️ - Compliance label

[qodo-merge-pro]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Fix resource leak on error In NewHnswSync, call sync.Destroy() before returning an error from sync.DownloadAll to prevent a resource leak. pkg/vectorindex/hnsw/sync.go [225-231] // save all model to local by LoadIndex and Unload err = sync.DownloadAll(sqlproc) if err != nil { + sync.Destroy() return nil, err } return sync, nil Apply / Chat Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a resource leak where sync.Destroy() is not called on an error path, which would prevent the cleanup of potentially significant resources.	Medium
	Fix resource leak in initialization In initIndex, replace explicit return err with return to allow the deferred cleanup function to execute correctly on error, preventing a resource leak. pkg/vectorindex/hnsw/model.go [81-112] func (idx *HnswModel[T]) initIndex(cfg vectorindex.IndexConfig) (err error) { idx.Index, err = usearch.NewIndex(cfg.Usearch) if err != nil { return err } defer func() { if err != nil { if idx.Index != nil { idx.Index.Destroy() idx.Index = nil } } }() err = idx.Index.Reserve(idx.MaxCapacity) if err != nil { - return err + return } err = idx.Index.ChangeThreadsAdd(idx.NThread) if err != nil { - return err + return } err = idx.Index.ChangeThreadsSearch(idx.NThread) if err != nil { - return err + return } return nil } Apply / Chat Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a resource leak due to early returns that bypass the deferred cleanup logic, and the proposed fix is the standard Go idiom for this pattern.	Medium
	Fix incorrect transaction handling logic Fix a bug in the deferred transaction handling logic to prevent losing the original error upon a successful rollback. The proposed change ensures proper error logging and propagation for both commit and rollback operations. pkg/iscp/mock_consumer.go [248-258] defer func() { if err != nil { - err = txn.Rollback(ctx) - } else { - err = txn.Commit(ctx) + if rollbackErr := txn.Rollback(ctx); rollbackErr != nil { + logutil.Error("InteralSqlConsumer Consume: failed to rollback transaction", + zap.Error(err), + zap.NamedError("rollbackErr", rollbackErr)) + } + return } - - if err != nil { - logutil.Error("InteralSqlConsumer Consume Tail failed") + if commitErr := txn.Commit(ctx); commitErr != nil { + logutil.Error("InteralSqlConsumer Consume: failed to commit transaction", zap.Error(commitErr)) + err = commitErr } }() Apply / Chat Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a significant bug in the deferred transaction handling logic where an original error could be lost if the rollback succeeds, leading to silent failures. The proposed fix implements a much more robust pattern for handling transaction commit/rollback and error propagation.	Medium
	Use correct table definition for cleanup In AlterTableCopy, use qry.CopyTableDef instead of newRel.CopyTableDef when calling DropAllIndexCdcTasks to ensure correct cleanup. pkg/sql/compile/alter.go [165-171] //5. ISCP: temp table already created pitr and iscp job with temp table name // and we don't want iscp to run with temp table so drop pitr and iscp job with the temp table here - newTmpTableDef := newRel.CopyTableDef(c.proc.Ctx) - err = DropAllIndexCdcTasks(c, newTmpTableDef, dbName, qry.CopyTableDef.Name) + err = DropAllIndexCdcTasks(c, qry.CopyTableDef, dbName, qry.CopyTableDef.Name) if err != nil { return err } Apply / Chat Suggestion importance[1-10]: 7 __ Why: The suggestion correctly points out that using qry.CopyTableDef is more robust for cleaning up CDC tasks, as it contains the original definition of the temporary table.	Medium
	Replace fixed sleep with polling Replace the fixed sleep(30) with a polling mechanism to reliably wait for asynchronous index creation to complete, preventing potential test flakiness. test/distributed/cases/vector/vector_ivf_async.sql [13-17] create index idx01 using ivfflat on vector_index_01(b) lists=5 op_type "vector_l2_ops" ASYNC; show create table vector_index_01; desc vector_index_01; -select sleep(30); +-- Replace sleep with a robust polling mechanism. For example (syntax is conceptual): +-- DO +-- SELECT sleep(1); +-- WHILE (SELECT status FROM information_schema.indexes WHERE table_name = 'vector_index_01' AND index_name = 'idx01') != 'built'; select * from vector_index_01 order by L2_DISTANCE(b, "[16, 15, 0, 0, 5, 46, 5, 5, 4, 0, 0, 0, 28, 118, 12, 5, 75, 44, 5, 0, 6, 32, 6, 49, 41, 74, 9, 1, 0, 0, 0, 9, 1, 9, 16, 41, 71, 80, 3, 0, 0, 4, 3, 5, 51, 106, 11, 3, 112, 28, 13, 1, 4, 8, 3, 104, 118, 14, 1, 1, 0, 0, 0, 88, 3, 27, 46, 118, 108, 49, 2, 0, 1, 46, 118, 118, 27, 12, 0, 0, 33, 118, 118, 8, 0, 0, 0, 4, 118, 95, 40, 0, 0, 0, 1, 11, 27, 38, 12, 12, 18, 29, 3, 2, 13, 30, 94, 78, 30, 19, 9, 3, 31, 45, 70, 42, 15, 1, 3, 12, 14, 22, 16, 2, 3, 17, 24, 13]") ASC LIMIT 2; [To ensure code accuracy, apply this suggestion manually] Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that using a fixed sleep can cause test flakiness and proposes a more robust polling mechanism, which would improve test reliability and efficiency.	Medium
More