You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by pradeepdalvi <gi...@git.apache.org> on 2016/05/11 16:19:12 UTC
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2082 Data Filter Op...
GitHub user pradeepdalvi opened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/270
APEXMALHAR-2082 Data Filter Operator
- Added Data Filter Operator
- Marked it as Evolving
- Added test cases around variations
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/pradeepdalvi/incubator-apex-malhar master
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/270.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 #270
----
commit 5c7a28aa24bdd3ab005779cbe2e28a2a10f6bb1f
Author: Pradeep A. Dalvi <pr...@datatorrent.com>
Date: 2016-05-11T16:06:02Z
APEXMALHAR-2082 Data Filter Operator
- Added Data Filter Operator
- Marked it as Evolving
- Added test cases around variations
----
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/270
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272110
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
--- End diff --
Please following this order of access:
1) default private
2) Protected if there is a valid usecases where this class can be overridden and that variable/function will need to be accessed from subclass.
3) Public - Avoid as much as possible IMO unless its a constant.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272112
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
--- End diff --
Should be private
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r64454657
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,210 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ private transient Class<?> inClazz = null;
+ private transient Expression<Boolean> expr = null;
+ private List<String> expressionFunctions = new LinkedList<>();
+
+ @AutoMetric
+ private long trueTuples;
+
+ @AutoMetric
+ private long falseTuples;
+
+ @AutoMetric
+ private long errorTuples;
+
+ protected String condition;
--- End diff --
Can this be made private? You already have setter and getter for this.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63493468
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
--- 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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272134
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
+ @AutoMetric
+ protected long trueTuples;
+
+ @AutoMetric
+ protected long falseTuples;
--- End diff --
If its a anticipation under which these variables are made protected, I suggest to make them private now and later on need basis change the access modifier.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63494078
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
+ {
+ filter.inClazz = DummyPrivatePOJO.class;
+ filter.setCondition("({$}.getVal() == 100)");
+ filter.activate(null);
+
+ /* when condition is true */
+ logger.debug("start round 0");
+ filter.beginWindow(0);
+
+ data.setVal(100);
+ filter.input.process(data);
--- 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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r64316099
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,210 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ private transient Class<?> inClazz = null;
+ private transient Expression<Boolean> expr = null;
+ private List<String> expressionFunctions = new LinkedList<>();
+
+ @AutoMetric
+ private long trueTuples;
+
+ @AutoMetric
+ private long falseTuples;
+
+ @AutoMetric
+ private long errorTuples;
+
+ protected String condition;
+
+ public final transient DefaultOutputPort<Object> truePort = new DefaultOutputPort<Object>();
+
+ public final transient DefaultOutputPort<Object> falsePort = new DefaultOutputPort<Object>();
+
+ public final transient DefaultOutputPort<Object> error = new DefaultOutputPort<Object>();
+
+ public FilterOperator()
+ {
+ expressionFunctions.add("java.lang.Math.*");
+ expressionFunctions.add("org.apache.commons.lang3.StringUtils.*");
+ expressionFunctions.add("org.apache.commons.lang3.StringEscapeUtils.*");
+ expressionFunctions.add("org.apache.commons.lang3.time.DurationFormatUtils.*");
+ expressionFunctions.add("org.apache.commons.lang3.time.DateFormatUtils.*");
+ }
+
+ @InputPortFieldAnnotation(schemaRequired = true)
+ public transient DefaultInputPort<Object> input = new DefaultInputPort<Object>()
--- End diff --
Please make this port final.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272116
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
--- End diff --
Yes it should be transient.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272281
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
+ {
+ filter.inClazz = DummyPrivatePOJO.class;
+ filter.setCondition("({$}.getVal() == 100)");
+ filter.activate(null);
+
+ /* when condition is true */
+ logger.debug("start round 0");
+ filter.beginWindow(0);
+
+ data.setVal(100);
+ filter.input.process(data);
+ Assert.assertEquals("true tuples", 1, filter.trueTuples);
+ Assert.assertEquals("false tuples", 0, filter.falseTuples);
+
+ filter.endWindow();
+ logger.debug("end round 0");
+
+ /* when condition is not true */
+ logger.debug("start round 1");
+ filter.beginWindow(1);
+
+ data.setVal(1000);
+ filter.input.process(data);
--- End diff --
filter.input.put(data);
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272121
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
+ @AutoMetric
+ protected long trueTuples;
--- End diff --
No need of this to be protected.. Please make it private.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63498339
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
+ @AutoMetric
+ protected long trueTuples;
+
+ @AutoMetric
+ protected long falseTuples;
+
+ @AutoMetric
+ protected long errorTuples;
+
+ @InputPortFieldAnnotation(schemaRequired = true)
+ public transient DefaultInputPort<Object> input = new DefaultInputPort<Object>()
+ {
+ public void setup(PortContext context)
+ {
+ inClazz = context.getValue(Context.PortContext.TUPLE_CLASS);
+ }
+
+ @Override
+ public void process(Object t)
+ {
+ processTuple(t);
+ }
+ };
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> truePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> falsePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> error = new DefaultOutputPort<Object>();
+
+ @Override
+ public void activate(Context context)
+ {
+ createExpression();
+ }
+
+ @Override
+ public void deactivate()
+ {
+ expr = null;
--- 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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r62939466
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
--- End diff --
Shouldn't expr and inClazz be marked as transient?
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63496694
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
+ {
+ filter.inClazz = DummyPrivatePOJO.class;
+ filter.setCondition("({$}.getVal() == 100)");
+ filter.activate(null);
+
+ /* when condition is true */
+ logger.debug("start round 0");
+ filter.beginWindow(0);
+
+ data.setVal(100);
+ filter.input.process(data);
+ Assert.assertEquals("true tuples", 1, filter.trueTuples);
+ Assert.assertEquals("false tuples", 0, filter.falseTuples);
+
+ filter.endWindow();
+ logger.debug("end round 0");
+
+ /* when condition is not true */
+ logger.debug("start round 1");
+ filter.beginWindow(1);
+
+ data.setVal(1000);
+ filter.input.process(data);
+ Assert.assertEquals("true tuples", 0, filter.trueTuples);
+ Assert.assertEquals("false tuples", 1, filter.falseTuples);
+
+ filter.endWindow();
+ logger.debug("end round 1");
+
+ filter.deactivate();
+ }
+
+ @Test
+ public void testFilterPublic()
+ {
+ filter.inClazz = DummyPublicPOJO.class;
+ filter.setCondition("({$}.val == 100)");
+ filter.activate(null);
--- 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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63507344
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
+ @AutoMetric
+ protected long trueTuples;
--- End diff --
These AutoMetrics were being referred in Unit Tests. Made them private and changed Unit Tests to refer CountTestSink.getCount()
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272276
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
+ {
+ filter.inClazz = DummyPrivatePOJO.class;
+ filter.setCondition("({$}.getVal() == 100)");
+ filter.activate(null);
+
+ /* when condition is true */
+ logger.debug("start round 0");
+ filter.beginWindow(0);
+
+ data.setVal(100);
+ filter.input.process(data);
+ Assert.assertEquals("true tuples", 1, filter.trueTuples);
+ Assert.assertEquals("false tuples", 0, filter.falseTuples);
+
+ filter.endWindow();
+ logger.debug("end round 0");
--- End diff --
Try not to log in test cases... You can have as much asserts as you want.. But that should indicate the information and not the logging.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63497690
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
+ {
+ filter.inClazz = DummyPrivatePOJO.class;
+ filter.setCondition("({$}.getVal() == 100)");
--- End diff --
This test is specifying Private member variable in condition. {$}.val raises following exception: Private member cannot be accessed from type "SC".
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63493173
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
+ @AutoMetric
+ protected long trueTuples;
+
+ @AutoMetric
+ protected long falseTuples;
+
+ @AutoMetric
+ protected long errorTuples;
+
+ @InputPortFieldAnnotation(schemaRequired = true)
+ public transient DefaultInputPort<Object> input = new DefaultInputPort<Object>()
+ {
+ public void setup(PortContext context)
+ {
+ inClazz = context.getValue(Context.PortContext.TUPLE_CLASS);
+ }
+
+ @Override
+ public void process(Object t)
+ {
+ processTuple(t);
+ }
+ };
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> truePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> falsePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> error = new DefaultOutputPort<Object>();
+
+ @Override
+ public void activate(Context context)
+ {
+ createExpression();
+ }
+
+ @Override
+ public void deactivate()
+ {
+ expr = null;
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ errorTuples = trueTuples = falseTuples = 0;
+ }
+
+ /**
+ * createExpression: create an expression from condition of POJO fields
+ * Override this function for custom field expressions
+ */
+ protected void createExpression()
+ {
+ logger.info("Creating an expression for condition {}", condition);
+ expr = PojoUtils.createExpression(inClazz, condition, Boolean.class);
+ }
+
+ /**
+ * evalExpression: Evaluate condition/expression
+ * Override this function for custom condition evaluation
+ */
+ protected Boolean evalExpression(Object t)
+ {
+ return expr.execute(t);
+ }
+
+ /**
+ * handleFilter: emit POJO meeting condition on truePort
+ * and if it did not meet condition then on falsePort
+ */
+ private void processTuple(Object t)
+ {
+ try {
+ if (evalExpression(t)) {
+ truePort.emit(t);
+ trueTuples++;
+ } else {
+ falsePort.emit(t);
+ falseTuples++;
+ }
+ } catch (Exception ex) {
+ logger.error("Error in expression eval: {}", ex.getMessage());
+ logger.debug("Exception: ", ex);
+ error.emit(t);
+ errorTuples++;
+ }
+ }
+
+ /**
+ * setCondition: setter method for condition parameter
+ */
+ public void setCondition(String condition)
+ {
+ logger.info("Changing condition from {} to {}", this.condition, condition);
--- End diff --
Conditions shall not change frequently and this debug message would immediately help understand that the condition has been changed at Operator level.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r64455136
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,210 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ private transient Class<?> inClazz = null;
+ private transient Expression<Boolean> expr = null;
+ private List<String> expressionFunctions = new LinkedList<>();
+
+ @AutoMetric
+ private long trueTuples;
+
+ @AutoMetric
+ private long falseTuples;
+
+ @AutoMetric
+ private long errorTuples;
+
+ protected String condition;
+
+ public final transient DefaultOutputPort<Object> truePort = new DefaultOutputPort<Object>();
+
+ public final transient DefaultOutputPort<Object> falsePort = new DefaultOutputPort<Object>();
+
+ public final transient DefaultOutputPort<Object> error = new DefaultOutputPort<Object>();
+
+ public FilterOperator()
+ {
+ expressionFunctions.add("java.lang.Math.*");
+ expressionFunctions.add("org.apache.commons.lang3.StringUtils.*");
+ expressionFunctions.add("org.apache.commons.lang3.StringEscapeUtils.*");
+ expressionFunctions.add("org.apache.commons.lang3.time.DurationFormatUtils.*");
+ expressionFunctions.add("org.apache.commons.lang3.time.DateFormatUtils.*");
+ }
+
+ @InputPortFieldAnnotation(schemaRequired = true)
+ public transient DefaultInputPort<Object> input = new DefaultInputPort<Object>()
+ {
+ public void setup(PortContext context)
+ {
+ inClazz = context.getValue(Context.PortContext.TUPLE_CLASS);
+ }
+
+ @Override
+ public void process(Object t)
+ {
+ processTuple(t);
+ }
+ };
+
+ @Override
+ public void activate(Context context)
+ {
+ createExpression();
+ }
+
+ @Override
+ public void deactivate()
+ {
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ errorTuples = trueTuples = falseTuples = 0;
+ }
+
+ /**
+ * createExpression: create an expression from condition of POJO fields
+ * Override this function for custom field expressions
+ */
+ protected void createExpression()
+ {
+ logger.info("Creating an expression for condition {}", condition);
+ expr = PojoUtils.createExpression(inClazz, condition, Boolean.class,
+ expressionFunctions.toArray(new String[expressionFunctions.size()]));
+ }
+
+ /**
+ * evalExpression: Evaluate condition/expression
+ * Override this function for custom condition evaluation
+ */
+ protected Boolean evalExpression(Object t)
+ {
+ return expr.execute(t);
+ }
+
+ /**
+ * handleFilter: emit POJO meeting condition on truePort
+ * and if it did not meet condition then on falsePort
+ */
+ private void processTuple(Object t)
+ {
+ try {
+ if (evalExpression(t)) {
+ truePort.emit(t);
+ trueTuples++;
+ } else {
+ falsePort.emit(t);
+ falseTuples++;
+ }
+ } catch (Exception ex) {
+ logger.error("Error in expression eval: {}", ex.getMessage());
+ logger.debug("Exception: ", ex);
+ error.emit(t);
+ errorTuples++;
+ }
+ }
+
+ /**
+ * Returns condition/expression with which Filtering is done
+ *
+ * @return Condition parameter of Filter Operator
--- End diff --
C in Condition should be small. Otherwise javadoc won't be able to identify it.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272300
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
+ {
+ filter.inClazz = DummyPrivatePOJO.class;
+ filter.setCondition("({$}.getVal() == 100)");
+ filter.activate(null);
+
+ /* when condition is true */
+ logger.debug("start round 0");
+ filter.beginWindow(0);
+
+ data.setVal(100);
+ filter.input.process(data);
+ Assert.assertEquals("true tuples", 1, filter.trueTuples);
+ Assert.assertEquals("false tuples", 0, filter.falseTuples);
+
+ filter.endWindow();
+ logger.debug("end round 0");
+
+ /* when condition is not true */
+ logger.debug("start round 1");
+ filter.beginWindow(1);
+
+ data.setVal(1000);
+ filter.input.process(data);
+ Assert.assertEquals("true tuples", 0, filter.trueTuples);
+ Assert.assertEquals("false tuples", 1, filter.falseTuples);
+
+ filter.endWindow();
+ logger.debug("end round 1");
+
+ filter.deactivate();
+ }
+
+ @Test
+ public void testFilterPublic()
+ {
+ filter.inClazz = DummyPublicPOJO.class;
+ filter.setCondition("({$}.val == 100)");
+ filter.activate(null);
--- End diff --
Please call setup and tear down of the operator.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272310
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
--- End diff --
Can you please add applicationTest for Filter operator?
This is an important test to see whether serialization and de-serialization happens correctlly.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63494101
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
+ {
+ filter.inClazz = DummyPrivatePOJO.class;
+ filter.setCondition("({$}.getVal() == 100)");
+ filter.activate(null);
+
+ /* when condition is true */
+ logger.debug("start round 0");
+ filter.beginWindow(0);
+
+ data.setVal(100);
+ filter.input.process(data);
+ Assert.assertEquals("true tuples", 1, filter.trueTuples);
+ Assert.assertEquals("false tuples", 0, filter.falseTuples);
+
+ filter.endWindow();
+ logger.debug("end round 0");
+
+ /* when condition is not true */
+ logger.debug("start round 1");
+ filter.beginWindow(1);
+
+ data.setVal(1000);
+ filter.input.process(data);
--- 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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272286
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
--- End diff --
Please have setup and teardown also called for this object.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272222
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
+ @AutoMetric
+ protected long trueTuples;
+
+ @AutoMetric
+ protected long falseTuples;
+
+ @AutoMetric
+ protected long errorTuples;
+
+ @InputPortFieldAnnotation(schemaRequired = true)
+ public transient DefaultInputPort<Object> input = new DefaultInputPort<Object>()
+ {
+ public void setup(PortContext context)
+ {
+ inClazz = context.getValue(Context.PortContext.TUPLE_CLASS);
+ }
+
+ @Override
+ public void process(Object t)
+ {
+ processTuple(t);
+ }
+ };
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> truePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> falsePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> error = new DefaultOutputPort<Object>();
+
+ @Override
+ public void activate(Context context)
+ {
+ createExpression();
+ }
+
+ @Override
+ public void deactivate()
+ {
+ expr = null;
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ errorTuples = trueTuples = falseTuples = 0;
+ }
+
+ /**
+ * createExpression: create an expression from condition of POJO fields
+ * Override this function for custom field expressions
+ */
+ protected void createExpression()
+ {
+ logger.info("Creating an expression for condition {}", condition);
+ expr = PojoUtils.createExpression(inClazz, condition, Boolean.class);
+ }
+
+ /**
+ * evalExpression: Evaluate condition/expression
+ * Override this function for custom condition evaluation
+ */
+ protected Boolean evalExpression(Object t)
+ {
+ return expr.execute(t);
+ }
+
+ /**
+ * handleFilter: emit POJO meeting condition on truePort
+ * and if it did not meet condition then on falsePort
+ */
+ private void processTuple(Object t)
+ {
+ try {
+ if (evalExpression(t)) {
+ truePort.emit(t);
+ trueTuples++;
+ } else {
+ falsePort.emit(t);
+ falseTuples++;
+ }
+ } catch (Exception ex) {
+ logger.error("Error in expression eval: {}", ex.getMessage());
+ logger.debug("Exception: ", ex);
--- End diff --
I'll suggest to put the exception also in log.error. In production environment, the app won't be run with debug level and in such case, please it might be hard to reproduce prod issues, hence its best to get all error information right away.
With exception being logged in debug, most important information about error is logs. ex.message, often doesn't have too much information.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r64455896
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,210 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
--- End diff --
spelling: language.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272237
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
--- End diff --
You need to add one more variable here.. Please check Transformation Operator.. There is a variable called as expressionFunctions... Idea is this variable is to allow user to add any static method in classpath directly usable in expression. Please add such a variable.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63492795
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
+ @AutoMetric
+ protected long trueTuples;
+
+ @AutoMetric
+ protected long falseTuples;
+
+ @AutoMetric
+ protected long errorTuples;
+
+ @InputPortFieldAnnotation(schemaRequired = true)
+ public transient DefaultInputPort<Object> input = new DefaultInputPort<Object>()
+ {
+ public void setup(PortContext context)
+ {
+ inClazz = context.getValue(Context.PortContext.TUPLE_CLASS);
+ }
+
+ @Override
+ public void process(Object t)
+ {
+ processTuple(t);
+ }
+ };
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> truePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> falsePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> error = new DefaultOutputPort<Object>();
+
+ @Override
+ public void activate(Context context)
+ {
+ createExpression();
+ }
+
+ @Override
+ public void deactivate()
+ {
+ expr = null;
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ errorTuples = trueTuples = falseTuples = 0;
+ }
+
+ /**
+ * createExpression: create an expression from condition of POJO fields
+ * Override this function for custom field expressions
+ */
+ protected void createExpression()
+ {
+ logger.info("Creating an expression for condition {}", condition);
+ expr = PojoUtils.createExpression(inClazz, condition, Boolean.class);
+ }
+
+ /**
+ * evalExpression: Evaluate condition/expression
+ * Override this function for custom condition evaluation
+ */
+ protected Boolean evalExpression(Object t)
+ {
+ return expr.execute(t);
+ }
+
+ /**
+ * handleFilter: emit POJO meeting condition on truePort
+ * and if it did not meet condition then on falsePort
+ */
+ private void processTuple(Object t)
+ {
+ try {
+ if (evalExpression(t)) {
+ truePort.emit(t);
+ trueTuples++;
+ } else {
+ falsePort.emit(t);
+ falseTuples++;
+ }
+ } catch (Exception ex) {
+ logger.error("Error in expression eval: {}", ex.getMessage());
+ logger.debug("Exception: ", ex);
--- End diff --
Basically the idea is to limit extended error messages in data path. This exception is on incorrect tuples, hence stacktrace would probably fill logs quickly if we start hitting the issue and make it hard to analyze debug logs. Hence we have added single line message in error and if required entire stacktrace can be retrieved in debug mode.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63493400
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
--- End diff --
Changed the order.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63509136
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
--- End diff --
Added expressionFunctions
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63493350
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
--- End diff --
Changed both to transient.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272261
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
+ {
+ filter.inClazz = DummyPrivatePOJO.class;
+ filter.setCondition("({$}.getVal() == 100)");
--- End diff --
$.val should also work.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272242
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
+ @AutoMetric
+ protected long trueTuples;
+
+ @AutoMetric
+ protected long falseTuples;
+
+ @AutoMetric
+ protected long errorTuples;
+
+ @InputPortFieldAnnotation(schemaRequired = true)
+ public transient DefaultInputPort<Object> input = new DefaultInputPort<Object>()
+ {
+ public void setup(PortContext context)
+ {
+ inClazz = context.getValue(Context.PortContext.TUPLE_CLASS);
+ }
+
+ @Override
+ public void process(Object t)
+ {
+ processTuple(t);
+ }
+ };
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> truePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> falsePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> error = new DefaultOutputPort<Object>();
+
+ @Override
+ public void activate(Context context)
+ {
+ createExpression();
+ }
+
+ @Override
+ public void deactivate()
+ {
+ expr = null;
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ errorTuples = trueTuples = falseTuples = 0;
+ }
+
+ /**
+ * createExpression: create an expression from condition of POJO fields
+ * Override this function for custom field expressions
+ */
+ protected void createExpression()
+ {
+ logger.info("Creating an expression for condition {}", condition);
+ expr = PojoUtils.createExpression(inClazz, condition, Boolean.class);
+ }
+
+ /**
+ * evalExpression: Evaluate condition/expression
+ * Override this function for custom condition evaluation
+ */
+ protected Boolean evalExpression(Object t)
+ {
+ return expr.execute(t);
+ }
+
+ /**
+ * handleFilter: emit POJO meeting condition on truePort
+ * and if it did not meet condition then on falsePort
+ */
+ private void processTuple(Object t)
+ {
+ try {
+ if (evalExpression(t)) {
+ truePort.emit(t);
+ trueTuples++;
+ } else {
+ falsePort.emit(t);
+ falseTuples++;
+ }
+ } catch (Exception ex) {
+ logger.error("Error in expression eval: {}", ex.getMessage());
+ logger.debug("Exception: ", ex);
+ error.emit(t);
+ errorTuples++;
+ }
+ }
+
+ /**
+ * setCondition: setter method for condition parameter
+ */
+ public void setCondition(String condition)
+ {
+ logger.info("Changing condition from {} to {}", this.condition, condition);
--- End diff --
No logging required. There are ample ways in platform to know what is the value of property.
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272266
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
+ {
+ filter.inClazz = DummyPrivatePOJO.class;
+ filter.setCondition("({$}.getVal() == 100)");
+ filter.activate(null);
+
+ /* when condition is true */
+ logger.debug("start round 0");
+ filter.beginWindow(0);
+
+ data.setVal(100);
+ filter.input.process(data);
--- End diff --
instead of directly calling process, please do filter.input.put(data)
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63272174
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
+ @AutoMetric
+ protected long trueTuples;
+
+ @AutoMetric
+ protected long falseTuples;
+
+ @AutoMetric
+ protected long errorTuples;
+
+ @InputPortFieldAnnotation(schemaRequired = true)
+ public transient DefaultInputPort<Object> input = new DefaultInputPort<Object>()
+ {
+ public void setup(PortContext context)
+ {
+ inClazz = context.getValue(Context.PortContext.TUPLE_CLASS);
+ }
+
+ @Override
+ public void process(Object t)
+ {
+ processTuple(t);
+ }
+ };
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> truePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> falsePort = new DefaultOutputPort<Object>();
+
+ @OutputPortFieldAnnotation(schemaRequired = true)
+ public final transient DefaultOutputPort<Object> error = new DefaultOutputPort<Object>();
+
+ @Override
+ public void activate(Context context)
+ {
+ createExpression();
+ }
+
+ @Override
+ public void deactivate()
+ {
+ expr = null;
--- End diff --
No need of this..
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63686579
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
--- End diff --
Added FilterAppTest
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63507383
--- Diff: library/src/main/java/com/datatorrent/lib/filter/FilterOperator.java ---
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.expression.Expression;
+import com.datatorrent.lib.util.PojoUtils;
+
+/**
+ * <b>FilterOperator</b>
+ * Filter Operator filter out tuples based on defined condition
+ *
+ * <b>Parameters</b>
+ * - condition: condition based on expression langugage
+ *
+ * <b>Input Port</b> takes POJOs as an input
+ *
+ * <b>Output Ports</b>
+ * - truePort emits POJOs meeting the given condition
+ * - falsePort emits POJOs not meeting the given condition
+ * - error port emits any error situation while evaluating expression
+ *
+ */
+@InterfaceStability.Evolving
+public class FilterOperator extends BaseOperator implements Operator.ActivationListener
+{
+ protected String condition;
+
+ protected Class<?> inClazz = null;
+
+ private Expression<Boolean> expr = null;
+
+ @AutoMetric
+ protected long trueTuples;
+
+ @AutoMetric
+ protected long falseTuples;
--- End diff --
Made AutoMetrics as private
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63497136
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
+ {
+ filter.inClazz = DummyPrivatePOJO.class;
+ filter.setCondition("({$}.getVal() == 100)");
+ filter.activate(null);
+
+ /* when condition is true */
+ logger.debug("start round 0");
+ filter.beginWindow(0);
+
+ data.setVal(100);
+ filter.input.process(data);
+ Assert.assertEquals("true tuples", 1, filter.trueTuples);
+ Assert.assertEquals("false tuples", 0, filter.falseTuples);
+
+ filter.endWindow();
+ logger.debug("end round 0");
--- End diff --
Removed debug logs
---
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-malhar pull request: APEXMALHAR-2082 Data Filter Op...
Posted by pradeepdalvi <gi...@git.apache.org>.
Github user pradeepdalvi commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/270#discussion_r63496712
--- Diff: library/src/test/java/com/datatorrent/lib/filter/FilterTest.java ---
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.datatorrent.lib.filter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for FilterOperator
+ */
+public class FilterTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilterTest.class);
+
+ public static class DummyPrivatePOJO
+ {
+ private long val;
+
+ public long getVal()
+ {
+ return val;
+ }
+
+ public void setVal(long val)
+ {
+ this.val = val;
+ }
+ }
+
+ public static class DummyPublicPOJO
+ {
+ public long val;
+ }
+
+
+ private static FilterOperator filter;
+ private static DummyPrivatePOJO data;
+ private static DummyPublicPOJO pdata;
+
+ @Test
+ public void testFilterPrivate()
--- 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.
---