>From Preetham Poluparthi <[email protected]>: Preetham Poluparthi has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21334?usp=email )
Change subject: WIP : Avoid self loops in activity graph ...................................................................... WIP : Avoid self loops in activity graph Change-Id: Ib521f376c3540db2cfc23c1afdb5370d61c8eb7d --- M asterixdb/NOTICE M asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/misc/query-ASTERIXDB-3415/query-ASTERIXDB-3415.3.query.sqlpp M hyracks-fullstack/NOTICE M hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java M hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/rewriter/ActivityClusterGraphRewriter.java 5 files changed, 105 insertions(+), 109 deletions(-) git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/34/21334/1 diff --git a/asterixdb/NOTICE b/asterixdb/NOTICE index 5118782..a44569b 100644 --- a/asterixdb/NOTICE +++ b/asterixdb/NOTICE @@ -1,5 +1,5 @@ Apache AsterixDB -Copyright 2015-2025 The Apache Software Foundation +Copyright 2015-2026 The Apache Software Foundation This product includes software developed at The Apache Software Foundation (http://www.apache.org/). diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/misc/query-ASTERIXDB-3415/query-ASTERIXDB-3415.3.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/misc/query-ASTERIXDB-3415/query-ASTERIXDB-3415.3.query.sqlpp index 517000d..fa59ccd 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/misc/query-ASTERIXDB-3415/query-ASTERIXDB-3415.3.query.sqlpp +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/misc/query-ASTERIXDB-3415/query-ASTERIXDB-3415.3.query.sqlpp @@ -24,10 +24,10 @@ use test; SELECT ( - SELECT c1.f1 - FROM collection1 AS c1 - WHERE c1.id IN c3.items - UNION ALL - SELECT c2.f2 - FROM collection2 c2 ) + SELECT c1.f1 + FROM collection1 AS c1 + WHERE c1.id IN c3.items +UNION ALL +SELECT c2.f2 +FROM collection2 c2 ) FROM collection3 AS c3 \ No newline at end of file diff --git a/hyracks-fullstack/NOTICE b/hyracks-fullstack/NOTICE index 722db88..9050c82 100644 --- a/hyracks-fullstack/NOTICE +++ b/hyracks-fullstack/NOTICE @@ -1,5 +1,5 @@ Apache Hyracks and Algebricks -Copyright 2015-2025 The Apache Software Foundation +Copyright 2015-2026 The Apache Software Foundation This product includes software developed at The Apache Software Foundation (http://www.apache.org/). diff --git a/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java b/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java index 14a44ac..883fc6b 100644 --- a/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java +++ b/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java @@ -222,7 +222,7 @@ } } boolean[] materializationFlags = computeMaterilizationFlags(group); - if (group.isEmpty() || isUnionAllSink(group)) { + if (group.isEmpty()) { continue; } candidate = group.get(0); diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/rewriter/ActivityClusterGraphRewriter.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/rewriter/ActivityClusterGraphRewriter.java index f3059c1..a1af8a8 100644 --- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/rewriter/ActivityClusterGraphRewriter.java +++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/rewriter/ActivityClusterGraphRewriter.java @@ -111,6 +111,75 @@ blocked2BlockerMap.putAll(replacedBlocked2BlockerMap); } + SuperActivity visit(ActivityId activityId, ActivityCluster ac, Map<ActivityId, SuperActivity> superActivities, + Map<IActivity, SuperActivity> invertedActivitySuperActivityMap, Map<ActivityId, Boolean> visited) { + + Map<ActivityId, IActivity> activities = ac.getActivityMap(); + Map<ActivityId, List<IConnectorDescriptor>> activityInputMap = ac.getActivityInputMap(); + Map<ConnectorDescriptorId, Pair<Pair<IActivity, Integer>, Pair<IActivity, Integer>>> connectorActivityMap = + ac.getConnectorActivityMap(); + + IActivity startActivity = activities.get(activityId); + + List<IConnectorDescriptor> inputConnectors = activityInputMap.get(activityId); + + SuperActivity superActivity = null; + + if (inputConnectors != null) { + for (IConnectorDescriptor inputConnector : inputConnectors) { + Pair<Pair<IActivity, Integer>, Pair<IActivity, Integer>> endPoints = + connectorActivityMap.get(inputConnector.getConnectorId()); + + IActivity child = endPoints.getLeft().getLeft(); + + SuperActivity childSuperActivity = invertedActivitySuperActivityMap.get(child); + if (childSuperActivity == null) { + childSuperActivity = visit(endPoints.getLeft().getLeft().getActivityId(), ac, superActivities, + invertedActivitySuperActivityMap, visited); + } + + if (inputConnector.getClass().getName().contains(ONE_TO_ONE_CONNECTOR)) { + + if (superActivity == null) { + superActivity = childSuperActivity; + superActivity.addActivity(startActivity); + invertedActivitySuperActivityMap.put(startActivity, superActivity); + // Swallow + } else if (superActivity == childSuperActivity) { + // Already swallowed + + } else { + // continue; + } + + } else { + + } + + } + } + + if (superActivity != null) { + return superActivity; + } + + return createNewSuperActivity(ac, superActivities, invertedActivitySuperActivityMap, activityId, startActivity); + + } + + void constructSuperAcitivites(ActivityCluster ac, Map<ActivityId, SuperActivity> superActivities, + Map<IActivity, SuperActivity> invertedActivitySuperActivityMap) { + Map<ActivityId, IActivity> activities = ac.getActivityMap(); + Map<ActivityId, List<IConnectorDescriptor>> activityOutputMap = ac.getActivityOutputMap(); + Map<ActivityId, Boolean> visited = new HashMap<>(); + + activities.forEach((activityId, activity) -> { + if (activityOutputMap.get(activityId) == null) { + visit(activityId, ac, superActivities, invertedActivitySuperActivityMap, visited); + } + }); + } + /** * rewrite an activity cluster internally * @@ -125,104 +194,9 @@ Map<ConnectorDescriptorId, Pair<Pair<IActivity, Integer>, Pair<IActivity, Integer>>> connectorActivityMap = ac.getConnectorActivityMap(); ActivityClusterGraph acg = ac.getActivityClusterGraph(); - Map<ActivityId, IActivity> startActivities = new HashMap<>(); + Map<ActivityId, SuperActivity> superActivities = new HashMap<>(); - Map<ActivityId, Queue<IActivity>> toBeExpendedMap = new HashMap<>(); - - /* - * Build the initial super activities - */ - activities.forEach((activityId, activity) -> { - if (activityInputMap.get(activityId) == null) { - startActivities.put(activityId, activity); - /* - * use the start activity's id as the id of the super activity - */ - createNewSuperActivity(ac, superActivities, toBeExpendedMap, invertedActivitySuperActivityMap, - activityId, activity); - } - }); - - /* - * expand one-to-one connected activity cluster by the BFS order. - * after the while-loop, the original activities are partitioned - * into equivalent classes, one-per-super-activity. - */ - Map<ActivityId, SuperActivity> clonedSuperActivities = new HashMap<ActivityId, SuperActivity>(); - while (toBeExpendedMap.size() > 0) { - clonedSuperActivities.clear(); - clonedSuperActivities.putAll(superActivities); - for (Entry<ActivityId, SuperActivity> entry : clonedSuperActivities.entrySet()) { - ActivityId superActivityId = entry.getKey(); - SuperActivity superActivity = entry.getValue(); - - /* - * for the case where the super activity has already been swallowed - */ - if (superActivities.get(superActivityId) == null) { - continue; - } - - /* - * expend the super activity - */ - Queue<IActivity> toBeExpended = toBeExpendedMap.get(superActivityId); - if (toBeExpended == null) { - /* - * Nothing to expand - */ - continue; - } - IActivity expendingActivity = toBeExpended.poll(); - List<IConnectorDescriptor> outputConnectors = activityOutputMap.get(expendingActivity.getActivityId()); - if (outputConnectors != null) { - for (IConnectorDescriptor outputConn : outputConnectors) { - Pair<Pair<IActivity, Integer>, Pair<IActivity, Integer>> endPoints = - connectorActivityMap.get(outputConn.getConnectorId()); - IActivity newActivity = endPoints.getRight().getLeft(); - SuperActivity existingSuperActivity = invertedActivitySuperActivityMap.get(newActivity); - if (outputConn.getClass().getName().contains(ONE_TO_ONE_CONNECTOR)) { - /* - * expend the super activity cluster on an one-to-one out-bound connection - */ - if (existingSuperActivity == null) { - superActivity.addActivity(newActivity); - toBeExpended.add(newActivity); - invertedActivitySuperActivityMap.put(newActivity, superActivity); - } else { - /* - * the two activities already in the same super activity - */ - if (existingSuperActivity == superActivity) { - continue; - } - /* - * swallow an existing super activity - */ - swallowExistingSuperActivity(superActivities, toBeExpendedMap, - invertedActivitySuperActivityMap, superActivity, superActivityId, - existingSuperActivity); - } - } else { - if (existingSuperActivity == null) { - /* - * create new activity - */ - createNewSuperActivity(ac, superActivities, toBeExpendedMap, - invertedActivitySuperActivityMap, newActivity.getActivityId(), newActivity); - } - } - } - } - - /* - * remove the to-be-expended queue if it is empty - */ - if (toBeExpended.isEmpty()) { - toBeExpendedMap.remove(superActivityId); - } - } - } + constructSuperAcitivites(ac, superActivities, invertedActivitySuperActivityMap); Map<ConnectorDescriptorId, IConnectorDescriptor> connMap = ac.getConnectorMap(); Map<ConnectorDescriptorId, RecordDescriptor> connRecordDesc = ac.getConnectorRecordDescriptorMap(); @@ -249,11 +223,13 @@ connectorActivityMap.forEach((connectorId, endPoints) -> { IActivity producerActivity = endPoints.getLeft().getLeft(); IActivity consumerActivity = endPoints.getRight().getLeft(); + SuperActivity producerSuperActivity = invertedActivitySuperActivityMap.get(producerActivity); + SuperActivity consumerSuperActivity = invertedActivitySuperActivityMap.get(consumerActivity); int producerPort = endPoints.getLeft().getRight(); int consumerPort = endPoints.getRight().getRight(); RecordDescriptor recordDescriptor = connRecordDesc.get(connectorId); IConnectorDescriptor conn = connMap.get(connectorId); - if (conn.getClass().getName().contains(ONE_TO_ONE_CONNECTOR)) { + if (producerSuperActivity == consumerSuperActivity) { /* * connection edge between inner activities */ @@ -264,8 +240,6 @@ /* * connection edge between super activities */ - SuperActivity producerSuperActivity = invertedActivitySuperActivityMap.get(producerActivity); - SuperActivity consumerSuperActivity = invertedActivitySuperActivityMap.get(consumerActivity); int producerSAPort = superActivityProducerPort.get(producerSuperActivity); int consumerSAPort = superActivityConsumerPort.get(consumerSuperActivity); newActivityCluster.addConnector(conn); @@ -338,6 +312,15 @@ invertedActivitySuperActivityMap.put(activity, superActivity); } + private SuperActivity createNewSuperActivity(ActivityCluster acg, Map<ActivityId, SuperActivity> superActivities, + Map<IActivity, SuperActivity> invertedActivitySuperActivityMap, ActivityId activityId, IActivity activity) { + SuperActivity superActivity = new SuperActivity(acg.getActivityClusterGraph(), acg.getId(), activityId); + superActivities.put(activityId, superActivity); + superActivity.addActivity(activity); + invertedActivitySuperActivityMap.put(activity, superActivity); + return superActivity; + } + /** * One super activity swallows another existing super activity. * @@ -371,4 +354,17 @@ tbeQueue.addAll(existingTbeQueque); } } + + private void swallowExistingSuperActivity(Map<ActivityId, SuperActivity> superActivities, + Map<IActivity, SuperActivity> invertedActivitySuperActivityMap, SuperActivity superActivity, + ActivityId superActivityId, SuperActivity existingSuperActivity) { + ActivityId existingSuperActivityId = existingSuperActivity.getActivityId(); + superActivities.remove(existingSuperActivityId); + for (Entry<ActivityId, IActivity> existingEntry : existingSuperActivity.getActivityMap().entrySet()) { + IActivity existingActivity = existingEntry.getValue(); + superActivity.addActivity(existingActivity); + invertedActivitySuperActivityMap.put(existingActivity, superActivity); + } + } + } -- To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21334?usp=email To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email Gerrit-MessageType: newchange Gerrit-Project: asterixdb Gerrit-Branch: master Gerrit-Change-Id: Ib521f376c3540db2cfc23c1afdb5370d61c8eb7d Gerrit-Change-Number: 21334 Gerrit-PatchSet: 1 Gerrit-Owner: Preetham Poluparthi <[email protected]>
