VECTOR_SEARCH(
    <SEARCH_TABLE>, 
    <COLUMN_TO_SEARCH>,
    <COLUMN_TO_QUERY>, 
    <TOP_K>[,
    <ON_TIME>,
    <CONFIG>])

-- users schema: id, name, features
-- products schema: id, name, prices, index 
-- output schema: id, name, features, id0, name, prices, index
SELECT * FROM users, LATERAL TABLE (VECTOR_SEARCH(
  TABLE products,
  DESCRIPTOR(`index`),
  users.features,
  3
))

public interface VectorSearchTableSource extends DynamicTableSource {

  VectorSearchRuntimeProvider getSearchRuntimeProvider(VectorSearchContext context);

  interface VectorSearchContext extends DynamicTableSource.Context {
  
     /**
      * Returns an array of key index paths that should be used during the search. The indices
      * are 0-based and support composite keys within (possibly nested) structures.
      *
      * <p>For example, given a table with data type {@code ROW < i INT, s STRING, r ROW < i2
      * INT, s2 STRING > >}, this method would return {@code [[0], [2, 1]]} when {@code i} and
      * {@code s2} are used for performing a lookup.
      *
      * @return array of key index paths
      */
     int[][] getSearchColumns();

     /**
      * Runtime config provided to provider. The config can be used by planner or vector search provider
      * at runtime. For example, async options can be used by planner to choose async inference.
      * Other config such as http timeout or retry can be used to configure search functions.
      */
     ReadableConfig runtimeConfig();

  }
  
  interface VectorSearchRuntimeProvider {  }
}

interface VectorSearchFunctionProvider extends VectorSearchRuntimeProvider {

  VectorSearchFunction createVectorSearchFunction();
  
}

interface VectorSearchFunction extends TableFunction<RowData> {

    /**
     * Using vector, search topK results. The search function should return most similar topK
     * results.
     *
     * <p>Notice: search function should return at most topK results even if multiple results have
     * same scores.
     */
    public abstract Collection<RowData> vectorSearch(int topK, RowData queryData) throws IOException;

    /** Invoke {@link #vectorSearch} and handle exceptions. */
    public final void eval(Object... args) {
        int topK = (int) args[0];
        GenericRowData argsData = GenericRowData.of(args[1]);
        try {
            Collection<RowData> results = vectorSearch(topK, argsData);
            if (results == null) {
                return;
            }
            results.forEach(this::collect);
        } catch  (Exception e) {
            throw new FlinkRuntimeException(
                    String.format("Failed to execute search with input row %s.", argsData), e);
        }
    }
}

interface AsyncVectorSearchFunctionProvider extends VectorSearchRuntimeProvider {

  AsyncVectorSearchFunction createAsyncVectorSearchFunction();  

}

interface AsyncVectorSearchFunction extends AsyncTableFunction {

    /**
     * Using vector, search topK results. The search function should return most similar topK
     * results.
     *
     * <p>Notice: search function should return at most topK results even if multiple results have
     * same scores.
     */
    public abstract CompletableFuture<Collection<RowData>> asyncVectorSearch(int topK, RowData queryData);

    /** Invoke {@link #asyncVectorSearch} and handle exceptions. */
    public final void eval(CompletableFuture<Collection<RowData>> future, Object... args) {
        int topK = (int) args[0];
        GenericRowData argsData = GenericRowData.of(args[1]);
        asyncVectorSearch(topK, argsData)
                .whenComplete(
                        (result, exception) -> {
                            if (exception != null) {
                                future.completeExceptionally(
                                        new TableException(
                                                String.format(
                                                        "Failed to asynchronously search entries with key '%s'",
                                                        args),
                                                exception));
                                return;
                            }
                            future.complete(result);
                        });
    }
}

SELECT * FROM users, LATERAL TABLE (VECTOR_SEARCH(
  TABLE products,
  DESCRIPTOR(`index`),
  users.features,
  3
))

LogicalProject(inputs=[0..5])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
   :- LogicalTableScan(table=[[users]])
   +- LogicalTableFunctionScan(invocation=[VECTOR_SEARCH(TABLE(#0), DESCRIPTOR(_UTF-16LE'index'), $cor0.features, 3)]])
      +- LogicalProject(inputs=[0..2])
         +- LogicalTableScan(table=[[products]])

VectorSearch(joinType=[INNER], searchTable=[products] queryColumn=[2], searchColumn=[2], topK=[3])
+- TableSourceScan(table=[[users]], fields=[id, name, features])

CREATE VIEW p AS SELECT * FROM products WHERE cat = 'toy';

SELECT * FROM users, LATERAL TABLE(VECTOR_SEARCH(
  TABLE p,
  DESCRIPTOR(`index`),
  users.features,
  'cosine',
  3
))

SELECT * FROM users, LATERAL TABLE(VECTOR_SEARCH(
  (SELECT * FROM products WHERE prices < users.threshold),
  DESCRIPTOR(`index`),
  users.features,
  'cosine',
  3
))

SELECT * FROM users, LATERAL TABLE(HYBIR_SEARCH(
  base_table => TABLE products,
  search_column => DESCRIPTOR(`sparse_index`, `dense_index`),
  probe_column => (users.features1,users.features2),
  weight => ARRAY(0.3, 0.7)
  distance_type =>'cosine',
  top_k => 3
)) AS p