Make GDS table function (#5048)

Author: andyfengHKU

extension/fts/src/fts_extension.cpp

@@ -25,7 +25,7 @@ static void initFTSEntries(const transaction::Transaction* transaction, catalog:
 void FTSExtension::load(main::ClientContext* context) {
     auto& db = *context->getDatabase();
     ExtensionUtils::addScalarFunc<StemFunction>(db);
-    ExtensionUtils::addGDSFunc<QueryFTSFunction>(db);
+    ExtensionUtils::addTableFunc<QueryFTSFunction>(db);
     ExtensionUtils::addStandaloneTableFunc<CreateFTSFunction>(db);
     ExtensionUtils::addStandaloneTableFunc<InternalCreateFTSFunction>(db, true /* isInternal */);
     ExtensionUtils::addStandaloneTableFunc<DropFTSFunction>(db);

extension/fts/src/function/query_fts_index.cpp

@@ -10,7 +10,6 @@
 #include "function/gds/gds_utils.h"
 #include "function/query_fts_bind_data.h"
 #include "processor/execution_context.h"
-#include "processor/operator/gds_call_shared_state.h"
 #include "storage/index/index_utils.h"
 #include "storage/storage_manager.h"
 #include "storage/store/node_table.h"
@@ -153,7 +152,7 @@ void QFTSOutputWriter::write(processor::FactorizedTable& scoreFT, nodeID_t docNo
 
 class QFTSVertexCompute final : public VertexCompute {
 public:
-    QFTSVertexCompute(MemoryManager* mm, processor::GDSCallSharedState* sharedState,
+    QFTSVertexCompute(MemoryManager* mm, GDSFuncSharedState* sharedState,
         std::unique_ptr<QFTSOutputWriter> writer)
         : mm{mm}, sharedState{sharedState}, writer{std::move(writer)} {
         localFT = sharedState->factorizedTablePool.claimLocalTable(mm);
@@ -175,18 +174,24 @@ class QFTSVertexCompute final : public VertexCompute {
 
 private:
     MemoryManager* mm;
-    processor::GDSCallSharedState* sharedState;
+    GDSFuncSharedState* sharedState;
     processor::FactorizedTable* localFT;
     std::unique_ptr<QFTSOutputWriter> writer;
 };
 
+static constexpr char SCORE_PROP_NAME[] = "score";
+static constexpr char DOC_FREQUENCY_PROP_NAME[] = "df";
+static constexpr char TERM_FREQUENCY_PROP_NAME[] = "tf";
+static constexpr char DOC_LEN_PROP_NAME[] = "len";
+static constexpr char DOC_ID_PROP_NAME[] = "docID";
+
 static std::unordered_map<offset_t, uint64_t> getDFs(main::ClientContext& context,
     const catalog::NodeTableCatalogEntry& termsEntry, const std::vector<std::string>& terms) {
     auto storageManager = context.getStorageManager();
     auto tableID = termsEntry.getTableID();
     auto& termsNodeTable = storageManager->getTable(tableID)->cast<NodeTable>();
     auto tx = context.getTransaction();
-    auto dfColumnID = termsEntry.getColumnID(QueryFTSAlgorithm::DOC_FREQUENCY_PROP_NAME);
+    auto dfColumnID = termsEntry.getColumnID(DOC_FREQUENCY_PROP_NAME);
     std::vector<LogicalType> vectorTypes;
     vectorTypes.push_back(LogicalType::INTERNAL_ID());
     vectorTypes.push_back(LogicalType::UINT64());
@@ -231,21 +236,22 @@ static uint64_t getSparseFrontierSize(uint64_t numRows) {
     return size;
 }
 
-void QueryFTSAlgorithm::exec(processor::ExecutionContext* executionContext) {
-    auto clientContext = executionContext->clientContext;
+static common::offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) {
+    auto clientContext = input.context->clientContext;
     auto transaction = clientContext->getTransaction();
+    auto sharedState = input.sharedState->ptrCast<GDSFuncSharedState>();
     auto graph = sharedState->graph.get();
     auto graphEntry = graph->getGraphEntry();
-    auto qFTSBindData = bindData->ptrCast<QueryFTSBindData>();
+    auto qFTSBindData = input.bindData->constPtrCast<QueryFTSBindData>();
     auto& termsEntry = graphEntry->nodeInfos[0].entry->constCast<catalog::NodeTableCatalogEntry>();
-    auto terms = bindData->ptrCast<QueryFTSBindData>()->getTerms(*executionContext->clientContext);
-    auto dfs = getDFs(*executionContext->clientContext, termsEntry, terms);
+    auto terms = qFTSBindData->getTerms(*input.context->clientContext);
+    auto dfs = getDFs(*input.context->clientContext, termsEntry, terms);
     // Do edge compute to extend terms -> docs and save the term frequency and document frequency
     // for each term-doc pair. The reason why we store the term frequency and document frequency
     // is that: we need the `len` property from the docs table which is only available during the
     // vertex compute.
-    auto currentFrontier = PathLengths::getUnvisitedFrontier(executionContext, graph);
-    auto nextFrontier = PathLengths::getUnvisitedFrontier(executionContext, graph);
+    auto currentFrontier = PathLengths::getUnvisitedFrontier(input.context, graph);
+    auto nextFrontier = PathLengths::getUnvisitedFrontier(input.context, graph);
     auto frontierPair =
         std::make_unique<DoublePathLengthsFrontierPair>(currentFrontier, nextFrontier);
     auto termsTableID = termsEntry.getTableID();
@@ -263,15 +269,15 @@ void QueryFTSAlgorithm::exec(processor::ExecutionContext* executionContext) {
     auto auxiliaryState = std::make_unique<EmptyGDSAuxiliaryState>();
     auto compState = GDSComputeState(std::move(frontierPair), std::move(edgeCompute),
         std::move(auxiliaryState), nullptr /* outputNodeMask */);
-    GDSUtils::runFrontiersUntilConvergence(executionContext, compState, graph, ExtendDirection::FWD,
+    GDSUtils::runFrontiersUntilConvergence(input.context, compState, graph, ExtendDirection::FWD,
         1 /* maxIters */, TERM_FREQUENCY_PROP_NAME);
 
     // Do vertex compute to calculate the score for doc with the length property.
     auto mm = clientContext->getMemoryManager();
     auto numUniqueTerms = getNumUniqueTerms(terms);
     auto writer = std::make_unique<QFTSOutputWriter>(scores, mm, qFTSBindData->getConfig(),
         *qFTSBindData, numUniqueTerms);
-    auto vc = std::make_unique<QFTSVertexCompute>(mm, sharedState.get(), std::move(writer));
+    auto vc = std::make_unique<QFTSVertexCompute>(mm, sharedState, std::move(writer));
     auto vertexPropertiesToScan = std::vector<std::string>{DOC_LEN_PROP_NAME, DOC_ID_PROP_NAME};
     auto docsEntry = graphEntry->nodeInfos[1].entry;
     auto numDocs = storageManager->getTable(docsEntry->getTableID())->getNumTotalRows(transaction);
@@ -284,65 +290,75 @@ void QueryFTSAlgorithm::exec(processor::ExecutionContext* executionContext) {
             }
         }
     } else {
-        GDSUtils::runVertexCompute(executionContext, graph, *vc, docsEntry, vertexPropertiesToScan);
+        GDSUtils::runVertexCompute(input.context, graph, *vc, docsEntry, vertexPropertiesToScan);
     }
     sharedState->factorizedTablePool.mergeLocalTables();
+    return 0;
 }
 
-expression_vector QueryFTSAlgorithm::getResultColumns(const GDSBindInput& bindInput) const {
-    expression_vector columns;
-    auto& docsNode = bindData->getNodeOutput()->constCast<NodeExpression>();
-    columns.push_back(docsNode.getInternalID());
-    std::string scoreColumnName = SCORE_PROP_NAME;
-    if (!bindInput.yieldVariables.empty()) {
-        scoreColumnName = bindColumnName(bindInput.yieldVariables[1], scoreColumnName);
-    }
-    auto scoreColumn = bindInput.binder->createVariable(scoreColumnName, LogicalType::DOUBLE());
-    columns.push_back(scoreColumn);
-    return columns;
-}
-
-static std::string getParamVal(const GDSBindInput& input, idx_t idx) {
+static std::string getParamVal(const TableFuncBindInput& input, idx_t idx) {
     if (input.getParam(idx)->expressionType != ExpressionType::LITERAL) {
         throw BinderException{"The table and index name must be literal expressions."};
     }
     return ExpressionUtil::getLiteralValue<std::string>(
         input.getParam(idx)->constCast<LiteralExpression>());
 }
 
-void QueryFTSAlgorithm::bind(const GDSBindInput& input, main::ClientContext& context) {
-    context.setUseInternalCatalogEntry(true /* useInternalCatalogEntry */);
+static std::unique_ptr<TableFuncBindData> bindFunc(main::ClientContext* context,
+    const TableFuncBindInput* input) {
+    context->setUseInternalCatalogEntry(true /* useInternalCatalogEntry */);
     // For queryFTS, the table and index name must be given at compile time while the user
     // can give the query at runtime.
-    auto inputTableName = getParamVal(input, 0);
-    auto indexName = getParamVal(input, 1);
-    auto query = input.getParam(2);
+    auto inputTableName = getParamVal(*input, 0);
+    auto indexName = getParamVal(*input, 1);
+    auto query = input->getParam(2);
 
     auto tableEntry =
-        IndexUtils::bindTable(context, inputTableName, indexName, IndexOperation::QUERY);
-    auto ftsIndexEntry = context.getCatalog()->getIndex(context.getTransaction(),
-        tableEntry->getTableID(), indexName);
-    auto entry =
-        context.getCatalog()->getTableCatalogEntry(context.getTransaction(), inputTableName);
-    auto nodeOutput = bindNodeOutput(input, {entry});
-
-    auto termsEntry = context.getCatalog()->getTableCatalogEntry(context.getTransaction(),
+        IndexUtils::bindTable(*context, inputTableName, indexName, IndexOperation::QUERY);
+    auto catalog = context->getCatalog();
+    auto transaction = context->getTransaction();
+    auto ftsIndexEntry = catalog->getIndex(transaction, tableEntry->getTableID(), indexName);
+    auto entry = catalog->getTableCatalogEntry(transaction, inputTableName);
+    auto nodeOutput = GDSFunction::bindNodeOutput(*input, {entry});
+
+    auto termsEntry = catalog->getTableCatalogEntry(transaction,
         FTSUtils::getTermsTableName(tableEntry->getTableID(), indexName));
-    auto docsEntry = context.getCatalog()->getTableCatalogEntry(context.getTransaction(),
+    auto docsEntry = catalog->getTableCatalogEntry(transaction,
         FTSUtils::getDocsTableName(tableEntry->getTableID(), indexName));
-    auto appearsInEntry = context.getCatalog()->getTableCatalogEntry(context.getTransaction(),
+    auto appearsInEntry = catalog->getTableCatalogEntry(transaction,
         FTSUtils::getAppearsInTableName(tableEntry->getTableID(), indexName));
     auto graphEntry = graph::GraphEntry({termsEntry, docsEntry}, {appearsInEntry});
-    bindData = std::make_unique<QueryFTSBindData>(std::move(graphEntry), nodeOutput,
-        std::move(query), *ftsIndexEntry, QueryFTSOptionalParams{input.optionalParams});
-    context.setUseInternalCatalogEntry(false /* useInternalCatalogEntry */);
+
+    expression_vector columns;
+    auto& docsNode = nodeOutput->constCast<NodeExpression>();
+    columns.push_back(docsNode.getInternalID());
+    std::string scoreColumnName = SCORE_PROP_NAME;
+    if (!input->yieldVariables.empty()) {
+        scoreColumnName = GDSFunction::bindColumnName(input->yieldVariables[1], scoreColumnName);
+    }
+    auto scoreColumn = input->binder->createVariable(scoreColumnName, LogicalType::DOUBLE());
+    columns.push_back(scoreColumn);
+    auto bindData =
+        std::make_unique<QueryFTSBindData>(std::move(columns), std::move(graphEntry), nodeOutput,
+            std::move(query), *ftsIndexEntry, QueryFTSOptionalParams{input->optionalParamsLegacy});
+    context->setUseInternalCatalogEntry(false /* useInternalCatalogEntry */);
+    return bindData;
 }
 
 function_set QueryFTSFunction::getFunctionSet() {
     function_set result;
-    auto algo = std::make_unique<QueryFTSAlgorithm>();
-    result.push_back(
-        std::make_unique<GDSFunction>(name, algo->getParameterTypeIDs(), std::move(algo)));
+    // inputs are tableName, indexName, query
+    auto func = std::make_unique<TableFunction>(QueryFTSFunction::name,
+        std::vector<LogicalTypeID>{LogicalTypeID::STRING, LogicalTypeID::STRING,
+            LogicalTypeID::STRING});
+    func->bindFunc = bindFunc;
+    func->tableFunc = tableFunc;
+    func->initSharedStateFunc = GDSFunction::initSharedState;
+    func->initLocalStateFunc = TableFunction::initEmptyLocalState;
+    func->canParallelFunc = [] { return false; };
+    func->getLogicalPlanFunc = GDSFunction::getLogicalPlan;
+    func->getPhysicalPlanFunc = GDSFunction::getPhysicalPlan;
+    result.push_back(std::move(func));
     return result;
 }
 

extension/fts/src/include/function/query_fts_bind_data.h

@@ -24,11 +24,11 @@ struct QueryFTSBindData final : function::GDSBindData {
     QueryFTSOptionalParams optionalParams;
     common::table_id_t outputTableID;
 
-    QueryFTSBindData(graph::GraphEntry graphEntry, std::shared_ptr<binder::Expression> docs,
-        std::shared_ptr<binder::Expression> query, const catalog::IndexCatalogEntry& entry,
-        QueryFTSOptionalParams optionalParams)
-        : GDSBindData{std::move(graphEntry), std::move(docs)}, query{std::move(query)},
-          entry{entry}, optionalParams{std::move(optionalParams)},
+    QueryFTSBindData(binder::expression_vector columns, graph::GraphEntry graphEntry,
+        std::shared_ptr<binder::Expression> docs, std::shared_ptr<binder::Expression> query,
+        const catalog::IndexCatalogEntry& entry, QueryFTSOptionalParams optionalParams)
+        : GDSBindData{std::move(columns), std::move(graphEntry), std::move(docs)},
+          query{std::move(query)}, entry{entry}, optionalParams{std::move(optionalParams)},
           outputTableID{
               nodeOutput->constCast<binder::NodeExpression>().getSingleEntry()->getTableID()} {}
     QueryFTSBindData(const QueryFTSBindData& other)
@@ -39,7 +39,7 @@ struct QueryFTSBindData final : function::GDSBindData {
 
     QueryFTSConfig getConfig() const { return optionalParams.getConfig(); }
 
-    std::unique_ptr<GDSBindData> copy() const override {
+    std::unique_ptr<function::TableFuncBindData> copy() const override {
         return std::make_unique<QueryFTSBindData>(*this);
     }
 };

extension/fts/src/include/function/query_fts_index.h

@@ -2,48 +2,10 @@
 
 #include "function/gds/gds.h"
 #include "function/gds/gds_frontier.h"
-#include "function/gds_function.h"
 
 namespace kuzu {
 namespace fts_extension {
 
-class QueryFTSAlgorithm : public function::GDSAlgorithm {
-public:
-    static constexpr char SCORE_PROP_NAME[] = "score";
-    static constexpr char DOC_FREQUENCY_PROP_NAME[] = "df";
-    static constexpr char TERM_FREQUENCY_PROP_NAME[] = "tf";
-    static constexpr char DOC_LEN_PROP_NAME[] = "len";
-    static constexpr char DOC_ID_PROP_NAME[] = "docID";
-
-public:
-    QueryFTSAlgorithm() = default;
-    QueryFTSAlgorithm(const QueryFTSAlgorithm& other) : GDSAlgorithm{other} {}
-
-    /*
-     * Inputs include the following:
-     *
-     * graph::ANY
-     * srcNode::NODE
-     * queryString: STRING
-     */
-    std::vector<common::LogicalTypeID> getParameterTypeIDs() const override {
-        return {common::LogicalTypeID::STRING /* tableName */,
-            common::LogicalTypeID::STRING /* indexName */,
-            common::LogicalTypeID::STRING /* query */};
-    }
-
-    void exec(processor::ExecutionContext* executionContext) override;
-
-    std::unique_ptr<GDSAlgorithm> copy() const override {
-        return std::make_unique<QueryFTSAlgorithm>(*this);
-    }
-
-    binder::expression_vector getResultColumns(
-        const function::GDSBindInput& bindInput) const override;
-
-    void bind(const function::GDSBindInput& input, main::ClientContext&) override;
-};
-
 struct QueryFTSFunction {
     static constexpr const char* name = "QUERY_FTS_INDEX";
 

src/binder/bind/bind_table_function.cpp

@@ -12,35 +12,35 @@ using namespace kuzu::function;
 namespace kuzu {
 namespace binder {
 
-static void validateParameterType(const expression_vector& positionalParams) {
-    for (auto& param : positionalParams) {
-        ExpressionUtil::validateExpressionType(*param,
-            {ExpressionType::LITERAL, ExpressionType::PARAMETER});
-    }
-}
-
 BoundTableScanInfo Binder::bindTableFunc(const std::string& tableFuncName,
     const parser::ParsedExpression& expr, std::vector<parser::YieldVariable> yieldVariables) {
     auto entry = clientContext->getCatalog()->getFunctionEntry(clientContext->getTransaction(),
         tableFuncName, clientContext->useInternalCatalogEntry());
     expression_vector positionalParams;
     std::vector<LogicalType> positionalParamTypes;
     optional_params_t optionalParams;
+    expression_vector optionalParamsLegacy;
     for (auto i = 0u; i < expr.getNumChildren(); i++) {
         auto& childExpr = *expr.getChild(i);
         auto param = expressionBinder.bindExpression(childExpr);
         if (!childExpr.hasAlias()) {
+            ExpressionUtil::validateExpressionType(*param,
+                {ExpressionType::LITERAL, ExpressionType::PARAMETER});
             positionalParams.push_back(param);
             positionalParamTypes.push_back(param->getDataType().copy());
         } else {
-            ExpressionUtil::validateExpressionType(*param, ExpressionType::LITERAL);
-            auto literalExpr = param->constPtrCast<LiteralExpression>();
-            optionalParams.emplace(childExpr.getAlias(), literalExpr->getValue());
+            ExpressionUtil::validateExpressionType(*param,
+                {ExpressionType::LITERAL, ExpressionType::PARAMETER});
+            if (param->expressionType == ExpressionType::LITERAL) {
+                auto literalExpr = param->constPtrCast<LiteralExpression>();
+                optionalParams.emplace(childExpr.getAlias(), literalExpr->getValue());
+            }
+            param->setAlias(expr.getChild(i)->getAlias());
+            optionalParamsLegacy.push_back(param);
         }
     }
     auto func = BuiltInFunctionsUtils::matchFunction(tableFuncName, positionalParamTypes,
         entry->ptrCast<catalog::FunctionCatalogEntry>());
-    validateParameterType(positionalParams);
     auto tableFunc = func->constPtrCast<TableFunction>();
     std::vector<common::LogicalType> inputTypes;
     if (tableFunc->inferInputTypes) {
@@ -66,6 +66,7 @@ BoundTableScanInfo Binder::bindTableFunc(const std::string& tableFuncName,
     auto bindInput = TableFuncBindInput();
     bindInput.params = std::move(positionalParams);
     bindInput.optionalParams = std::move(optionalParams);
+    bindInput.optionalParamsLegacy = std::move(optionalParamsLegacy);
     bindInput.binder = this;
     bindInput.yieldVariables = std::move(yieldVariables);
     return BoundTableScanInfo{*tableFunc, tableFunc->bindFunc(clientContext, &bindInput)};

src/binder/bind/read/bind_in_query_call.cpp

@@ -1,11 +1,7 @@
 #include "binder/binder.h"
-#include "binder/expression/expression_util.h"
-#include "binder/query/reading_clause/bound_gds_call.h"
 #include "binder/query/reading_clause/bound_table_function_call.h"
 #include "catalog/catalog.h"
 #include "common/exception/binder.h"
-#include "function/built_in_function_utils.h"
-#include "function/gds_function.h"
 #include "main/client_context.h"
 #include "parser/expression/parsed_function_expression.h"
 #include "parser/query/reading_clause/in_query_call_clause.h"
@@ -34,35 +30,6 @@ std::unique_ptr<BoundReadingClause> Binder::bindInQueryCall(const ReadingClause&
         boundReadingClause =
             std::make_unique<BoundTableFunctionCall>(std::move(boundTableFunction));
     } break;
-    case CatalogEntryType::GDS_FUNCTION_ENTRY: {
-        expression_vector children;
-        std::vector<LogicalType> childrenTypes;
-        expression_vector optionalParams;
-        for (auto i = 0u; i < functionExpr->getNumChildren(); i++) {
-            auto child = expressionBinder.bindExpression(*functionExpr->getChild(i));
-            if (!functionExpr->getChild(i)->hasAlias()) {
-                children.push_back(child);
-                childrenTypes.push_back(child->getDataType().copy());
-            } else {
-                ExpressionUtil::validateExpressionType(*child,
-                    {ExpressionType::LITERAL, ExpressionType::PARAMETER});
-                child->setAlias(functionExpr->getChild(i)->getAlias());
-                optionalParams.push_back(child);
-            }
-        }
-        auto func = BuiltInFunctionsUtils::matchFunction(functionName, childrenTypes,
-            entry->ptrCast<FunctionCatalogEntry>());
-        auto gdsFunc = func->constPtrCast<GDSFunction>()->copy();
-        auto input = GDSBindInput();
-        input.params = children;
-        input.binder = this;
-        input.optionalParams = std::move(optionalParams);
-        input.yieldVariables = call.getYieldVariables();
-        gdsFunc.gds->bind(input, *clientContext);
-        auto columns = gdsFunc.gds->getResultColumns(input);
-        auto info = BoundGDSCallInfo(gdsFunc.copy(), std::move(columns));
-        boundReadingClause = std::make_unique<BoundGDSCall>(std::move(info));
-    } break;
     default:
         throw BinderException(
             stringFormat("{} is not a table or algorithm function.", functionName));