Repository: apache/flink
File: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecVectorSearchTableFunction.java (Java)

Code before:
        this.vectorSearchSpec = vectorSearchSpec;
        this.asyncOptions = asyncOptions;
    }

    @Override
    protected Transformation<RowData> translateToPlanInternal(
            PlannerBase planner, ExecNodeConfig config) {
        // 1. translate input node
        ExecEdge inputEdge = getInputEdges().get(0);
        Transformation<RowData> inputTransformation =
                (Transformation<RowData>) inputEdge.translateToPlan(planner);
        // 2. extract search function
        TableSourceTable searchTable = vectorSearchTableSourceSpec.getSearchTable();
        boolean isAsyncEnabled = asyncOptions != null;
        UserDefinedFunction vectorSearchFunction =
                findVectorSearchFunction(
                        VectorSearchUtil.createVectorSearchRuntimeProvider(
                                searchTable, vectorSearchSpec.getSearchColumns().keySet()),
                        isAsyncEnabled);
        UserDefinedFunctionHelper.prepareInstance(config, vectorSearchFunction);
        // 3. build the operator
        RowType inputType = (RowType) inputEdge.getOutputType();
        RowType outputType = (RowType) getOutputType();
        StreamOperatorFactory<RowData> operatorFactory =
                isAsyncEnabled
                        ? createAsyncVectorSearchOperator()
                        : createSyncVectorSearchOperator(
                                searchTable,
                                config,
                                planner.getFlinkContext().getClassLoader(),
                                (VectorSearchFunction) vectorSearchFunction,
                                ShortcutUtils.unwrapContext(planner.getFlinkContext())
                                        .getCatalogManager()
                                        .getDataTypeFactory(),
                                inputType,
                                vectorSearchSpec.getOutputType(),
                                outputType);
        return ExecNodeUtil.createOneInputTransformation(
                inputTransformation,
                createTransformationMeta(VECTOR_SEARCH_TRANSFORMATION, config),
                operatorFactory,
                InternalTypeInfo.of(outputType),
                inputTransformation.getParallelism(),
                false);
    }

    // ~ Utilities --------------------------------------------------------------

    private UserDefinedFunction findVectorSearchFunction(
            VectorSearchTableSource.VectorSearchRuntimeProvider provider, boolean async) {
        if (async) {

Review comment (suggestion):
I am curious what this is doing , there is one implementation that throws an Exception. I thought this change is adding options for the async search.

Code after:
                ExecNodeContext.newPersistedConfig(
                        StreamExecVectorSearchTableFunction.class, tableConfig),
                Collections.singletonList(inputProperty),
                outputType,
                description);
        this.vectorSearchTableSourceSpec = vectorSearchTableSourceSpec;
        this.vectorSearchSpec = vectorSearchSpec;
        this.asyncOptions = asyncOptions;
    }

    @Override
    protected Transformation<RowData> translateToPlanInternal(
            PlannerBase planner, ExecNodeConfig config) {
        // 1. translate input node
        ExecEdge inputEdge = getInputEdges().get(0);
        Transformation<RowData> inputTransformation =
                (Transformation<RowData>) inputEdge.translateToPlan(planner);
        // 2. extract search function
        TableSourceTable searchTable = vectorSearchTableSourceSpec.getSearchTable();
        boolean isAsyncEnabled = asyncOptions != null;
        UserDefinedFunction vectorSearchFunction =
                findVectorSearchFunction(
                        VectorSearchUtil.createVectorSearchRuntimeProvider(
                                searchTable, vectorSearchSpec.getSearchColumns().keySet()),
                        isAsyncEnabled);
        UserDefinedFunctionHelper.prepareInstance(config, vectorSearchFunction);
        // 3. build the operator
        RowType inputType = (RowType) inputEdge.getOutputType();
        RowType outputType = (RowType) getOutputType();
        DataTypeFactory dataTypeFactory =
                ShortcutUtils.unwrapContext(planner.getFlinkContext())
                        .getCatalogManager()
                        .getDataTypeFactory();
        StreamOperatorFactory<RowData> operatorFactory =
                isAsyncEnabled
                        ? createAsyncVectorSearchOperator(
                                searchTable,
                                config,
                                planner.getFlinkContext().getClassLoader(),
                                (AsyncVectorSearchFunction) vectorSearchFunction,
                                dataTypeFactory,
                                inputType,
                                vectorSearchSpec.getOutputType(),
                                outputType)
                        : createSyncVectorSearchOperator(
                                searchTable,
                                config,
                                planner.getFlinkContext().getClassLoader(),
                                (VectorSearchFunction) vectorSearchFunction,
                                dataTypeFactory,
                                inputType,