From 453572a2b17e9ceeb7e0600d44fc5e19a56cff9a Mon Sep 17 00:00:00 2001 From: blokhinnv Date: Mon, 29 Jun 2026 17:25:14 +0300 Subject: [PATCH 1/2] Add teleportation weights parameter to PageRank algorithm (based on internal id for now) --- algo/src/function/page_rank.cpp | 133 ++++++++++++++++-- .../function/config/page_rank_config.h | 4 + algo/test/test_files/page_rank.test | 23 +++ 3 files changed, 145 insertions(+), 15 deletions(-) diff --git a/algo/src/function/page_rank.cpp b/algo/src/function/page_rank.cpp index 377e969b..151bc261 100644 --- a/algo/src/function/page_rank.cpp +++ b/algo/src/function/page_rank.cpp @@ -1,7 +1,9 @@ #include "binder/binder.h" +#include "binder/expression/expression_util.h" #include "common/exception/binder.h" #include "common/string_utils.h" #include "common/task_system/progress_bar.h" +#include "common/types/value/nested.h" #include "function/algo_function.h" #include "function/config/max_iterations_config.h" #include "function/config/page_rank_config.h" @@ -11,6 +13,7 @@ #include "function/table/bind_input.h" #include "processor/execution_context.h" #include "transaction/transaction.h" +#include using namespace lbug::processor; using namespace lbug::common; @@ -22,30 +25,83 @@ using namespace lbug::function; namespace lbug { namespace algo_extension { +struct TeleportationWeightsParam { + std::shared_ptr param = nullptr; + std::optional paramVal; + + TeleportationWeightsParam() = default; + + explicit TeleportationWeightsParam(std::shared_ptr param) + : param{std::move(param)} {} + + static void validate(const Value& mapVal) { + if (mapVal.getDataType().getLogicalTypeID() != LogicalTypeID::MAP) { + throw BinderException{"Teleportation weights must be a MAP(INTERNAL_ID, DOUBLE)."}; + } + auto weightSum = 0.0; + for (auto i = 0u; i < mapVal.getChildrenSize(); ++i) { + auto* entry = NestedVal::getChildVal(&mapVal, i); + auto* keyVal = NestedVal::getChildVal(entry, 0); + auto* weightVal = NestedVal::getChildVal(entry, 1); + if (keyVal->getDataType().getLogicalTypeID() != LogicalTypeID::INTERNAL_ID) { + throw BinderException{"Teleportation weight keys must be INTERNAL_ID."}; + } + if (weightVal->getDataType().getLogicalTypeID() != LogicalTypeID::DOUBLE) { + throw BinderException{"Teleportation weight values must be DOUBLE."}; + } + auto weight = weightVal->getValue(); + if (weight < 0) { + throw BinderException{"Teleportation weights must be non-negative."}; + } + weightSum += weight; + } + if (weightSum <= 0) { + throw BinderException{"Sum of teleportation weights must be positive."}; + } + } + + void evaluateParam(main::ClientContext* /*context*/) { + if (!param) { + paramVal.reset(); + return; + } + auto value = ExpressionUtil::evaluateAsLiteralValue(*param); + validate(value); + paramVal = std::move(value); + } + + bool isSet() const { return param != nullptr; } + + const Value& getParamVal() const { return paramVal.value(); } +}; + struct PageRankOptionalParams final : public MaxIterationOptionalParams { OptionalParam dampingFactor; OptionalParam tolerance; OptionalParam normalize; + TeleportationWeightsParam teleportationWeights; explicit PageRankOptionalParams(const expression_vector& optionalParams); // For copy only PageRankOptionalParams(OptionalParam maxIterations, OptionalParam dampingFactor, OptionalParam tolerance, - OptionalParam normalize) + OptionalParam normalize, TeleportationWeightsParam teleportationWeights) : MaxIterationOptionalParams{maxIterations}, dampingFactor{std::move(dampingFactor)}, - tolerance{std::move(tolerance)}, normalize{std::move(normalize)} {} + tolerance{std::move(tolerance)}, normalize{std::move(normalize)}, + teleportationWeights{std::move(teleportationWeights)} {} void evaluateParams(main::ClientContext* context) override { MaxIterationOptionalParams::evaluateParams(context); dampingFactor.evaluateParam(context); tolerance.evaluateParam(context); normalize.evaluateParam(context); + teleportationWeights.evaluateParam(context); } std::unique_ptr copy() override { return std::make_unique(maxIterations, dampingFactor, tolerance, - normalize); + normalize, teleportationWeights); } }; @@ -61,6 +117,8 @@ PageRankOptionalParams::PageRankOptionalParams(const expression_vector& optional tolerance = function::OptionalParam(optionalParam); } else if (paramName == NormalizeInitial::NAME) { normalize = function::OptionalParam(optionalParam); + } else if (paramName == TeleportationWeights::NAME) { + teleportationWeights = TeleportationWeightsParam{optionalParam}; } else { throw BinderException{"Unknown optional parameter: " + optionalParam->getAlias()}; } @@ -165,32 +223,41 @@ class PNextUpdateEdgeCompute : public EdgeCompute { PValues& pNext; }; -// Evaluate rank = above result * dampingFactor + {(1 - dampingFactor) / |V|} (constant) +// Evaluate rank = above result * dampingFactor + teleport (uniform or per-node) class PNextUpdateVertexCompute : public GDSVertexCompute { public: - PNextUpdateVertexCompute(double dampingFactor, double constant, PValues& pNext, - NodeOffsetMaskMap* nodeMask) - : GDSVertexCompute{nodeMask}, dampingFactor{dampingFactor}, constant{constant}, - pNext{pNext} {} + PNextUpdateVertexCompute(double dampingFactor, double uniformTeleport, PValues* perNodeTeleport, + PValues& pNext, NodeOffsetMaskMap* nodeMask) + : GDSVertexCompute{nodeMask}, dampingFactor{dampingFactor}, + uniformTeleport{uniformTeleport}, perNodeTeleport{perNodeTeleport}, pNext{pNext} {} - void beginOnTableInternal(table_id_t tableID) override { pNext.pinTable(tableID); } + void beginOnTableInternal(table_id_t tableID) override { + pNext.pinTable(tableID); + if (perNodeTeleport != nullptr) { + perNodeTeleport->pinTable(tableID); + } + } void vertexCompute(offset_t startOffset, offset_t endOffset, table_id_t) override { for (auto i = startOffset; i < endOffset; ++i) { if (skip(i)) { continue; } - pNext.setValue(i, pNext.getValue(i) * dampingFactor + constant); + auto teleport = + perNodeTeleport != nullptr ? perNodeTeleport->getValue(i) : uniformTeleport; + pNext.setValue(i, pNext.getValue(i) * dampingFactor + teleport); } } std::unique_ptr copy() override { - return std::make_unique(dampingFactor, constant, pNext, nodeMask); + return std::make_unique(dampingFactor, uniformTeleport, + perNodeTeleport, pNext, nodeMask); } private: double dampingFactor; - double constant; + double uniformTeleport; + PValues* perNodeTeleport; PValues& pNext; }; @@ -264,6 +331,31 @@ class PageRankResultVertexCompute : public GDSResultVertexCompute { std::unique_ptr rankVector; }; +static PValues buildTeleportConstants(const table_id_map_t& maxOffsetMap, + MemoryManager* mm, const Value& mapVal, double teleportScale) { + PValues rawWeights(maxOffsetMap, mm, 0); + auto weightSum = 0.0; + for (auto i = 0u; i < mapVal.getChildrenSize(); ++i) { + auto* entry = NestedVal::getChildVal(&mapVal, i); + auto nodeID = NestedVal::getChildVal(entry, 0)->getValue(); + auto weight = NestedVal::getChildVal(entry, 1)->getValue(); + rawWeights.pinTable(nodeID.tableID); + rawWeights.setValue(nodeID.offset, weight); + weightSum += weight; + } + + PValues teleportConstants(maxOffsetMap, mm, 0); + for (const auto& [tableID, maxOffset] : maxOffsetMap) { + teleportConstants.pinTable(tableID); + rawWeights.pinTable(tableID); + for (auto offset = 0u; offset < maxOffset; ++offset) { + teleportConstants.setValue(offset, + teleportScale * rawWeights.getValue(offset) / weightSum); + } + } + return teleportConstants; +} + static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { auto clientContext = input.context->clientContext; auto transaction = transaction::Transaction::Get(*clientContext); @@ -290,7 +382,18 @@ static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { auto frontierPair = std::make_unique(std::move(currentFrontier), std::move(nextFrontier)); auto computeState = GDSComputeState(std::move(frontierPair), nullptr, nullptr); - auto pNextUpdateConstant = (1 - config.dampingFactor.getParamVal()) * initialValue; + auto dampingFactor = config.dampingFactor.getParamVal(); + double uniformTeleport = 0; + PValues teleportConstants(maxOffsetMap, mm, 0); + PValues* perNodeTeleport = nullptr; + if (!config.teleportationWeights.isSet()) { + uniformTeleport = (1 - dampingFactor) * initialValue; + } else { + auto teleportScale = (1 - dampingFactor) * initialValue * numNodes; + teleportConstants = buildTeleportConstants(maxOffsetMap, mm, + config.teleportationWeights.getParamVal(), teleportScale); + perNodeTeleport = &teleportConstants; + } while (currentIter < config.maxIterations.getParamVal()) { computeState.frontierPair->resetCurrentIter(); computeState.frontierPair->setActiveNodesForNextIter(); @@ -300,8 +403,8 @@ static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { std::make_unique(degrees, *pCurrent, *pNext); GDSUtils::runAlgorithmEdgeCompute(input.context, computeState, graph, ExtendDirection::BWD, 1); - auto pNextUpdateVC = PNextUpdateVertexCompute(config.dampingFactor.getParamVal(), - pNextUpdateConstant, *pNext, sharedState->getGraphNodeMaskMap()); + auto pNextUpdateVC = PNextUpdateVertexCompute(dampingFactor, uniformTeleport, perNodeTeleport, + *pNext, sharedState->getGraphNodeMaskMap()); GDSUtils::runVertexCompute(input.context, GDSDensityState::DENSE, graph, pNextUpdateVC); std::atomic diff; diff.store(0); diff --git a/algo/src/include/function/config/page_rank_config.h b/algo/src/include/function/config/page_rank_config.h index 83111b97..472001d6 100644 --- a/algo/src/include/function/config/page_rank_config.h +++ b/algo/src/include/function/config/page_rank_config.h @@ -36,5 +36,9 @@ struct NormalizeInitial { static constexpr bool DEFAULT_VALUE = true; }; +struct TeleportationWeights { + static constexpr const char* NAME = "teleportationweights"; +}; + } // namespace function } // namespace lbug diff --git a/algo/test/test_files/page_rank.test b/algo/test/test_files/page_rank.test index af5d6274..7f326741 100644 --- a/algo/test/test_files/page_rank.test +++ b/algo/test/test_files/page_rank.test @@ -104,3 +104,26 @@ Greg||0.021375 |ABFsUni|0.058538 |CsWork|0.015000 |DEsWork|0.015000 + +-STATEMENT CALL page_rank('PK', + teleportationWeights := map([internal_id(0, 0), internal_id(0, 1)], [0.7, 0.3]) +) RETURN node.fName, rank; +---- 8 +Alice|0.296897 +Bob|0.250144 +Carol|0.215079 +Dan|0.215079 +Elizabeth|0.000000 +Farooq|0.000000 +Greg|0.000000 +Hubert Blaine Wolfeschlegelsteinhausenbergerdorff|0.000000 +-STATEMENT CALL page_rank('PK', + teleportationWeights := map([internal_id(0, 0), internal_id(0, 1)], [-0.7, 0.3]) +) RETURN node.fName, rank; +---- error +Binder exception: Teleportation weights must be non-negative. +-STATEMENT CALL page_rank('PK', + teleportationWeights := map([internal_id(0, 0), internal_id(0, 1)], [0.0, 0.0]) +) RETURN node.fName, rank; +---- error +Binder exception: Sum of teleportation weights must be positive. From 545f479a8a4d80b6f23ebf0e2c46bed18e13b9d9 Mon Sep 17 00:00:00 2001 From: blokhinnv Date: Mon, 29 Jun 2026 18:31:59 +0300 Subject: [PATCH 2/2] Update PPR to support structured teleportation weights. --- algo/src/function/page_rank.cpp | 250 ++++++++++++++++++++++++---- algo/test/test_files/page_rank.test | 19 ++- 2 files changed, 231 insertions(+), 38 deletions(-) diff --git a/algo/src/function/page_rank.cpp b/algo/src/function/page_rank.cpp index 151bc261..40d442ed 100644 --- a/algo/src/function/page_rank.cpp +++ b/algo/src/function/page_rank.cpp @@ -1,6 +1,8 @@ #include "binder/binder.h" #include "binder/expression/expression_util.h" +#include "catalog/catalog_entry/table_catalog_entry.h" #include "common/exception/binder.h" +#include "common/exception/runtime.h" #include "common/string_utils.h" #include "common/task_system/progress_bar.h" #include "common/types/value/nested.h" @@ -11,6 +13,7 @@ #include "function/gds/gds_utils.h" #include "function/gds/gds_vertex_compute.h" #include "function/table/bind_input.h" +#include "graph/graph.h" #include "processor/execution_context.h" #include "transaction/transaction.h" #include @@ -21,10 +24,148 @@ using namespace lbug::binder; using namespace lbug::storage; using namespace lbug::graph; using namespace lbug::function; +using namespace lbug::catalog; namespace lbug { namespace algo_extension { +namespace { + +static constexpr char KEY_PROPERTY_FIELD[] = "keyproperty"; +static constexpr char WEIGHTS_FIELD[] = "weights"; + +struct TableTeleportSpec { + std::string tableName; + std::string keyProperty; + Value weights; +}; + +static const Value* getStructFieldVal(const Value& structVal, const std::string& fieldName) { + if (structVal.getDataType().getLogicalTypeID() != LogicalTypeID::STRUCT) { + return nullptr; + } + auto lowerFieldName = StringUtils::getLower(fieldName); + for (auto i = 0u; i < StructType::getNumFields(structVal.getDataType()); ++i) { + auto& field = StructType::getField(structVal.getDataType(), i); + if (StringUtils::getLower(field.getName()) == lowerFieldName) { + return NestedVal::getChildVal(&structVal, i); + } + } + return nullptr; +} + +static void validateWeightsMap(const Value& weightsVal) { + if (weightsVal.getDataType().getLogicalTypeID() != LogicalTypeID::MAP) { + throw BinderException{"Teleportation weights must be a MAP."}; + } + auto weightSum = 0.0; + for (auto i = 0u; i < weightsVal.getChildrenSize(); ++i) { + auto* entry = NestedVal::getChildVal(&weightsVal, i); + auto* weightVal = NestedVal::getChildVal(entry, 1); + if (weightVal->getDataType().getLogicalTypeID() != LogicalTypeID::DOUBLE) { + throw BinderException{"Teleportation weight values must be DOUBLE."}; + } + auto weight = weightVal->getValue(); + if (weight < 0) { + throw BinderException{"Teleportation weights must be non-negative."}; + } + weightSum += weight; + } + if (weightSum <= 0) { + throw BinderException{"Sum of teleportation weights must be positive."}; + } +} + +static TableTeleportSpec parseTableTeleportSpec(const std::string& tableName, const Value& tableVal) { + if (tableVal.getDataType().getLogicalTypeID() != LogicalTypeID::STRUCT) { + throw BinderException{std::format( + "Teleportation weights for table {} must be a STRUCT with keyProperty and weights.", + tableName)}; + } + auto* keyPropertyVal = getStructFieldVal(tableVal, KEY_PROPERTY_FIELD); + auto* weightsVal = getStructFieldVal(tableVal, WEIGHTS_FIELD); + if (keyPropertyVal == nullptr || weightsVal == nullptr) { + throw BinderException{std::format( + "Teleportation weights for table {} must contain keyProperty and weights.", tableName)}; + } + if (keyPropertyVal->getDataType().getLogicalTypeID() != LogicalTypeID::STRING) { + throw BinderException{"Teleportation keyProperty must be a STRING."}; + } + validateWeightsMap(*weightsVal); + return TableTeleportSpec{tableName, keyPropertyVal->getValue(), *weightsVal}; +} + +static std::vector parseTeleportationWeights(const Value& configVal) { + if (configVal.getDataType().getLogicalTypeID() != LogicalTypeID::STRUCT) { + throw BinderException{ + "Teleportation weights must be a STRUCT mapping node table names to weight specs."}; + } + std::vector specs; + for (auto i = 0u; i < StructType::getNumFields(configVal.getDataType()); ++i) { + auto& field = StructType::getField(configVal.getDataType(), i); + auto* tableVal = NestedVal::getChildVal(&configVal, i); + specs.push_back(parseTableTeleportSpec(field.getName(), *tableVal)); + } + if (specs.empty()) { + throw BinderException{"Teleportation weights must specify at least one node table."}; + } + return specs; +} + +static const NativeGraphEntryTableInfo* findNodeTableInfo(const NativeGraphEntry& graphEntry, + const std::string& tableName) { + for (auto& nodeInfo : graphEntry.nodeInfos) { + if (StringUtils::getLower(nodeInfo.entry->getName()) == StringUtils::getLower(tableName)) { + return &nodeInfo; + } + } + return nullptr; +} + +static void validateTeleportationWeightsAgainstGraph(const Value& configVal, + const NativeGraphEntry& graphEntry) { + for (auto& spec : parseTeleportationWeights(configVal)) { + auto* nodeInfo = findNodeTableInfo(graphEntry, spec.tableName); + if (nodeInfo == nullptr) { + throw BinderException{std::format("Unknown node table in teleportation weights: {}.", + spec.tableName)}; + } + if (!nodeInfo->entry->containsProperty(spec.keyProperty)) { + throw BinderException{std::format("Unknown property {} on node table {}.", + spec.keyProperty, spec.tableName)}; + } + auto propertyTypeID = nodeInfo->entry->getProperty(spec.keyProperty).getType().getLogicalTypeID(); + auto mapKeyTypeID = MapType::getKeyType(spec.weights.getDataType()).getLogicalTypeID(); + if (propertyTypeID != mapKeyTypeID) { + throw BinderException{std::format( + "Teleportation weight keys for table {} must have type {} to match property {}.", + spec.tableName, nodeInfo->entry->getProperty(spec.keyProperty).getType().toString(), + spec.keyProperty)}; + } + } +} + +static std::optional lookupWeight(const Value& weightsMap, const Value& key) { + for (auto i = 0u; i < weightsMap.getChildrenSize(); ++i) { + auto* entry = NestedVal::getChildVal(&weightsMap, i); + auto* keyVal = NestedVal::getChildVal(entry, 0); + if (*keyVal == key) { + return NestedVal::getChildVal(entry, 1)->getValue(); + } + } + return std::nullopt; +} + +static bool isNodeActive(NodeOffsetMaskMap* nodeMask, table_id_t tableID, offset_t offset) { + if (nodeMask == nullptr) { + return true; + } + nodeMask->pin(tableID); + return nodeMask->valid(offset); +} + +} // namespace + struct TeleportationWeightsParam { std::shared_ptr param = nullptr; std::optional paramVal; @@ -34,30 +175,11 @@ struct TeleportationWeightsParam { explicit TeleportationWeightsParam(std::shared_ptr param) : param{std::move(param)} {} - static void validate(const Value& mapVal) { - if (mapVal.getDataType().getLogicalTypeID() != LogicalTypeID::MAP) { - throw BinderException{"Teleportation weights must be a MAP(INTERNAL_ID, DOUBLE)."}; - } - auto weightSum = 0.0; - for (auto i = 0u; i < mapVal.getChildrenSize(); ++i) { - auto* entry = NestedVal::getChildVal(&mapVal, i); - auto* keyVal = NestedVal::getChildVal(entry, 0); - auto* weightVal = NestedVal::getChildVal(entry, 1); - if (keyVal->getDataType().getLogicalTypeID() != LogicalTypeID::INTERNAL_ID) { - throw BinderException{"Teleportation weight keys must be INTERNAL_ID."}; - } - if (weightVal->getDataType().getLogicalTypeID() != LogicalTypeID::DOUBLE) { - throw BinderException{"Teleportation weight values must be DOUBLE."}; - } - auto weight = weightVal->getValue(); - if (weight < 0) { - throw BinderException{"Teleportation weights must be non-negative."}; - } - weightSum += weight; - } - if (weightSum <= 0) { - throw BinderException{"Sum of teleportation weights must be positive."}; - } + static void validate(const Value& configVal) { parseTeleportationWeights(configVal); } + + static void validate(const Value& configVal, const NativeGraphEntry& graphEntry) { + validate(configVal); + validateTeleportationWeightsAgainstGraph(configVal, graphEntry); } void evaluateParam(main::ClientContext* /*context*/) { @@ -331,17 +453,68 @@ class PageRankResultVertexCompute : public GDSResultVertexCompute { std::unique_ptr rankVector; }; -static PValues buildTeleportConstants(const table_id_map_t& maxOffsetMap, - MemoryManager* mm, const Value& mapVal, double teleportScale) { +static Value readChunkProperty(const graph::VertexScanState::Chunk& chunk, sel_t pos, + LogicalTypeID typeID) { + switch (typeID) { + case LogicalTypeID::STRING: + return Value(LogicalType::STRING(), chunk.getProperties(0)[pos].getAsString()); + case LogicalTypeID::INT64: + return Value(chunk.getProperties(0)[pos]); + case LogicalTypeID::INT32: + return Value(chunk.getProperties(0)[pos]); + case LogicalTypeID::INT16: + return Value(chunk.getProperties(0)[pos]); + case LogicalTypeID::INT8: + return Value(chunk.getProperties(0)[pos]); + case LogicalTypeID::UINT64: + return Value(chunk.getProperties(0)[pos]); + case LogicalTypeID::UINT32: + return Value(chunk.getProperties(0)[pos]); + case LogicalTypeID::UINT16: + return Value(chunk.getProperties(0)[pos]); + case LogicalTypeID::UINT8: + return Value(chunk.getProperties(0)[pos]); + default: + throw RuntimeException{std::format( + "Unsupported teleportation key property type: {}.", LogicalTypeUtils::toString(typeID))}; + } +} + +static PValues buildTeleportConstants(Graph* graph, const NativeGraphEntry& graphEntry, + const table_id_map_t& maxOffsetMap, MemoryManager* mm, const Value& configVal, + double teleportScale, NodeOffsetMaskMap* nodeMask) { PValues rawWeights(maxOffsetMap, mm, 0); auto weightSum = 0.0; - for (auto i = 0u; i < mapVal.getChildrenSize(); ++i) { - auto* entry = NestedVal::getChildVal(&mapVal, i); - auto nodeID = NestedVal::getChildVal(entry, 0)->getValue(); - auto weight = NestedVal::getChildVal(entry, 1)->getValue(); - rawWeights.pinTable(nodeID.tableID); - rawWeights.setValue(nodeID.offset, weight); - weightSum += weight; + for (auto& spec : parseTeleportationWeights(configVal)) { + auto* nodeInfo = findNodeTableInfo(graphEntry, spec.tableName); + if (nodeInfo == nullptr) { + continue; + } + auto tableID = nodeInfo->entry->getTableID(); + auto maxOffset = maxOffsetMap.at(tableID); + auto keyTypeID = MapType::getKeyType(spec.weights.getDataType()).getLogicalTypeID(); + auto scanState = graph->prepareVertexScan(nodeInfo->entry, {spec.keyProperty}); + rawWeights.pinTable(tableID); + for (auto chunk : graph->scanVertices(0, maxOffset, *scanState)) { + auto nodeIDs = chunk.getNodeIDs(); + for (auto i = 0u; i < chunk.size(); ++i) { + auto offset = nodeIDs[i].offset; + if (!isNodeActive(nodeMask, tableID, offset)) { + continue; + } + auto propertyValue = readChunkProperty(chunk, i, keyTypeID); + auto weight = lookupWeight(spec.weights, propertyValue); + if (!weight.has_value()) { + continue; + } + rawWeights.setValue(offset, weight.value()); + weightSum += weight.value(); + } + } + } + if (weightSum <= 0) { + throw RuntimeException{ + "Teleportation weights did not match any node in the projected graph."}; } PValues teleportConstants(maxOffsetMap, mm, 0); @@ -390,8 +563,9 @@ static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { uniformTeleport = (1 - dampingFactor) * initialValue; } else { auto teleportScale = (1 - dampingFactor) * initialValue * numNodes; - teleportConstants = buildTeleportConstants(maxOffsetMap, mm, - config.teleportationWeights.getParamVal(), teleportScale); + teleportConstants = buildTeleportConstants(graph, pageRankBindData->graphEntry, maxOffsetMap, + mm, config.teleportationWeights.getParamVal(), teleportScale, + sharedState->getGraphNodeMaskMap()); perNodeTeleport = &teleportConstants; } while (currentIter < config.maxIterations.getParamVal()) { @@ -431,6 +605,12 @@ static std::unique_ptr bindFunc(main::ClientContext* context, const TableFuncBindInput* input) { auto graphName = input->getLiteralVal(0); auto graphEntry = GDSFunction::bindGraphEntry(*context, graphName); + for (auto& optionalParam : input->optionalParamsLegacy) { + if (StringUtils::getLower(optionalParam->getAlias()) == TeleportationWeights::NAME) { + auto value = ExpressionUtil::evaluateAsLiteralValue(*optionalParam); + TeleportationWeightsParam::validate(value, graphEntry); + } + } auto nodeOutput = GDSFunction::bindNodeOutput(*input, graphEntry.getNodeEntries()); expression_vector columns; columns.push_back(nodeOutput->constCast().getInternalID()); diff --git a/algo/test/test_files/page_rank.test b/algo/test/test_files/page_rank.test index 7f326741..ad33d9f9 100644 --- a/algo/test/test_files/page_rank.test +++ b/algo/test/test_files/page_rank.test @@ -106,7 +106,9 @@ Greg||0.021375 |DEsWork|0.015000 -STATEMENT CALL page_rank('PK', - teleportationWeights := map([internal_id(0, 0), internal_id(0, 1)], [0.7, 0.3]) + teleportationWeights := { + person: {keyProperty: 'fName', weights: map(['Alice', 'Bob'], [0.7, 0.3])} + } ) RETURN node.fName, rank; ---- 8 Alice|0.296897 @@ -118,12 +120,23 @@ Farooq|0.000000 Greg|0.000000 Hubert Blaine Wolfeschlegelsteinhausenbergerdorff|0.000000 -STATEMENT CALL page_rank('PK', - teleportationWeights := map([internal_id(0, 0), internal_id(0, 1)], [-0.7, 0.3]) + teleportationWeights := { + person: {keyProperty: 'fName', weights: map(['Alice', 'Bob'], [-0.7, 0.3])} + } ) RETURN node.fName, rank; ---- error Binder exception: Teleportation weights must be non-negative. -STATEMENT CALL page_rank('PK', - teleportationWeights := map([internal_id(0, 0), internal_id(0, 1)], [0.0, 0.0]) + teleportationWeights := { + person: {keyProperty: 'fName', weights: map(['Alice', 'Bob'], [0.0, 0.0])} + } ) RETURN node.fName, rank; ---- error Binder exception: Sum of teleportation weights must be positive. +-STATEMENT CALL page_rank('PK', + teleportationWeights := { + unknown: {keyProperty: 'fName', weights: map(['Alice'], [1.0])} + } +) RETURN node.fName, rank; +---- error +Binder exception: Unknown node table in teleportation weights: unknown.