Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/311#discussion_r23163896
--- Diff:
flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputSemanticProperties.java
---
@@ -56,218 +56,135 @@
public DualInputSemanticProperties() {
- init();
+ this.fieldMapping1 = new HashMap<Integer,FieldSet>();
+ this.fieldMapping2 = new HashMap<Integer,FieldSet>();
+ this.readFields1 = null;
+ this.readFields2 = null;
}
-
- /**
- * Adds, to the existing information, a field that is forwarded directly
- * from the source record(s) in the first input to the destination
- * record(s).
- *
- * @param sourceField the position in the source record(s) from the
first input
- * @param destinationField the position in the destination record(s)
- */
- public void addForwardedField1(int sourceField, int destinationField) {
- FieldSet old = this.forwardedFields1.get(sourceField);
- if (old == null) {
- old = FieldSet.EMPTY_SET;
+
+ @Override
+ public FieldSet getTargetFields(int input, int sourceField) {
+
+ if (input != 0 && input != 1) {
+ throw new IndexOutOfBoundsException();
+ } else if (input == 0) {
+
+ return fieldMapping1.get(sourceField);
+ } else {
+ return fieldMapping2.get(sourceField);
}
-
- FieldSet fs = old.addField(destinationField);
- this.forwardedFields1.put(sourceField, fs);
}
-
- /**
- * Adds, to the existing information, a field that is forwarded directly
- * from the source record(s) in the first input to multiple fields in
- * the destination record(s).
- *
- * @param sourceField the position in the source record(s)
- * @param destinationFields the position in the destination record(s)
- */
- public void addForwardedField1(int sourceField, FieldSet
destinationFields) {
- FieldSet old = this.forwardedFields1.get(sourceField);
- if (old == null) {
- old = FieldSet.EMPTY_SET;
+
+ @Override
+ public int getSourceField(int input, int targetField) {
+ Map<Integer, FieldSet> fieldMapping;
+
+ if (input != 0 && input != 1) {
+ throw new IndexOutOfBoundsException();
+ } else if (input == 0) {
+ fieldMapping = fieldMapping1;
+ } else {
+ fieldMapping = fieldMapping2;
+ }
+
+ for (int sourceField : fieldMapping.keySet()) {
--- End diff --
Good point
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
