You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by tushargosavi <gi...@git.apache.org> on 2015/12/23 18:05:50 UTC
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
GitHub user tushargosavi opened a pull request:
https://github.com/apache/incubator-apex-core/pull/191
APEXCORE-272 copy operator and port attributes from module dag to parent dag.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tushargosavi/incubator-apex-core APEXCORE-272
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-core/pull/191.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #191
----
commit 0dfe868bd39fdbb36a24c83d6cbda8a201974f77
Author: Tushar R. Gosavi <tu...@apache.org>
Date: 2015-12-23T16:31:50Z
APEXCORE-272 copy operator and port attributes from module dag to parent dag.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48647941
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -907,6 +912,51 @@ else if (field.getType() == float.class || field.getType() == Float.class ||
getValue(OperatorContext.METRICS_DIMENSIONS_SCHEME));
}
+ /**
+ * Copy attribute from source attributeMap to destination attributeMap.
+ *
+ * @param dest destination attribute map.
+ * @param source source attribute map.
+ */
+ private void copyAttributes(AttributeMap dest, AttributeMap source)
+ {
+ for (Entry<Attribute<?>, ?> a : source.entrySet()) {
+ dest.put((Attribute<Object>)a.getKey(), a.getValue());
+ }
+ }
+
+ /**
+ * Copy attribute of operator and port from provided operatorMeta. This function requires
+ * operatorMeta argument is for the same operator.
+ *
+ * @param operatorMeta copy attribute from this OperatorMeta to the object.
+ */
+ private void copyFrom(OperatorMeta operatorMeta)
+ {
+ if (operator != operatorMeta.getOperator()) {
+ throw new IllegalArgumentException("Operator meta is not for the same operator ");
+ }
+
+ // copy operator attributes
+ copyAttributes(attributes, operatorMeta.getAttributes());
+
+ // copy Input port attributes
+ for (Map.Entry<InputPort<?>, InputPortMeta> entry : operatorMeta.getPortMapping().inPortMap.entrySet()) {
+ InputPort<?> key = entry.getKey();
+ InputPortMeta dest = getPortMapping().inPortMap.get(key);
--- End diff --
done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48522048
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -1359,7 +1409,8 @@ private void addDAGToCurrentDAG(ModuleMeta moduleMeta)
String name;
for (OperatorMeta operatorMeta : subDag.getAllOperators()) {
name = subDAGName + MODULE_NAMESPACE_SEPARATOR + operatorMeta.getName();
- this.addOperator(name, operatorMeta.getOperator());
+ Operator op = this.addOperator(name, operatorMeta.getOperator());
--- End diff --
OperatorMeta has name and id declared as final field. In our case we want to add the operator to top level dag with expanded name and wants to have a unique id in the parent DAG. Hence reused the addOperator api without adding a new one. This has additional cost of allocating new OperatorMeta though.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48569555
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -907,6 +912,51 @@ else if (field.getType() == float.class || field.getType() == Float.class ||
getValue(OperatorContext.METRICS_DIMENSIONS_SCHEME));
}
+ /**
+ * Copy attribute from source attributeMap to destination attributeMap.
+ *
+ * @param dest destination attribute map.
+ * @param source source attribute map.
+ */
+ private void copyAttributes(AttributeMap dest, AttributeMap source)
+ {
+ for (Entry<Attribute<?>, ?> a : source.entrySet()) {
+ dest.put((Attribute<Object>)a.getKey(), a.getValue());
+ }
+ }
+
+ /**
+ * Copy attribute of operator and port from provided operatorMeta. This function requires
+ * operatorMeta argument is for the same operator.
+ *
+ * @param operatorMeta copy attribute from this OperatorMeta to the object.
+ */
+ private void copyFrom(OperatorMeta operatorMeta)
--- End diff --
This function is in OperatorMeta class so function name gives an impression that it does deep clone of operatorMeta from given operatorMeta but actually it is only copying Attributes. I think it should be renamed to something different
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48568829
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -719,6 +719,11 @@ public void resetStreamPersistanceOnSinkRemoval(InputPortMeta sinkBeingRemoved)
removeOperator(persistOpMeta.getOperator());
}
}
+
+ private void copyFrom(StreamMeta meta)
+ {
+ this.locality = meta.getLocality();
--- End diff --
This is used at only one place and it has only one statement, so why another function call?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48647480
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -719,6 +719,11 @@ public void resetStreamPersistanceOnSinkRemoval(InputPortMeta sinkBeingRemoved)
removeOperator(persistOpMeta.getOperator());
}
}
+
+ private void copyFrom(StreamMeta meta)
+ {
+ this.locality = meta.getLocality();
--- End diff --
Removed this function and made the call inline at called site.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48569051
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -907,6 +912,51 @@ else if (field.getType() == float.class || field.getType() == Float.class ||
getValue(OperatorContext.METRICS_DIMENSIONS_SCHEME));
}
+ /**
+ * Copy attribute from source attributeMap to destination attributeMap.
+ *
+ * @param dest destination attribute map.
+ * @param source source attribute map.
+ */
+ private void copyAttributes(AttributeMap dest, AttributeMap source)
+ {
+ for (Entry<Attribute<?>, ?> a : source.entrySet()) {
+ dest.put((Attribute<Object>)a.getKey(), a.getValue());
+ }
+ }
+
+ /**
+ * Copy attribute of operator and port from provided operatorMeta. This function requires
+ * operatorMeta argument is for the same operator.
+ *
+ * @param operatorMeta copy attribute from this OperatorMeta to the object.
+ */
+ private void copyFrom(OperatorMeta operatorMeta)
+ {
+ if (operator != operatorMeta.getOperator()) {
+ throw new IllegalArgumentException("Operator meta is not for the same operator ");
+ }
+
+ // copy operator attributes
+ copyAttributes(attributes, operatorMeta.getAttributes());
+
+ // copy Input port attributes
+ for (Map.Entry<InputPort<?>, InputPortMeta> entry : operatorMeta.getPortMapping().inPortMap.entrySet()) {
+ InputPort<?> key = entry.getKey();
+ InputPortMeta dest = getPortMapping().inPortMap.get(key);
+ InputPortMeta source = entry.getValue();
+ copyAttributes(dest.attributes, source.attributes);
+ }
+
+ // copy Output port attributes
+ for (Map.Entry<OutputPort<?>, OutputPortMeta> entry : operatorMeta.getPortMapping().outPortMap.entrySet()) {
+ OutputPort<?> key = entry.getKey();
+ OutputPortMeta dest = getPortMapping().outPortMap.get(key);
+ OutputPortMeta source = entry.getValue();
+ copyAttributes(dest.attributes, source.attributes);
--- End diff --
same feedback as for input ports
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on the pull request:
https://github.com/apache/incubator-apex-core/pull/191#issuecomment-167599384
@gauravgopi123 @tweise can you take a look at this pull request?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48647938
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -907,6 +912,51 @@ else if (field.getType() == float.class || field.getType() == Float.class ||
getValue(OperatorContext.METRICS_DIMENSIONS_SCHEME));
}
+ /**
+ * Copy attribute from source attributeMap to destination attributeMap.
+ *
+ * @param dest destination attribute map.
+ * @param source source attribute map.
+ */
+ private void copyAttributes(AttributeMap dest, AttributeMap source)
+ {
+ for (Entry<Attribute<?>, ?> a : source.entrySet()) {
+ dest.put((Attribute<Object>)a.getKey(), a.getValue());
+ }
+ }
+
+ /**
+ * Copy attribute of operator and port from provided operatorMeta. This function requires
+ * operatorMeta argument is for the same operator.
+ *
+ * @param operatorMeta copy attribute from this OperatorMeta to the object.
+ */
+ private void copyFrom(OperatorMeta operatorMeta)
+ {
+ if (operator != operatorMeta.getOperator()) {
+ throw new IllegalArgumentException("Operator meta is not for the same operator ");
+ }
+
+ // copy operator attributes
+ copyAttributes(attributes, operatorMeta.getAttributes());
+
+ // copy Input port attributes
+ for (Map.Entry<InputPort<?>, InputPortMeta> entry : operatorMeta.getPortMapping().inPortMap.entrySet()) {
+ InputPort<?> key = entry.getKey();
+ InputPortMeta dest = getPortMapping().inPortMap.get(key);
+ InputPortMeta source = entry.getValue();
+ copyAttributes(dest.attributes, source.attributes);
+ }
+
+ // copy Output port attributes
+ for (Map.Entry<OutputPort<?>, OutputPortMeta> entry : operatorMeta.getPortMapping().outPortMap.entrySet()) {
+ OutputPort<?> key = entry.getKey();
+ OutputPortMeta dest = getPortMapping().outPortMap.get(key);
+ OutputPortMeta source = entry.getValue();
+ copyAttributes(dest.attributes, source.attributes);
--- End diff --
done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48647940
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -907,6 +912,51 @@ else if (field.getType() == float.class || field.getType() == Float.class ||
getValue(OperatorContext.METRICS_DIMENSIONS_SCHEME));
}
+ /**
+ * Copy attribute from source attributeMap to destination attributeMap.
+ *
+ * @param dest destination attribute map.
+ * @param source source attribute map.
+ */
+ private void copyAttributes(AttributeMap dest, AttributeMap source)
+ {
+ for (Entry<Attribute<?>, ?> a : source.entrySet()) {
+ dest.put((Attribute<Object>)a.getKey(), a.getValue());
+ }
+ }
+
+ /**
+ * Copy attribute of operator and port from provided operatorMeta. This function requires
+ * operatorMeta argument is for the same operator.
+ *
+ * @param operatorMeta copy attribute from this OperatorMeta to the object.
+ */
+ private void copyFrom(OperatorMeta operatorMeta)
+ {
+ if (operator != operatorMeta.getOperator()) {
+ throw new IllegalArgumentException("Operator meta is not for the same operator ");
+ }
+
+ // copy operator attributes
+ copyAttributes(attributes, operatorMeta.getAttributes());
+
+ // copy Input port attributes
+ for (Map.Entry<InputPort<?>, InputPortMeta> entry : operatorMeta.getPortMapping().inPortMap.entrySet()) {
+ InputPort<?> key = entry.getKey();
+ InputPortMeta dest = getPortMapping().inPortMap.get(key);
+ InputPortMeta source = entry.getValue();
--- End diff --
done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48646622
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -1359,7 +1409,8 @@ private void addDAGToCurrentDAG(ModuleMeta moduleMeta)
String name;
for (OperatorMeta operatorMeta : subDag.getAllOperators()) {
name = subDAGName + MODULE_NAMESPACE_SEPARATOR + operatorMeta.getName();
- this.addOperator(name, operatorMeta.getOperator());
+ Operator op = this.addOperator(name, operatorMeta.getOperator());
--- End diff --
Are we keeping module DAG after adding it's operators to the final DAG? If not, there is no meta object sharing between DAGs.
Introducing copy constructor will allow to share some of final collection references. As operator reference is shared between meta objects, should other references be also shared?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48492262
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -1359,7 +1409,8 @@ private void addDAGToCurrentDAG(ModuleMeta moduleMeta)
String name;
for (OperatorMeta operatorMeta : subDag.getAllOperators()) {
name = subDAGName + MODULE_NAMESPACE_SEPARATOR + operatorMeta.getName();
- this.addOperator(name, operatorMeta.getOperator());
+ Operator op = this.addOperator(name, operatorMeta.getOperator());
--- End diff --
Can LogicalPlan class provide a new method that allows to add Operator with existing OperatorMeta? Something like
```
private <T extends Operator> T addOperator(final String name, final T operator, final OperatorMeta meta)
{
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48589935
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -1359,7 +1409,8 @@ private void addDAGToCurrentDAG(ModuleMeta moduleMeta)
String name;
for (OperatorMeta operatorMeta : subDag.getAllOperators()) {
name = subDAGName + MODULE_NAMESPACE_SEPARATOR + operatorMeta.getName();
- this.addOperator(name, operatorMeta.getOperator());
+ Operator op = this.addOperator(name, operatorMeta.getOperator());
--- End diff --
We are adding an operator with a different name to a different DAG and as result the new meta object is created. Naturally that's a new object vs. sharing the meta object between DAGs... The meta object is result of the add operation and the name field should remain final. I don't see the need to add a copy constructor, setting attributes is public API.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48571279
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -1359,7 +1409,8 @@ private void addDAGToCurrentDAG(ModuleMeta moduleMeta)
String name;
for (OperatorMeta operatorMeta : subDag.getAllOperators()) {
name = subDAGName + MODULE_NAMESPACE_SEPARATOR + operatorMeta.getName();
- this.addOperator(name, operatorMeta.getOperator());
+ Operator op = this.addOperator(name, operatorMeta.getOperator());
+ this.getMeta(op).copyFrom(operatorMeta);
OperatorMeta operatorMetaNew = this.getOperatorMeta(name);
--- End diff --
Is operatormeta returned by `this.getMeta(op)` different from `this.getOperatorMeta(name)`? Can you not use `operatorMetaNew.copyFrom(operatorMeta)`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48647933
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -1359,7 +1409,8 @@ private void addDAGToCurrentDAG(ModuleMeta moduleMeta)
String name;
for (OperatorMeta operatorMeta : subDag.getAllOperators()) {
name = subDAGName + MODULE_NAMESPACE_SEPARATOR + operatorMeta.getName();
- this.addOperator(name, operatorMeta.getOperator());
+ Operator op = this.addOperator(name, operatorMeta.getOperator());
+ this.getMeta(op).copyFrom(operatorMeta);
OperatorMeta operatorMetaNew = this.getOperatorMeta(name);
--- End diff --
Yes, they are same. I have done the change to reuse operatorMetaNew
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48647937
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -907,6 +912,51 @@ else if (field.getType() == float.class || field.getType() == Float.class ||
getValue(OperatorContext.METRICS_DIMENSIONS_SCHEME));
}
+ /**
+ * Copy attribute from source attributeMap to destination attributeMap.
+ *
+ * @param dest destination attribute map.
+ * @param source source attribute map.
+ */
+ private void copyAttributes(AttributeMap dest, AttributeMap source)
+ {
+ for (Entry<Attribute<?>, ?> a : source.entrySet()) {
+ dest.put((Attribute<Object>)a.getKey(), a.getValue());
+ }
+ }
+
+ /**
+ * Copy attribute of operator and port from provided operatorMeta. This function requires
+ * operatorMeta argument is for the same operator.
+ *
+ * @param operatorMeta copy attribute from this OperatorMeta to the object.
+ */
+ private void copyFrom(OperatorMeta operatorMeta)
--- End diff --
changed name to copyAttributesFrom
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48570005
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -1374,7 +1425,8 @@ private void addDAGToCurrentDAG(ModuleMeta moduleMeta)
InputPort[] inputPorts = ports.toArray(new InputPort[]{});
name = subDAGName + MODULE_NAMESPACE_SEPARATOR + streamMeta.getName();
- this.addStream(name, sourceMeta.getPortObject(), inputPorts);
+ StreamMeta streamMetaNew = this.addStream(name, sourceMeta.getPortObject(), inputPorts);
+ streamMetaNew.copyFrom(streamMeta);
--- End diff --
instead of copyFrom can we do
`streamMetaNew.setLocality(streamMeta.getLocality());`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-apex-core/pull/191
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48648830
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -1359,7 +1409,8 @@ private void addDAGToCurrentDAG(ModuleMeta moduleMeta)
String name;
for (OperatorMeta operatorMeta : subDag.getAllOperators()) {
name = subDAGName + MODULE_NAMESPACE_SEPARATOR + operatorMeta.getName();
- this.addOperator(name, operatorMeta.getOperator());
+ Operator op = this.addOperator(name, operatorMeta.getOperator());
--- End diff --
Yes, the module DAG is kept after adding it's operator to the final DAG.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48554311
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -1359,7 +1409,8 @@ private void addDAGToCurrentDAG(ModuleMeta moduleMeta)
String name;
for (OperatorMeta operatorMeta : subDag.getAllOperators()) {
name = subDAGName + MODULE_NAMESPACE_SEPARATOR + operatorMeta.getName();
- this.addOperator(name, operatorMeta.getOperator());
+ Operator op = this.addOperator(name, operatorMeta.getOperator());
--- End diff --
@tweise should we drop "final" qualifier from name and id to avoid the complexity of copying from the original OperatorMeta? If not, I suggest introducing copy constructor for OperatorMeta that takes original OperatorMeta as a parameter and encapsulates the copy operation.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48569034
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -907,6 +912,51 @@ else if (field.getType() == float.class || field.getType() == Float.class ||
getValue(OperatorContext.METRICS_DIMENSIONS_SCHEME));
}
+ /**
+ * Copy attribute from source attributeMap to destination attributeMap.
+ *
+ * @param dest destination attribute map.
+ * @param source source attribute map.
+ */
+ private void copyAttributes(AttributeMap dest, AttributeMap source)
+ {
+ for (Entry<Attribute<?>, ?> a : source.entrySet()) {
+ dest.put((Attribute<Object>)a.getKey(), a.getValue());
+ }
+ }
+
+ /**
+ * Copy attribute of operator and port from provided operatorMeta. This function requires
+ * operatorMeta argument is for the same operator.
+ *
+ * @param operatorMeta copy attribute from this OperatorMeta to the object.
+ */
+ private void copyFrom(OperatorMeta operatorMeta)
+ {
+ if (operator != operatorMeta.getOperator()) {
+ throw new IllegalArgumentException("Operator meta is not for the same operator ");
+ }
+
+ // copy operator attributes
+ copyAttributes(attributes, operatorMeta.getAttributes());
+
+ // copy Input port attributes
+ for (Map.Entry<InputPort<?>, InputPortMeta> entry : operatorMeta.getPortMapping().inPortMap.entrySet()) {
+ InputPort<?> key = entry.getKey();
+ InputPortMeta dest = getPortMapping().inPortMap.get(key);
+ InputPortMeta source = entry.getValue();
--- End diff --
same you don't need source declared. Directly use it in copyAttributes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48589840
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -907,6 +912,51 @@ else if (field.getType() == float.class || field.getType() == Float.class ||
getValue(OperatorContext.METRICS_DIMENSIONS_SCHEME));
}
+ /**
+ * Copy attribute from source attributeMap to destination attributeMap.
+ *
+ * @param dest destination attribute map.
+ * @param source source attribute map.
+ */
+ private void copyAttributes(AttributeMap dest, AttributeMap source)
--- End diff --
Shouldn't this be putAll on AttributeMap?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48647943
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -907,6 +912,51 @@ else if (field.getType() == float.class || field.getType() == Float.class ||
getValue(OperatorContext.METRICS_DIMENSIONS_SCHEME));
}
+ /**
+ * Copy attribute from source attributeMap to destination attributeMap.
+ *
+ * @param dest destination attribute map.
+ * @param source source attribute map.
+ */
+ private void copyAttributes(AttributeMap dest, AttributeMap source)
--- End diff --
AttributeMap does not have pullAll method. Do you want me to add this in AttributeMap?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by tushargosavi <gi...@git.apache.org>.
Github user tushargosavi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48647935
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -1374,7 +1425,8 @@ private void addDAGToCurrentDAG(ModuleMeta moduleMeta)
InputPort[] inputPorts = ports.toArray(new InputPort[]{});
name = subDAGName + MODULE_NAMESPACE_SEPARATOR + streamMeta.getName();
- this.addStream(name, sourceMeta.getPortObject(), inputPorts);
+ StreamMeta streamMetaNew = this.addStream(name, sourceMeta.getPortObject(), inputPorts);
+ streamMetaNew.copyFrom(streamMeta);
--- End diff --
Done ..
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: APEXCORE-272 copy operator and p...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/191#discussion_r48568967
--- Diff: engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java ---
@@ -907,6 +912,51 @@ else if (field.getType() == float.class || field.getType() == Float.class ||
getValue(OperatorContext.METRICS_DIMENSIONS_SCHEME));
}
+ /**
+ * Copy attribute from source attributeMap to destination attributeMap.
+ *
+ * @param dest destination attribute map.
+ * @param source source attribute map.
+ */
+ private void copyAttributes(AttributeMap dest, AttributeMap source)
+ {
+ for (Entry<Attribute<?>, ?> a : source.entrySet()) {
+ dest.put((Attribute<Object>)a.getKey(), a.getValue());
+ }
+ }
+
+ /**
+ * Copy attribute of operator and port from provided operatorMeta. This function requires
+ * operatorMeta argument is for the same operator.
+ *
+ * @param operatorMeta copy attribute from this OperatorMeta to the object.
+ */
+ private void copyFrom(OperatorMeta operatorMeta)
+ {
+ if (operator != operatorMeta.getOperator()) {
+ throw new IllegalArgumentException("Operator meta is not for the same operator ");
+ }
+
+ // copy operator attributes
+ copyAttributes(attributes, operatorMeta.getAttributes());
+
+ // copy Input port attributes
+ for (Map.Entry<InputPort<?>, InputPortMeta> entry : operatorMeta.getPortMapping().inPortMap.entrySet()) {
+ InputPort<?> key = entry.getKey();
+ InputPortMeta dest = getPortMapping().inPortMap.get(key);
--- End diff --
You don't need key variable. You can directly do `getPortMapping().inPortMap.get(entry.getKey())`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---