diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/BADConstants.java b/asterix-bad/src/main/java/org/apache/asterix/bad/BADConstants.java
index 391d84e..7178db1 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/BADConstants.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/BADConstants.java
@@ -1,18 +1,18 @@
 /*
  * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
+ * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
+ * 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
+ * with the License.  You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *   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
+ * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/extension/BADMetadataExtension.java b/asterix-bad/src/main/java/org/apache/asterix/bad/extension/BADMetadataExtension.java
index d0639e2..28f7a79 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/extension/BADMetadataExtension.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/extension/BADMetadataExtension.java
@@ -135,6 +135,7 @@
         BuiltinFunctions.addFunction(BADFunctions.CURRENT_CHANNEL_TIME, ADateTimeTypeComputer.INSTANCE, false);
         BuiltinFunctions.addFunction(BADFunctions.PREVIOUS_CHANNEL_TIME, ADateTimeTypeComputer.INSTANCE, false);
         BuiltinFunctions.addFunction(BADFunctions.IS_NEW, ABooleanTypeComputer.INSTANCE, true);
+        BuiltinFunctions.addFunction(BADFunctions.ACTIVE_TIMESTAMP, ADateTimeTypeComputer.INSTANCE, true);
 
         // to shadow the master feed rewriter
         BuiltinFunctions.addPrivateFunction(BuiltinFunctions.FEED_COLLECT, BADFeedRewriter.INSTANCE, true);
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryInsertOperatorNodePushable.java b/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryInsertOperatorNodePushable.java
index 813fd0b..3239584 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryInsertOperatorNodePushable.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryInsertOperatorNodePushable.java
@@ -44,13 +44,15 @@
 
     @Override
     protected void beforeModification(ITupleReference tuple) {
-        if ((tuple.getFieldCount() == 3
-                && tuple.getFieldData(0)[tuple.getFieldStart(2)] == ATypeTag.SERIALIZED_RECORD_TYPE_TAG)
-                || (tuple.getFieldCount() == 4
-                        && tuple.getFieldData(0)[tuple.getFieldStart(2)] == ATypeTag.SERIALIZED_RECORD_TYPE_TAG)) {
-            int targetIdx = tuple.getFieldStart(2) + 14;
+        if (tuple.getFieldCount() >= 3
+                && tuple.getFieldData(0)[tuple.getFieldStart(2)] == ATypeTag.SERIALIZED_RECORD_TYPE_TAG
+                && tuple.getFieldLength(2) == 22) {
+            long currMilli = System.currentTimeMillis();
             ByteBuffer tupleBuff = ByteBuffer.wrap(tuple.getFieldData(0));
-            tupleBuff.putLong(targetIdx, System.currentTimeMillis());
+            tupleBuff.putLong(tuple.getFieldStart(2) + 14, currMilli);
+            if (tuple.getFieldCount() == 4) {
+                tupleBuff.putLong(tuple.getFieldStart(3) + 1, currMilli);
+            }
         }
     }
 }
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryUpsertOperatorDescriptor.java b/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryUpsertOperatorDescriptor.java
index 46f9bc8..0f1ce21 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryUpsertOperatorDescriptor.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryUpsertOperatorDescriptor.java
@@ -37,11 +37,11 @@
             IMissingWriterFactory missingWriterFactory,
             IModificationOperationCallbackFactory modificationOpCallbackFactory,
             ISearchOperationCallbackFactory searchOpCallbackFactory,
-            IFrameOperationCallbackFactory frameOpCallbackFactory, int numPrimaryKeys, ARecordType recordType,
-            int filterIndex, boolean hasSecondaries) {
+            IFrameOperationCallbackFactory frameOpCallbackFactory, int numPrimaryKeys, Integer filterSourceIndicator,
+            ARecordType itemType, int filterIndex, boolean hasSecondaries) {
         super(spec, outRecDesc, fieldPermutation, indexHelperFactory, missingWriterFactory,
                 modificationOpCallbackFactory, searchOpCallbackFactory, frameOpCallbackFactory, numPrimaryKeys,
-                recordType, filterIndex, hasSecondaries);
+                filterSourceIndicator, itemType, filterIndex, hasSecondaries);
     }
 
     @Override
@@ -49,7 +49,7 @@
             IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException {
         RecordDescriptor intputRecDesc = recordDescProvider.getInputRecordDescriptor(getActivityId(), 0);
         return new BADLSMPrimaryUpsertOperatorNodePushable(ctx, partition, indexHelperFactory, fieldPermutation,
-                intputRecDesc, modCallbackFactory, searchOpCallbackFactory, numPrimaryKeys, recordType, filterIndex,
-                frameOpCallbackFactory, missingWriterFactory, hasSecondaries);
+                intputRecDesc, modCallbackFactory, searchOpCallbackFactory, numPrimaryKeys, filterSourceIndicator,
+                filterItemType, filterIndex, frameOpCallbackFactory, missingWriterFactory, hasSecondaries);
     }
 }
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryUpsertOperatorNodePushable.java b/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryUpsertOperatorNodePushable.java
index b0c79d5..fc5ccde 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryUpsertOperatorNodePushable.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/feed/operators/BADLSMPrimaryUpsertOperatorNodePushable.java
@@ -36,23 +36,25 @@
     public BADLSMPrimaryUpsertOperatorNodePushable(IHyracksTaskContext ctx, int partition,
             IIndexDataflowHelperFactory indexHelperFactory, int[] fieldPermutation, RecordDescriptor inputRecDesc,
             IModificationOperationCallbackFactory modCallbackFactory,
-            ISearchOperationCallbackFactory searchCallbackFactory, int numOfPrimaryKeys, ARecordType recordType,
-            int filterFieldIndex, IFrameOperationCallbackFactory frameOpCallbackFactory,
+            ISearchOperationCallbackFactory searchCallbackFactory, int numOfPrimaryKeys, Integer filterSourceIndicator,
+            ARecordType recordType, int filterFieldIndex, IFrameOperationCallbackFactory frameOpCallbackFactory,
             IMissingWriterFactory missingWriterFactory, boolean hasSecondaries) throws HyracksDataException {
         super(ctx, partition, indexHelperFactory, fieldPermutation, inputRecDesc, modCallbackFactory,
-                searchCallbackFactory, numOfPrimaryKeys, recordType, filterFieldIndex, frameOpCallbackFactory,
-                missingWriterFactory, hasSecondaries);
+                searchCallbackFactory, numOfPrimaryKeys, filterSourceIndicator, recordType, filterFieldIndex,
+                frameOpCallbackFactory, missingWriterFactory, hasSecondaries);
     }
 
     @Override
     protected void beforeModification(ITupleReference tuple) {
-        if ((tuple.getFieldCount() == 3
-                && tuple.getFieldData(0)[tuple.getFieldStart(2)] == ATypeTag.SERIALIZED_RECORD_TYPE_TAG)
-                || (tuple.getFieldCount() == 4
-                        && tuple.getFieldData(0)[tuple.getFieldStart(2)] == ATypeTag.SERIALIZED_RECORD_TYPE_TAG)) {
-            int targetIdx = tuple.getFieldStart(2) + 14;
+        if (tuple.getFieldCount() >= 3
+                && tuple.getFieldData(0)[tuple.getFieldStart(2)] == ATypeTag.SERIALIZED_RECORD_TYPE_TAG
+                && tuple.getFieldLength(2) == 22) {
+            long currMilli = System.currentTimeMillis();
             ByteBuffer tupleBuff = ByteBuffer.wrap(tuple.getFieldData(0));
-            tupleBuff.putLong(targetIdx, System.currentTimeMillis());
+            tupleBuff.putLong(tuple.getFieldStart(2) + 14, currMilli);
+            if (tuple.getFieldCount() == 4) {
+                tupleBuff.putLong(tuple.getFieldStart(3) + 1, currMilli);
+            }
         }
     }
 }
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctionCollection.java b/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctionCollection.java
index 15bb617..e6d2309 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctionCollection.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctionCollection.java
@@ -1,22 +1,20 @@
 /*
+ * 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
  *
- *  * 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.
+ * 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 org.apache.asterix.bad.function;
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctionRegistrant.java b/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctionRegistrant.java
index 15a5fc5..5dfb62d 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctionRegistrant.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctionRegistrant.java
@@ -1,22 +1,20 @@
 /*
+ * 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
  *
- *  * 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.
+ * 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 org.apache.asterix.bad.function;
 
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctions.java b/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctions.java
index a342210..085bd80 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctions.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/function/BADFunctions.java
@@ -1,22 +1,20 @@
 /*
+ * 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
  *
- *  * 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.
+ *   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 org.apache.asterix.bad.function;
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/function/runtime/CurrentChannelTimeDescriptor.java b/asterix-bad/src/main/java/org/apache/asterix/bad/function/runtime/CurrentChannelTimeDescriptor.java
index f3392ad..90191a3 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/function/runtime/CurrentChannelTimeDescriptor.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/function/runtime/CurrentChannelTimeDescriptor.java
@@ -1,22 +1,20 @@
 /*
+ * 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
  *
- *  * 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.
+ *   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 org.apache.asterix.bad.function.runtime;
 
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/function/runtime/PreviousChannelTimeDescriptor.java b/asterix-bad/src/main/java/org/apache/asterix/bad/function/runtime/PreviousChannelTimeDescriptor.java
index f4dd1f5..800c3c4 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/function/runtime/PreviousChannelTimeDescriptor.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/function/runtime/PreviousChannelTimeDescriptor.java
@@ -1,22 +1,20 @@
 /*
+ * 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
  *
- *  * 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.
+ *   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 org.apache.asterix.bad.function.runtime;
 
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/BADExpressionToPlanTranslator.java b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/BADExpressionToPlanTranslator.java
index 23d10c5..abcd70d 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/BADExpressionToPlanTranslator.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/BADExpressionToPlanTranslator.java
@@ -27,6 +27,7 @@
 import org.apache.asterix.lang.common.struct.VarIdentifier;
 import org.apache.asterix.metadata.declared.DatasetDataSource;
 import org.apache.asterix.metadata.declared.MetadataProvider;
+import org.apache.asterix.metadata.utils.DatasetUtil;
 import org.apache.asterix.om.base.ADateTime;
 import org.apache.asterix.om.base.ARecord;
 import org.apache.asterix.om.base.IAObject;
@@ -72,32 +73,43 @@
 
     @Override
     protected ILogicalOperator translateDelete(DatasetDataSource targetDatasource, Mutable<ILogicalExpression> varRef,
-            List<Mutable<ILogicalExpression>> varRefsForLoading,
-            List<Mutable<ILogicalExpression>> additionalFilteringExpressions, ILogicalOperator inputOp,
+            List<Mutable<ILogicalExpression>> varRefsForLoading, LogicalVariable seqVar, ILogicalOperator pkeyAssignOp,
             CompiledStatements.ICompiledDmlStatement stmt) throws AlgebricksException {
         SourceLocation sourceLoc = stmt.getSourceLocation();
         InsertDeleteUpsertOperator deleteOp;
+        LogicalVariable metaVar = null;
+
         if (!targetDatasource.getDataset().hasMetaPart()) {
             deleteOp = new InsertDeleteUpsertOperator(targetDatasource, varRef, varRefsForLoading,
                     InsertDeleteUpsertOperator.Kind.DELETE, false);
         } else {
             // prepare meta record
             IAType metaType = metadataProvider.findMetaType(targetDatasource.getDataset());
-            LogicalVariable metaVar = context.newVar();
+            metaVar = context.newVar();
             AssignOperator metaVariableAssignOp =
                     new AssignOperator(metaVar, new MutableObject<>(makeMetaRecordExpr(metaType)));
-            metaVariableAssignOp.getInputs().add(new MutableObject<>(inputOp));
+            metaVariableAssignOp.getInputs().add(new MutableObject<>(pkeyAssignOp));
             metaVariableAssignOp.setSourceLocation(sourceLoc);
             // create insert op uses meta record
             deleteOp = new InsertDeleteUpsertOperator(targetDatasource, varRef, varRefsForLoading,
                     Collections.singletonList(new MutableObject<>(new VariableReferenceExpression(metaVar))),
                     InsertDeleteUpsertOperator.Kind.DELETE, false);
             // change current inputOp to be meta op
-            inputOp = metaVariableAssignOp;
+            pkeyAssignOp = metaVariableAssignOp;
         }
-        deleteOp.setAdditionalFilteringExpressions(additionalFilteringExpressions);
-        deleteOp.getInputs().add(new MutableObject<>(inputOp));
+
+        deleteOp.getInputs().add(new MutableObject<>(pkeyAssignOp));
         deleteOp.setSourceLocation(sourceLoc);
+
+        List<String> filterField = DatasetUtil.getFilterField(targetDatasource.getDataset());
+        List<Mutable<ILogicalExpression>> filterExprs = null;
+        Integer filterSourceIndicator = DatasetUtil.getFilterSourceIndicator(targetDatasource.getDataset());
+        if (filterField != null) {
+            filterExprs = generatedFilterExprs(deleteOp, filterField, filterSourceIndicator == 0 ? seqVar : metaVar,
+                    sourceLoc);
+        }
+        deleteOp.setAdditionalFilteringExpressions(filterExprs);
+
         DelegateOperator leafOperator = new DelegateOperator(new CommitOperator(true));
         leafOperator.getInputs().add(new MutableObject<>(deleteOp));
         leafOperator.setSourceLocation(sourceLoc);
@@ -105,23 +117,23 @@
     }
 
     @Override
-    protected ILogicalOperator translateUpsert(DatasetDataSource targetDatasource, Mutable<ILogicalExpression> varRef,
-            List<Mutable<ILogicalExpression>> varRefsForLoading, List<Mutable<ILogicalExpression>> filterExprs,
-            ILogicalOperator pkeyAssignOp, List<String> additionalFilteringField, LogicalVariable unnestVar,
-            ILogicalOperator topOp, List<Mutable<ILogicalExpression>> exprs, LogicalVariable resVar,
-            AssignOperator additionalFilteringAssign, CompiledStatements.ICompiledDmlStatement stmt,
-            IResultMetadata resultMetadata) throws AlgebricksException {
+    protected ILogicalOperator translateUpsert(DatasetDataSource targetDatasource,
+            Mutable<ILogicalExpression> payloadVarRef, List<Mutable<ILogicalExpression>> varRefsForLoading,
+            ILogicalOperator pkeyAssignOp, LogicalVariable unnestVar, ILogicalOperator topOp,
+            List<Mutable<ILogicalExpression>> pkeyExprs, LogicalVariable seqVar,
+            CompiledStatements.ICompiledDmlStatement stmt, IResultMetadata resultMetadata) throws AlgebricksException {
         SourceLocation sourceLoc = stmt.getSourceLocation();
         CompiledStatements.CompiledUpsertStatement compiledUpsert = (CompiledStatements.CompiledUpsertStatement) stmt;
         Expression returnExpression = compiledUpsert.getReturnExpression();
         InsertDeleteUpsertOperator upsertOp;
         ILogicalOperator rootOperator;
+        LogicalVariable metaVar = null;
 
         ARecordType recordType = (ARecordType) targetDatasource.getItemType();
 
         if (targetDatasource.getDataset().hasMetaPart()) {
             IAType metaType = metadataProvider.findMetaType(targetDatasource.getDataset());
-            LogicalVariable metaVar = context.newVar();
+            metaVar = context.newVar();
             AssignOperator metaVariableAssignOp =
                     new AssignOperator(metaVar, new MutableObject<>(makeMetaRecordExpr(metaType)));
             metaVariableAssignOp.getInputs().add(new MutableObject<>(pkeyAssignOp));
@@ -131,7 +143,7 @@
             List<Mutable<ILogicalExpression>> metaExprs = new ArrayList<>(1);
             VariableReferenceExpression metaVarRef = new VariableReferenceExpression(metaVar);
             metaExprs.add(new MutableObject<>(metaVarRef));
-            upsertOp = new InsertDeleteUpsertOperator(targetDatasource, varRef, varRefsForLoading, metaExprs,
+            upsertOp = new InsertDeleteUpsertOperator(targetDatasource, payloadVarRef, varRefsForLoading, metaExprs,
                     InsertDeleteUpsertOperator.Kind.UPSERT, false);
 
             // set previous meta vars
@@ -142,23 +154,33 @@
             metaTypes.add(targetDatasource.getMetaItemType());
             upsertOp.setPrevAdditionalNonFilteringTypes(metaTypes);
         } else {
-            upsertOp = new InsertDeleteUpsertOperator(targetDatasource, varRef, varRefsForLoading,
+            upsertOp = new InsertDeleteUpsertOperator(targetDatasource, payloadVarRef, varRefsForLoading,
                     InsertDeleteUpsertOperator.Kind.UPSERT, false);
-            // Create and add a new variable used for representing the original record
-            if (additionalFilteringField != null) {
-                upsertOp.setPrevFilterVar(context.newVar());
-                upsertOp.setPrevFilterType(recordType.getFieldType(additionalFilteringField.get(0)));
-            }
         }
+
         // Create and add a new variable used for representing the original record
         upsertOp.setUpsertIndicatorVar(context.newVar());
         upsertOp.setUpsertIndicatorVarType(BuiltinType.ABOOLEAN);
         upsertOp.setPrevRecordVar(context.newVar());
         upsertOp.setPrevRecordType(recordType);
         upsertOp.setSourceLocation(sourceLoc);
-        upsertOp.setAdditionalFilteringExpressions(filterExprs);
         upsertOp.getInputs().add(new MutableObject<>(pkeyAssignOp));
 
+        List<String> filterField = DatasetUtil.getFilterField(targetDatasource.getDataset());
+        List<Mutable<ILogicalExpression>> filterExprs = null;
+        Integer filterSourceIndicator = DatasetUtil.getFilterSourceIndicator(targetDatasource.getDataset());
+        if (filterField != null) {
+            filterExprs = generatedFilterExprs(upsertOp, filterField, filterSourceIndicator == 0 ? seqVar : metaVar,
+                    sourceLoc);
+            ARecordType filterSourceType = filterSourceIndicator == 0 ? (ARecordType) targetDatasource.getItemType()
+                    : (ARecordType) targetDatasource.getMetaItemType();
+            upsertOp.setAdditionalFilteringExpressions(filterExprs);
+            upsertOp.setPrevFilterVar(context.newVar());
+            upsertOp.setPrevFilterType(filterSourceType.getFieldType(filterField.get(0)));
+        } else {
+            upsertOp.setAdditionalFilteringExpressions(null);
+        }
+
         // Set up delegate operator
         DelegateOperator delegateOperator = new DelegateOperator(new CommitOperator(returnExpression == null));
         delegateOperator.getInputs().add(new MutableObject<>(upsertOp));
@@ -171,10 +193,10 @@
 
     @Override
     protected ILogicalOperator translateInsert(DatasetDataSource targetDatasource, Mutable<ILogicalExpression> varRef,
-            List<Mutable<ILogicalExpression>> varRefsForLoading, List<Mutable<ILogicalExpression>> filterExprs,
-            ILogicalOperator inputOp, CompiledStatements.ICompiledDmlStatement stmt, IResultMetadata resultMetadata)
-            throws AlgebricksException {
+            List<Mutable<ILogicalExpression>> varRefsForLoading, LogicalVariable seqVar, ILogicalOperator pkeyAssignOp,
+            CompiledStatements.ICompiledDmlStatement stmt, IResultMetadata resultMetadata) throws AlgebricksException {
         SourceLocation sourceLoc = stmt.getSourceLocation();
+        LogicalVariable metaVar = null;
 
         InsertDeleteUpsertOperator insertOp;
         if (!targetDatasource.getDataset().hasMetaPart()) {
@@ -183,21 +205,30 @@
         } else {
             // prepare meta record
             IAType metaType = metadataProvider.findMetaType(targetDatasource.getDataset());
-            LogicalVariable metaVar = context.newVar();
+            metaVar = context.newVar();
             AssignOperator metaVariableAssignOp =
                     new AssignOperator(metaVar, new MutableObject<>(makeMetaRecordExpr(metaType)));
-            metaVariableAssignOp.getInputs().add(new MutableObject<>(inputOp));
+            metaVariableAssignOp.getInputs().add(new MutableObject<>(pkeyAssignOp));
             metaVariableAssignOp.setSourceLocation(sourceLoc);
             // create insert op uses meta record
             insertOp = new InsertDeleteUpsertOperator(targetDatasource, varRef, varRefsForLoading,
                     Collections.singletonList(new MutableObject<>(new VariableReferenceExpression(metaVar))),
                     InsertDeleteUpsertOperator.Kind.INSERT, false);
             // change current inputOp to be meta op
-            inputOp = metaVariableAssignOp;
+            pkeyAssignOp = metaVariableAssignOp;
+        }
+
+        insertOp.getInputs().add(new MutableObject<>(pkeyAssignOp));
+        insertOp.setSourceLocation(sourceLoc);
+
+        List<String> filterField = DatasetUtil.getFilterField(targetDatasource.getDataset());
+        List<Mutable<ILogicalExpression>> filterExprs = null;
+        Integer filterSourceIndicator = DatasetUtil.getFilterSourceIndicator(targetDatasource.getDataset());
+        if (filterField != null) {
+            filterExprs = generatedFilterExprs(insertOp, filterField, filterSourceIndicator == 0 ? seqVar : metaVar,
+                    sourceLoc);
         }
         insertOp.setAdditionalFilteringExpressions(filterExprs);
-        insertOp.getInputs().add(new MutableObject<>(inputOp));
-        insertOp.setSourceLocation(sourceLoc);
 
         // Adds the commit operator.
         CompiledStatements.CompiledInsertStatement compiledInsert = (CompiledStatements.CompiledInsertStatement) stmt;
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/BADRuleSetFactory.java b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/BADRuleSetFactory.java
index 3c3ec59..be93512 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/BADRuleSetFactory.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/BADRuleSetFactory.java
@@ -22,6 +22,7 @@
 import java.util.List;
 
 import org.apache.asterix.bad.rules.InsertBrokerNotifierForChannelRule;
+import org.apache.asterix.bad.rules.RewriteActiveFunctionsRule;
 import org.apache.asterix.bad.rules.RewriteChannelTimeFunctionToLocalVarRule;
 import org.apache.asterix.common.dataflow.ICcApplicationContext;
 import org.apache.asterix.compiler.provider.DefaultRuleSetFactory;
@@ -92,6 +93,7 @@
             if (rule instanceof FeedScanCollectionToUnnest) {
                 alteredRuleCollection.add(i + 1, new MetaFunctionToMetaVariableRule());
                 alteredRuleCollection.add(i + 1, new RewriteChannelTimeFunctionToLocalVarRule());
+                alteredRuleCollection.add(i + 1, new RewriteActiveFunctionsRule());
             }
         }
 
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/AbstractCreateChannelStatement.java b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/AbstractCreateChannelStatement.java
index 43f11fd..943e0a4 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/AbstractCreateChannelStatement.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/AbstractCreateChannelStatement.java
@@ -136,7 +136,7 @@
         List<String> fieldNames = new ArrayList<>();
         fieldNames.add(BADConstants.SubscriptionId);
         partitionFields.add(fieldNames);
-        IDatasetDetailsDecl idd = new InternalDetailsDecl(partitionFields, keyIndicators, true, null);
+        IDatasetDetailsDecl idd = new InternalDetailsDecl(partitionFields, keyIndicators, true, null, null);
         TypeExpression subItemType = new TypeReferenceExpression(
                 new Pair<>(MetadataConstants.METADATA_DATAVERSE_NAME, subscriptionsTypeName));
         DatasetDecl createSubscriptionsDataset = new DatasetDecl(dataverseName, new Identifier(subscriptionsTableName),
@@ -151,7 +151,7 @@
             fieldNames = new ArrayList<>();
             fieldNames.add(BADConstants.ResultId);
             partitionFields.add(fieldNames);
-            idd = new InternalDetailsDecl(partitionFields, keyIndicators, true, null);
+            idd = new InternalDetailsDecl(partitionFields, keyIndicators, true, null, null);
             TypeExpression resultItemType =
                     new TypeReferenceExpression(new Pair<>(MetadataConstants.METADATA_DATAVERSE_NAME, resultsTypeName));
             DatasetDecl createResultsDataset = new DatasetDecl(dataverseName, new Identifier(resultsTableName),
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataProvider.java b/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataProvider.java
index 290db49..5a1e151 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataProvider.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataProvider.java
@@ -120,9 +120,13 @@
         }
         // add the previous filter third
         int fieldIdx = -1;
+        Integer filterSourceIndicator = null;
+        ARecordType filterItemType = null;
         if (numFilterFields > 0) {
             String filterField = DatasetUtil.getFilterField(dataset).get(0);
-            String[] fieldNames = itemType.getFieldNames();
+            filterSourceIndicator = DatasetUtil.getFilterSourceIndicator(dataset);
+            filterItemType = filterSourceIndicator == 0 ? itemType : metaItemType;
+            String[] fieldNames = filterItemType.getFieldNames();
             int i = 0;
             for (; i < fieldNames.length; i++) {
                 if (fieldNames[i].equals(filterField)) {
@@ -130,9 +134,10 @@
                 }
             }
             fieldIdx = i;
-            outputTypeTraits[f] = dataFormat.getTypeTraitProvider().getTypeTrait(itemType.getFieldTypes()[fieldIdx]);
+            outputTypeTraits[f] =
+                    dataFormat.getTypeTraitProvider().getTypeTrait(filterItemType.getFieldTypes()[fieldIdx]);
             outputSerDes[f] =
-                    dataFormat.getSerdeProvider().getSerializerDeserializer(itemType.getFieldTypes()[fieldIdx]);
+                    dataFormat.getSerdeProvider().getSerializerDeserializer(filterItemType.getFieldTypes()[fieldIdx]);
             f++;
         }
         for (int j = 0; j < inputRecordDesc.getFieldCount(); j++) {
@@ -142,7 +147,8 @@
         RecordDescriptor outputRecordDesc = new RecordDescriptor(outputSerDes, outputTypeTraits);
         op = new BADLSMPrimaryUpsertOperatorDescriptor(spec, outputRecordDesc, fieldPermutation, idfh,
                 missingWriterFactory, modificationCallbackFactory, searchCallbackFactory,
-                dataset.getFrameOpCallbackFactory(metadataProvider), numKeys, itemType, fieldIdx, hasSecondaries);
+                dataset.getFrameOpCallbackFactory(metadataProvider), numKeys, filterSourceIndicator, filterItemType,
+                fieldIdx, hasSecondaries);
         return new Pair<>(op, splitsAndConstraint.second);
     }
 }
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/rules/RewriteActiveFunctionsRule.java b/asterix-bad/src/main/java/org/apache/asterix/bad/rules/RewriteActiveFunctionsRule.java
new file mode 100644
index 0000000..a58341f
--- /dev/null
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/rules/RewriteActiveFunctionsRule.java
@@ -0,0 +1,143 @@
+/*
+ * 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 org.apache.asterix.bad.rules;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.asterix.bad.BADConstants;
+import org.apache.asterix.bad.function.BADFunctions;
+import org.apache.asterix.lang.common.util.FunctionUtil;
+import org.apache.asterix.om.base.AString;
+import org.apache.asterix.om.constants.AsterixConstantValue;
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.commons.lang3.mutable.MutableObject;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalExpressionTag;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
+import org.apache.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
+import org.apache.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
+import org.apache.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
+import org.apache.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AssignOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.InnerJoinOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.SelectOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
+import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+
+public class RewriteActiveFunctionsRule implements IAlgebraicRewriteRule {
+
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
+        Mutable<ILogicalExpression> exprRef;
+        if (opRef.getValue().getOperatorTag() == LogicalOperatorTag.INNERJOIN) {
+            InnerJoinOperator selectOp = (InnerJoinOperator) opRef.getValue();
+            exprRef = selectOp.getCondition();
+        } else if (opRef.getValue().getOperatorTag() == LogicalOperatorTag.SELECT) {
+            SelectOperator selectOp = (SelectOperator) opRef.getValue();
+            exprRef = selectOp.getCondition();
+        } else {
+            return false;
+        }
+        return visit(exprRef, opRef, context);
+    }
+
+    private ScalarFunctionCallExpression makeActiveTsAccessExpr(LogicalVariable dsVar) {
+        ILogicalExpression activeTsFunc =
+                new ScalarFunctionCallExpression(FunctionUtil.getFunctionInfo(BuiltinFunctions.META),
+                        new MutableObject<>(new VariableReferenceExpression(dsVar)));
+        return new ScalarFunctionCallExpression(FunctionUtil.getFunctionInfo(BuiltinFunctions.FIELD_ACCESS_BY_NAME),
+                new MutableObject<>(activeTsFunc), new MutableObject<>(new ConstantExpression(
+                        new AsterixConstantValue(new AString(BADConstants.FIELD_NAME_ACTIVE_TS)))));
+    }
+
+    private AbstractFunctionCallExpression makeAndForIsNew(AbstractFunctionCallExpression funcExpr) {
+        LogicalVariable dsVar = getDsVar(funcExpr);
+        ILogicalExpression previousChannelTimeExpr = new ScalarFunctionCallExpression(
+                BuiltinFunctions.getBuiltinFunctionInfo(BADFunctions.PREVIOUS_CHANNEL_TIME),
+                new MutableObject<>(new VariableReferenceExpression(dsVar)));
+        ILogicalExpression currentChannelTimeExpr = new ScalarFunctionCallExpression(
+                BuiltinFunctions.getBuiltinFunctionInfo(BADFunctions.CURRENT_CHANNEL_TIME),
+                new MutableObject<>(new VariableReferenceExpression(dsVar)));
+        ILogicalExpression lessThanExpr = new ScalarFunctionCallExpression(
+                FunctionUtil.getFunctionInfo(AlgebricksBuiltinFunctions.LT),
+                new MutableObject<>(makeActiveTsAccessExpr(dsVar)), new MutableObject<>(currentChannelTimeExpr));
+        ILogicalExpression greaterThanExpr = new ScalarFunctionCallExpression(
+                FunctionUtil.getFunctionInfo(AlgebricksBuiltinFunctions.GT),
+                new MutableObject<>(makeActiveTsAccessExpr(dsVar)), new MutableObject<>(previousChannelTimeExpr));
+        return new ScalarFunctionCallExpression(FunctionUtil.getFunctionInfo(BuiltinFunctions.AND),
+                new MutableObject<>(lessThanExpr), new MutableObject<>(greaterThanExpr));
+    }
+
+    private LogicalVariable getDsVar(AbstractFunctionCallExpression funcExpr) {
+        return ((VariableReferenceExpression) funcExpr.getArguments().get(0).getValue()).getVariableReference();
+    }
+
+    private boolean visit(Mutable<ILogicalExpression> exprRef, Mutable<ILogicalOperator> opRef,
+            IOptimizationContext context) throws AlgebricksException {
+        boolean changed = false;
+        if (exprRef.getValue().getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL) {
+            AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) exprRef.getValue();
+
+            if (funcExpr.getFunctionIdentifier() == BADFunctions.IS_NEW) {
+                exprRef.setValue(makeAndForIsNew(funcExpr));
+                changed = true;
+            } else if (funcExpr.getFunctionIdentifier() == BADFunctions.ACTIVE_TIMESTAMP) {
+                LogicalVariable channelTimeVar = context.newVar();
+                LogicalVariable dsVar = getDsVar(funcExpr);
+                ILogicalExpression activeTsFunc =
+                        new ScalarFunctionCallExpression(FunctionUtil.getFunctionInfo(BuiltinFunctions.META),
+                                new MutableObject<>(new VariableReferenceExpression(dsVar)));
+                ScalarFunctionCallExpression faExpr = new ScalarFunctionCallExpression(
+                        FunctionUtil.getFunctionInfo(BuiltinFunctions.FIELD_ACCESS_BY_NAME),
+                        new MutableObject<>(activeTsFunc), new MutableObject<>(new ConstantExpression(
+                                new AsterixConstantValue(new AString(BADConstants.FIELD_NAME_ACTIVE_TS)))));
+
+                AssignOperator assignOp = new AssignOperator(channelTimeVar, new MutableObject<>(faExpr));
+
+                List<LogicalVariable> liveVars = new ArrayList<>();
+                for (int i = 0; i < opRef.getValue().getInputs().size(); i++) {
+                    Mutable<ILogicalOperator> inputOpRef = opRef.getValue().getInputs().get(i);
+                    VariableUtilities.getLiveVariables(inputOpRef.getValue(), liveVars);
+                    if (liveVars.contains(dsVar)) {
+                        assignOp.getInputs().add(new MutableObject<>(inputOpRef.getValue()));
+                        inputOpRef.setValue(assignOp);
+                        exprRef.setValue(new VariableReferenceExpression(channelTimeVar));
+                        context.computeAndSetTypeEnvironmentForOperator(assignOp);
+                    }
+                }
+
+                changed = true;
+            } else {
+                for (Mutable<ILogicalExpression> argExpr : funcExpr.getArguments()) {
+                    changed = changed || visit(argExpr, opRef, context);
+                }
+            }
+        }
+        return changed;
+    }
+
+}
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/rules/RewriteChannelTimeFunctionToLocalVarRule.java b/asterix-bad/src/main/java/org/apache/asterix/bad/rules/RewriteChannelTimeFunctionToLocalVarRule.java
index 56f5e39..1542d1e 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/rules/RewriteChannelTimeFunctionToLocalVarRule.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/rules/RewriteChannelTimeFunctionToLocalVarRule.java
@@ -1,22 +1,20 @@
 /*
+ * 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
  *
- *  * 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.
+ * 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 org.apache.asterix.bad.rules;
 
@@ -27,7 +25,6 @@
 
 import org.apache.asterix.bad.BADConstants;
 import org.apache.asterix.bad.function.BADFunctions;
-import org.apache.asterix.lang.common.util.FunctionUtil;
 import org.apache.asterix.metadata.declared.DataSource;
 import org.apache.asterix.om.base.AString;
 import org.apache.asterix.om.constants.AsterixConstantValue;
@@ -45,7 +42,6 @@
 import org.apache.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
 import org.apache.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
 import org.apache.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
-import org.apache.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AssignOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.DataSourceScanOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.InnerJoinOperator;
@@ -57,7 +53,6 @@
     @Override
     public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
             throws AlgebricksException {
-
         Mutable<ILogicalExpression> exprRef;
         if (opRef.getValue().getOperatorTag() == LogicalOperatorTag.INNERJOIN) {
             InnerJoinOperator selectOp = (InnerJoinOperator) opRef.getValue();
@@ -72,32 +67,30 @@
         Set<Mutable<ILogicalExpression>> activeFunctionSet = new HashSet<>();
         Set<LogicalVariable> needPrevDsSet = new HashSet<>();
         Set<LogicalVariable> needCurrDsSet = new HashSet<>();
-        Set<LogicalVariable> needActiveDsSet = new HashSet<>();
+        String channelName =
+                (String) context.getMetadataProvider().getConfig().getOrDefault(BADConstants.CONFIG_CHANNEL_NAME, "");
 
         // collect active functions
-        collectChannelTimeFunctions(exprRef, activeFunctionSet, needPrevDsSet, needCurrDsSet, needActiveDsSet);
+        boolean rewriteFunc = collectChannelTimeFunctions(exprRef, activeFunctionSet, needPrevDsSet, needCurrDsSet,
+                true, channelName);
         if (activeFunctionSet.size() == 0) {
-            return false;
+            return rewriteFunc;
         }
 
         // add assigns for active functions
         Map<LogicalVariable, LogicalVariable> prevMap = new HashMap<>();
         Map<LogicalVariable, LogicalVariable> currMap = new HashMap<>();
-        Map<LogicalVariable, LogicalVariable> activeMap = new HashMap<>();
-        createChannelTimeAssignOps(opRef, needPrevDsSet, needCurrDsSet, needActiveDsSet, prevMap, currMap, activeMap,
-                context);
+        createChannelTimeAssignOps(opRef, needPrevDsSet, needCurrDsSet, prevMap, currMap, context, channelName);
 
         // update expressions with new vars
-        updateActiveFuncExprsWithVars(prevMap, currMap, activeMap, activeFunctionSet);
+        updateActiveFuncExprsWithVars(prevMap, currMap, activeFunctionSet);
 
         context.computeAndSetTypeEnvironmentForOperator(opRef.getValue());
         return true;
     }
 
     private void updateActiveFuncExprsWithVars(Map<LogicalVariable, LogicalVariable> prevMap,
-            Map<LogicalVariable, LogicalVariable> currMap, Map<LogicalVariable, LogicalVariable> activeMap,
-            Set<Mutable<ILogicalExpression>> activeFuncExprs) {
-
+            Map<LogicalVariable, LogicalVariable> currMap, Set<Mutable<ILogicalExpression>> activeFuncExprs) {
         for (Mutable<ILogicalExpression> expr : activeFuncExprs) {
             AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) expr.getValue();
             LogicalVariable dsVar = ((VariableReferenceExpression) ((AbstractFunctionCallExpression) expr.getValue())
@@ -106,36 +99,20 @@
                 expr.setValue(new VariableReferenceExpression(currMap.get(dsVar)));
             } else if (funcExpr.getFunctionIdentifier() == BADFunctions.PREVIOUS_CHANNEL_TIME) {
                 expr.setValue(new VariableReferenceExpression(prevMap.get(dsVar)));
-            } else {
-                ILogicalExpression lessThanExpr =
-                        new ScalarFunctionCallExpression(FunctionUtil.getFunctionInfo(AlgebricksBuiltinFunctions.LT),
-                                new MutableObject<>(new VariableReferenceExpression(activeMap.get(dsVar))),
-                                new MutableObject<>(new VariableReferenceExpression(currMap.get(dsVar))));
-                ILogicalExpression greaterThanExpr =
-                        new ScalarFunctionCallExpression(FunctionUtil.getFunctionInfo(AlgebricksBuiltinFunctions.GT),
-                                new MutableObject<>(new VariableReferenceExpression(activeMap.get(dsVar))),
-                                new MutableObject<>(new VariableReferenceExpression(prevMap.get(dsVar))));
-                ScalarFunctionCallExpression andExpr =
-                        new ScalarFunctionCallExpression(FunctionUtil.getFunctionInfo(BuiltinFunctions.AND),
-                                new MutableObject<>(lessThanExpr), new MutableObject<>(greaterThanExpr));
-                expr.setValue(andExpr);
             }
         }
     }
 
     private void createChannelTimeAssignOps(Mutable<ILogicalOperator> opRef, Set<LogicalVariable> needPrevDsSet,
-            Set<LogicalVariable> needCurrDsSet, Set<LogicalVariable> needActiveDsSet,
-            Map<LogicalVariable, LogicalVariable> prevMap, Map<LogicalVariable, LogicalVariable> currMap,
-            Map<LogicalVariable, LogicalVariable> activeMap, IOptimizationContext context) {
+            Set<LogicalVariable> needCurrDsSet, Map<LogicalVariable, LogicalVariable> prevMap,
+            Map<LogicalVariable, LogicalVariable> currMap, IOptimizationContext context, String channelName) {
         ILogicalOperator currOp = opRef.getValue();
-        String channelName =
-                (String) context.getMetadataProvider().getConfig().getOrDefault(BADConstants.CONFIG_CHANNEL_NAME, "");
         if (opRef.getValue().getOperatorTag() == LogicalOperatorTag.DATASOURCESCAN) {
             DataSourceScanOperator dataScanOp = (DataSourceScanOperator) opRef.getValue();
             DataSource ds = (DataSource) dataScanOp.getDataSource();
             LogicalVariable dsVar = ds.getDataRecordVariable(dataScanOp.getScanVariables());
 
-            if (needPrevDsSet.contains(dsVar) || needActiveDsSet.contains(dsVar)) {
+            if (needPrevDsSet.contains(dsVar)) {
                 LogicalVariable channelTimeVar = context.newVar();
                 ILogicalExpression previousChannelTimeExpr = new ScalarFunctionCallExpression(
                         BuiltinFunctions.getBuiltinFunctionInfo(BADFunctions.PREVIOUS_CHANNEL_TIME),
@@ -148,7 +125,7 @@
                 prevMap.put(dsVar, channelTimeVar);
             }
 
-            if (needCurrDsSet.contains(dsVar) || needActiveDsSet.contains(dsVar)) {
+            if (needCurrDsSet.contains(dsVar)) {
                 LogicalVariable channelTimeVar = context.newVar();
                 ILogicalExpression previousChannelTimeExpr = new ScalarFunctionCallExpression(
                         BuiltinFunctions.getBuiltinFunctionInfo(BADFunctions.CURRENT_CHANNEL_TIME), new MutableObject<>(
@@ -159,55 +136,51 @@
                 opRef.setValue(assignOp);
                 currMap.put(dsVar, channelTimeVar);
             }
-
-            if (needActiveDsSet.contains(dsVar)) {
-                LogicalVariable channelTimeVar = context.newVar();
-
-                ILogicalExpression activeTsFunc =
-                        new ScalarFunctionCallExpression(FunctionUtil.getFunctionInfo(BuiltinFunctions.META),
-                                new MutableObject<>(new VariableReferenceExpression(dsVar)));
-                ScalarFunctionCallExpression faExpr = new ScalarFunctionCallExpression(
-                        FunctionUtil.getFunctionInfo(BuiltinFunctions.FIELD_ACCESS_BY_NAME),
-                        new MutableObject<>(activeTsFunc), new MutableObject<>(new ConstantExpression(
-                                new AsterixConstantValue(new AString(BADConstants.FIELD_NAME_ACTIVE_TS)))));
-                AssignOperator assignOp = new AssignOperator(channelTimeVar, new MutableObject<>(faExpr));
-                assignOp.getInputs().add(new MutableObject<>(opRef.getValue()));
-                opRef.setValue(assignOp);
-                activeMap.put(dsVar, channelTimeVar);
-            }
         }
         for (Mutable<ILogicalOperator> input : currOp.getInputs()) {
-            createChannelTimeAssignOps(input, needPrevDsSet, needCurrDsSet, needActiveDsSet, prevMap, currMap,
-                    activeMap, context);
+            createChannelTimeAssignOps(input, needPrevDsSet, needCurrDsSet, prevMap, currMap, context, channelName);
         }
     }
 
-    private void collectChannelTimeFunctions(Mutable<ILogicalExpression> exprRef,
+    private boolean collectChannelTimeFunctions(Mutable<ILogicalExpression> exprRef,
             Set<Mutable<ILogicalExpression>> activeFunctionSet, Set<LogicalVariable> needPrevDsSet,
-            Set<LogicalVariable> needCurrDsSet, Set<LogicalVariable> needActive) {
+            Set<LogicalVariable> needCurrDsSet, boolean disjunctiveFlag, String channelName) {
+        boolean rewriteFunc = false;
         if (exprRef.getValue().getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL) {
             AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) exprRef.getValue();
-            if (funcExpr.getFunctionIdentifier() == BADFunctions.IS_NEW
-                    || funcExpr.getFunctionIdentifier() == BADFunctions.PREVIOUS_CHANNEL_TIME
+            if (funcExpr.getFunctionIdentifier().equals(BuiltinFunctions.OR)) {
+                disjunctiveFlag = false;
+            }
+            if (funcExpr.getFunctionIdentifier() == BADFunctions.PREVIOUS_CHANNEL_TIME
                     || funcExpr.getFunctionIdentifier() == BADFunctions.CURRENT_CHANNEL_TIME) {
-                // add to active func set for later replacement
-                activeFunctionSet.add(exprRef);
                 // collect ds var to see what assign op needs to be added
-                LogicalVariable dsVar = ((VariableReferenceExpression) funcExpr.getArguments().get(0).getValue())
-                        .getVariableReference();
-                if (funcExpr.getFunctionIdentifier() == BADFunctions.PREVIOUS_CHANNEL_TIME) {
-                    needPrevDsSet.add(dsVar);
-                } else if (funcExpr.getFunctionIdentifier() == BADFunctions.CURRENT_CHANNEL_TIME) {
-                    needCurrDsSet.add(dsVar);
+                LogicalExpressionTag arg0ExprTag = funcExpr.getArguments().get(0).getValue().getExpressionTag();
+                if (arg0ExprTag == LogicalExpressionTag.CONSTANT) {
+                    return false;
+                }
+                if (!disjunctiveFlag) {
+                    LogicalVariable dsVar = ((VariableReferenceExpression) funcExpr.getArguments().get(0).getValue())
+                            .getVariableReference();
+                    activeFunctionSet.add(exprRef);
+                    if (funcExpr.getFunctionIdentifier() == BADFunctions.PREVIOUS_CHANNEL_TIME) {
+                        needPrevDsSet.add(dsVar);
+                    } else if (funcExpr.getFunctionIdentifier() == BADFunctions.CURRENT_CHANNEL_TIME) {
+                        needCurrDsSet.add(dsVar);
+                    }
                 } else {
-                    needActive.add(dsVar);
+                    // if disjunctive, modify to get ts and wait for introduce filter rule to work
+                    funcExpr.getArguments().set(0, new MutableObject<>(
+                            new ConstantExpression(new AsterixConstantValue(new AString(channelName)))));
+                    rewriteFunc = true;
                 }
             } else {
                 for (Mutable<ILogicalExpression> argExpr : funcExpr.getArguments()) {
-                    collectChannelTimeFunctions(argExpr, activeFunctionSet, needPrevDsSet, needCurrDsSet, needActive);
+                    rewriteFunc = rewriteFunc || collectChannelTimeFunctions(argExpr, activeFunctionSet, needPrevDsSet,
+                            needCurrDsSet, disjunctiveFlag, channelName);
                 }
             }
         }
+        return rewriteFunc;
     }
 
 }
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/ActiveTimestampManager.java b/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/ActiveTimestampManager.java
index adbbd03..0a24a72 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/ActiveTimestampManager.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/ActiveTimestampManager.java
@@ -1,22 +1,20 @@
 /*
+ * 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
  *
- *  * 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.
+ *   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 org.apache.asterix.bad.runtime;
 
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/ActiveTimestampState.java b/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/ActiveTimestampState.java
index ad7e424..9067073 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/ActiveTimestampState.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/ActiveTimestampState.java
@@ -1,22 +1,20 @@
 /*
+ * 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
  *
- *  * 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.
+ *   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 org.apache.asterix.bad.runtime;
 
diff --git a/asterix-bad/src/main/resources/lang-extension/lang.txt b/asterix-bad/src/main/resources/lang-extension/lang.txt
index 49f0cb4..f7eba70 100644
--- a/asterix-bad/src/main/resources/lang-extension/lang.txt
+++ b/asterix-bad/src/main/resources/lang-extension/lang.txt
@@ -144,12 +144,12 @@
     ( <HINTS> hints = Properties() )?
     ( <WITH> withRecord = RecordConstructor() )?
       {
-        // TODO: add filters on meta records
+        try{
         InternalDetailsDecl idd = new InternalDetailsDecl(primaryKeyFields.second,
                                                           primaryKeyFields.first,
                                                           autogenerated,
-                                                          null);
-        try{
+                                                          1,
+                                                          Collections.singletonList("_active_timestamp"));
         stmt = new DatasetDecl(nameComponents.first,
                                    nameComponents.second,
                                    datasetTypeExpr,
diff --git a/asterix-bad/src/test/resources/optimizerts/queries/bad_cq/new_tweets_active_ts.sqlpp b/asterix-bad/src/test/resources/optimizerts/queries/bad_cq/new_tweets_active_ts.sqlpp
new file mode 100644
index 0000000..86b0f28
--- /dev/null
+++ b/asterix-bad/src/test/resources/optimizerts/queries/bad_cq/new_tweets_active_ts.sqlpp
@@ -0,0 +1,40 @@
+/*
+ * 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.
+ */
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+USE test;
+
+CREATE TYPE Tweet AS OPEN {
+ tid: bigint,
+ area_code: string,
+ text: string,
+ location: point,
+ timestamp: datetime
+};
+
+
+CREATE ACTIVE DATASET Tweets(Tweet) PRIMARY KEY tid;
+
+write output to nc1:"rttest/new_tweets.sqlpp";
+set `compiler.indexonly` "false";
+
+CREATE CONTINUOUS CHANNEL NearbyTweets(oid) PERIOD duration("PT3S")
+{
+ SELECT t FROM Tweets t WHERE active_timestamp(t) > previous_channel_time(t) AND active_timestamp(t) < current_channel_time(t)
+};
\ No newline at end of file
diff --git a/asterix-bad/src/test/resources/optimizerts/queries/bad_cq/new_tweets_meta_and.sqlpp b/asterix-bad/src/test/resources/optimizerts/queries/bad_cq/new_tweets_meta_and.sqlpp
new file mode 100644
index 0000000..27184bf
--- /dev/null
+++ b/asterix-bad/src/test/resources/optimizerts/queries/bad_cq/new_tweets_meta_and.sqlpp
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+USE test;
+
+CREATE TYPE Tweet AS OPEN {
+ tid: bigint,
+ area_code: string,
+ text: string,
+ location: point,
+ timestamp: datetime
+};
+
+
+CREATE ACTIVE DATASET Tweets(Tweet) PRIMARY KEY tid;
+
+write output to nc1:"rttest/new_tweets.sqlpp";
+set `compiler.indexonly` "false";
+
+CREATE CONTINUOUS CHANNEL NearbyTweets(oid) PERIOD duration("PT3S")
+{
+ SELECT t FROM Tweets t WHERE meta(t).`_active_timestamp` > current_channel_time(t)
+ AND meta(t).`_active_timestamp` < previous_channel_time(t)
+};
diff --git a/asterix-bad/src/test/resources/optimizerts/queries/bad_cq/new_tweets_meta_or.sqlpp b/asterix-bad/src/test/resources/optimizerts/queries/bad_cq/new_tweets_meta_or.sqlpp
new file mode 100644
index 0000000..311caf0
--- /dev/null
+++ b/asterix-bad/src/test/resources/optimizerts/queries/bad_cq/new_tweets_meta_or.sqlpp
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+USE test;
+
+CREATE TYPE Tweet AS OPEN {
+ tid: bigint,
+ area_code: string,
+ text: string,
+ location: point,
+ timestamp: datetime
+};
+
+
+CREATE ACTIVE DATASET Tweets(Tweet) PRIMARY KEY tid;
+
+write output to nc1:"rttest/new_tweets.sqlpp";
+set `compiler.indexonly` "false";
+
+CREATE CONTINUOUS CHANNEL NearbyTweets(oid) PERIOD duration("PT3S")
+{
+ SELECT t FROM Tweets t WHERE meta(t).`_active_timestamp` > current_channel_time(t)
+ OR meta(t).`_active_timestamp` < previous_channel_time(t)
+};
diff --git a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_nearby_tweets.plan b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_nearby_tweets.plan
index e4c1ec7..617ce18 100644
--- a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_nearby_tweets.plan
+++ b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_nearby_tweets.plan
@@ -1,13 +1,13 @@
 -- NOTIFY_BROKERS  |PARTITIONED|
   -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-    -- PRE_CLUSTERED_GROUP_BY[$$120, $$121, $$channelExecutionTime]  |PARTITIONED|
+    -- PRE_CLUSTERED_GROUP_BY[$$118, $$119, $$channelExecutionTime]  |PARTITIONED|
             {
               -- AGGREGATE  |LOCAL|
                 -- NESTED_TUPLE_SOURCE  |LOCAL|
             }
       -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-        -- STABLE_SORT [$$120(ASC), $$121(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
-          -- HASH_PARTITION_EXCHANGE [$$120, $$121, $$channelExecutionTime]  |PARTITIONED|
+        -- STABLE_SORT [$$118(ASC), $$119(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
+          -- HASH_PARTITION_EXCHANGE [$$118, $$119, $$channelExecutionTime]  |PARTITIONED|
             -- PRE_SORTED_DISTINCT_BY  |PARTITIONED|
               -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                 -- STABLE_SORT [$$97(ASC)]  |PARTITIONED|
@@ -58,16 +58,14 @@
                                                                     -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                                                       -- STREAM_PROJECT  |PARTITIONED|
                                                                         -- STREAM_SELECT  |PARTITIONED|
-                                                                          -- ASSIGN  |PARTITIONED|
+                                                                          -- STREAM_PROJECT  |PARTITIONED|
                                                                             -- ASSIGN  |PARTITIONED|
-                                                                              -- ASSIGN  |PARTITIONED|
-                                                                                -- STREAM_PROJECT  |PARTITIONED|
-                                                                                  -- ASSIGN  |PARTITIONED|
-                                                                                    -- STREAM_PROJECT  |PARTITIONED|
-                                                                                      -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                                                                        -- DATASOURCE_SCAN  |PARTITIONED|
-                                                                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                                                                            -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                                                                              -- STREAM_PROJECT  |PARTITIONED|
+                                                                                -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                                  -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                                    -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                                                      -- ASSIGN  |PARTITIONED|
+                                                                                        -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
                                                                     -- BROADCAST_EXCHANGE  |PARTITIONED|
                                                                       -- STREAM_PROJECT  |PARTITIONED|
                                                                         -- ASSIGN  |PARTITIONED|
diff --git a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets.plan b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets.plan
index ebf5624..6134bd4 100644
--- a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets.plan
+++ b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets.plan
@@ -1,13 +1,13 @@
 -- NOTIFY_BROKERS  |PARTITIONED|
   -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-    -- PRE_CLUSTERED_GROUP_BY[$$93, $$94, $$channelExecutionTime]  |PARTITIONED|
+    -- PRE_CLUSTERED_GROUP_BY[$$91, $$92, $$channelExecutionTime]  |PARTITIONED|
             {
               -- AGGREGATE  |LOCAL|
                 -- NESTED_TUPLE_SOURCE  |LOCAL|
             }
       -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-        -- STABLE_SORT [$$93(ASC), $$94(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
-          -- HASH_PARTITION_EXCHANGE [$$93, $$94, $$channelExecutionTime]  |PARTITIONED|
+        -- STABLE_SORT [$$91(ASC), $$92(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
+          -- HASH_PARTITION_EXCHANGE [$$91, $$92, $$channelExecutionTime]  |PARTITIONED|
             -- PRE_SORTED_DISTINCT_BY  |PARTITIONED|
               -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                 -- STABLE_SORT [$$79(ASC)]  |PARTITIONED|
@@ -55,12 +55,11 @@
                                                               -- ASSIGN  |PARTITIONED|
                                                                 -- STREAM_PROJECT  |PARTITIONED|
                                                                   -- STREAM_SELECT  |PARTITIONED|
-                                                                    -- ASSIGN  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
                                                                       -- ASSIGN  |PARTITIONED|
                                                                         -- STREAM_PROJECT  |PARTITIONED|
-                                                                          -- ASSIGN  |PARTITIONED|
-                                                                            -- STREAM_PROJECT  |PARTITIONED|
-                                                                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                                                                -- DATASOURCE_SCAN  |PARTITIONED|
-                                                                                  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                                                                    -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                                                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                            -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                              -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                                                -- ASSIGN  |PARTITIONED|
+                                                                                  -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
diff --git a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_active_ts.plan b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_active_ts.plan
new file mode 100644
index 0000000..7ac1e96
--- /dev/null
+++ b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_active_ts.plan
@@ -0,0 +1,65 @@
+-- NOTIFY_BROKERS  |PARTITIONED|
+  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+    -- PRE_CLUSTERED_GROUP_BY[$$96, $$97, $$channelExecutionTime]  |PARTITIONED|
+            {
+              -- AGGREGATE  |LOCAL|
+                -- NESTED_TUPLE_SOURCE  |LOCAL|
+            }
+      -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+        -- STABLE_SORT [$$96(ASC), $$97(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
+          -- HASH_PARTITION_EXCHANGE [$$96, $$97, $$channelExecutionTime]  |PARTITIONED|
+            -- PRE_SORTED_DISTINCT_BY  |PARTITIONED|
+              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                -- STABLE_SORT [$$83(ASC)]  |PARTITIONED|
+                  -- HASH_PARTITION_EXCHANGE [$$83]  |PARTITIONED|
+                    -- STREAM_PROJECT  |PARTITIONED|
+                      -- COMMIT  |PARTITIONED|
+                        -- STREAM_PROJECT  |PARTITIONED|
+                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                            -- INDEX_INSERT_DELETE  |PARTITIONED|
+                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                -- STREAM_PROJECT  |PARTITIONED|
+                                  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                    -- INSERT_DELETE  |PARTITIONED|
+                                      -- HASH_PARTITION_EXCHANGE [$$77]  |PARTITIONED|
+                                        -- ASSIGN  |PARTITIONED|
+                                          -- STREAM_PROJECT  |PARTITIONED|
+                                            -- ASSIGN  |PARTITIONED|
+                                              -- STREAM_PROJECT  |PARTITIONED|
+                                                -- ASSIGN  |PARTITIONED|
+                                                  -- STREAM_PROJECT  |PARTITIONED|
+                                                    -- ASSIGN  |PARTITIONED|
+                                                      -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                        -- NESTED_LOOP  |PARTITIONED|
+                                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                            -- STREAM_PROJECT  |PARTITIONED|
+                                                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                -- HYBRID_HASH_JOIN [$$90, $$88][$$84, $$85]  |PARTITIONED|
+                                                                  -- HASH_PARTITION_EXCHANGE [$$90, $$88]  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                          -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                            -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                                              -- ASSIGN  |UNPARTITIONED|
+                                                                                -- EMPTY_TUPLE_SOURCE  |UNPARTITIONED|
+                                                                  -- HASH_PARTITION_EXCHANGE [$$84, $$85]  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                          -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                            -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                              -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                                                          -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                            -- STREAM_PROJECT  |PARTITIONED|
+                                                              -- ASSIGN  |PARTITIONED|
+                                                                -- STREAM_PROJECT  |PARTITIONED|
+                                                                  -- STREAM_SELECT  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- STREAM_PROJECT  |PARTITIONED|
+                                                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                            -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                              -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                                                -- ASSIGN  |PARTITIONED|
+                                                                                  -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
diff --git a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_meta_active_ts.plan b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_meta_active_ts.plan
new file mode 100644
index 0000000..2bcef50
--- /dev/null
+++ b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_meta_active_ts.plan
@@ -0,0 +1,65 @@
+-- NOTIFY_BROKERS  |PARTITIONED|
+  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+    -- PRE_CLUSTERED_GROUP_BY[$$95, $$96, $$channelExecutionTime]  |PARTITIONED|
+            {
+              -- AGGREGATE  |LOCAL|
+                -- NESTED_TUPLE_SOURCE  |LOCAL|
+            }
+      -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+        -- STABLE_SORT [$$95(ASC), $$96(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
+          -- HASH_PARTITION_EXCHANGE [$$95, $$96, $$channelExecutionTime]  |PARTITIONED|
+            -- PRE_SORTED_DISTINCT_BY  |PARTITIONED|
+              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                -- STABLE_SORT [$$85(ASC)]  |PARTITIONED|
+                  -- HASH_PARTITION_EXCHANGE [$$85]  |PARTITIONED|
+                    -- STREAM_PROJECT  |PARTITIONED|
+                      -- COMMIT  |PARTITIONED|
+                        -- STREAM_PROJECT  |PARTITIONED|
+                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                            -- INDEX_INSERT_DELETE  |PARTITIONED|
+                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                -- STREAM_PROJECT  |PARTITIONED|
+                                  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                    -- INSERT_DELETE  |PARTITIONED|
+                                      -- HASH_PARTITION_EXCHANGE [$$79]  |PARTITIONED|
+                                        -- ASSIGN  |PARTITIONED|
+                                          -- STREAM_PROJECT  |PARTITIONED|
+                                            -- ASSIGN  |PARTITIONED|
+                                              -- STREAM_PROJECT  |PARTITIONED|
+                                                -- ASSIGN  |PARTITIONED|
+                                                  -- STREAM_PROJECT  |PARTITIONED|
+                                                    -- ASSIGN  |PARTITIONED|
+                                                      -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                        -- NESTED_LOOP  |PARTITIONED|
+                                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                            -- STREAM_PROJECT  |PARTITIONED|
+                                                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                -- HYBRID_HASH_JOIN [$$92, $$90][$$86, $$87]  |PARTITIONED|
+                                                                  -- HASH_PARTITION_EXCHANGE [$$92, $$90]  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                          -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                            -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                                              -- ASSIGN  |UNPARTITIONED|
+                                                                                -- EMPTY_TUPLE_SOURCE  |UNPARTITIONED|
+                                                                  -- HASH_PARTITION_EXCHANGE [$$86, $$87]  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                          -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                            -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                              -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                                                          -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                            -- STREAM_PROJECT  |PARTITIONED|
+                                                              -- ASSIGN  |PARTITIONED|
+                                                                -- STREAM_PROJECT  |PARTITIONED|
+                                                                  -- STREAM_SELECT  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- STREAM_PROJECT  |PARTITIONED|
+                                                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                            -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                              -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                                                -- ASSIGN  |PARTITIONED|
+                                                                                  -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
diff --git a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_meta_and.plan b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_meta_and.plan
new file mode 100644
index 0000000..2bcef50
--- /dev/null
+++ b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_meta_and.plan
@@ -0,0 +1,65 @@
+-- NOTIFY_BROKERS  |PARTITIONED|
+  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+    -- PRE_CLUSTERED_GROUP_BY[$$95, $$96, $$channelExecutionTime]  |PARTITIONED|
+            {
+              -- AGGREGATE  |LOCAL|
+                -- NESTED_TUPLE_SOURCE  |LOCAL|
+            }
+      -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+        -- STABLE_SORT [$$95(ASC), $$96(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
+          -- HASH_PARTITION_EXCHANGE [$$95, $$96, $$channelExecutionTime]  |PARTITIONED|
+            -- PRE_SORTED_DISTINCT_BY  |PARTITIONED|
+              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                -- STABLE_SORT [$$85(ASC)]  |PARTITIONED|
+                  -- HASH_PARTITION_EXCHANGE [$$85]  |PARTITIONED|
+                    -- STREAM_PROJECT  |PARTITIONED|
+                      -- COMMIT  |PARTITIONED|
+                        -- STREAM_PROJECT  |PARTITIONED|
+                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                            -- INDEX_INSERT_DELETE  |PARTITIONED|
+                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                -- STREAM_PROJECT  |PARTITIONED|
+                                  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                    -- INSERT_DELETE  |PARTITIONED|
+                                      -- HASH_PARTITION_EXCHANGE [$$79]  |PARTITIONED|
+                                        -- ASSIGN  |PARTITIONED|
+                                          -- STREAM_PROJECT  |PARTITIONED|
+                                            -- ASSIGN  |PARTITIONED|
+                                              -- STREAM_PROJECT  |PARTITIONED|
+                                                -- ASSIGN  |PARTITIONED|
+                                                  -- STREAM_PROJECT  |PARTITIONED|
+                                                    -- ASSIGN  |PARTITIONED|
+                                                      -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                        -- NESTED_LOOP  |PARTITIONED|
+                                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                            -- STREAM_PROJECT  |PARTITIONED|
+                                                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                -- HYBRID_HASH_JOIN [$$92, $$90][$$86, $$87]  |PARTITIONED|
+                                                                  -- HASH_PARTITION_EXCHANGE [$$92, $$90]  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                          -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                            -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                                              -- ASSIGN  |UNPARTITIONED|
+                                                                                -- EMPTY_TUPLE_SOURCE  |UNPARTITIONED|
+                                                                  -- HASH_PARTITION_EXCHANGE [$$86, $$87]  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                          -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                            -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                              -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                                                          -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                            -- STREAM_PROJECT  |PARTITIONED|
+                                                              -- ASSIGN  |PARTITIONED|
+                                                                -- STREAM_PROJECT  |PARTITIONED|
+                                                                  -- STREAM_SELECT  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- STREAM_PROJECT  |PARTITIONED|
+                                                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                            -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                              -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                                                -- ASSIGN  |PARTITIONED|
+                                                                                  -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
diff --git a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_meta_or.plan b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_meta_or.plan
new file mode 100644
index 0000000..7ed194c
--- /dev/null
+++ b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_meta_or.plan
@@ -0,0 +1,66 @@
+-- NOTIFY_BROKERS  |PARTITIONED|
+  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+    -- PRE_CLUSTERED_GROUP_BY[$$97, $$98, $$channelExecutionTime]  |PARTITIONED|
+            {
+              -- AGGREGATE  |LOCAL|
+                -- NESTED_TUPLE_SOURCE  |LOCAL|
+            }
+      -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+        -- STABLE_SORT [$$97(ASC), $$98(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
+          -- HASH_PARTITION_EXCHANGE [$$97, $$98, $$channelExecutionTime]  |PARTITIONED|
+            -- PRE_SORTED_DISTINCT_BY  |PARTITIONED|
+              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                -- STABLE_SORT [$$85(ASC)]  |PARTITIONED|
+                  -- HASH_PARTITION_EXCHANGE [$$85]  |PARTITIONED|
+                    -- STREAM_PROJECT  |PARTITIONED|
+                      -- COMMIT  |PARTITIONED|
+                        -- STREAM_PROJECT  |PARTITIONED|
+                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                            -- INDEX_INSERT_DELETE  |PARTITIONED|
+                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                -- STREAM_PROJECT  |PARTITIONED|
+                                  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                    -- INSERT_DELETE  |PARTITIONED|
+                                      -- HASH_PARTITION_EXCHANGE [$$79]  |PARTITIONED|
+                                        -- ASSIGN  |PARTITIONED|
+                                          -- STREAM_PROJECT  |PARTITIONED|
+                                            -- ASSIGN  |PARTITIONED|
+                                              -- STREAM_PROJECT  |PARTITIONED|
+                                                -- ASSIGN  |PARTITIONED|
+                                                  -- STREAM_PROJECT  |PARTITIONED|
+                                                    -- ASSIGN  |PARTITIONED|
+                                                      -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                        -- NESTED_LOOP  |PARTITIONED|
+                                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                            -- STREAM_PROJECT  |PARTITIONED|
+                                                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                -- HYBRID_HASH_JOIN [$$92, $$90][$$86, $$87]  |PARTITIONED|
+                                                                  -- HASH_PARTITION_EXCHANGE [$$92, $$90]  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                          -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                            -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                                              -- ASSIGN  |UNPARTITIONED|
+                                                                                -- EMPTY_TUPLE_SOURCE  |UNPARTITIONED|
+                                                                  -- HASH_PARTITION_EXCHANGE [$$86, $$87]  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                          -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                            -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                              -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                                                          -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                            -- STREAM_PROJECT  |PARTITIONED|
+                                                              -- ASSIGN  |PARTITIONED|
+                                                                -- STREAM_PROJECT  |PARTITIONED|
+                                                                  -- STREAM_SELECT  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
+                                                                      -- ASSIGN  |PARTITIONED|
+                                                                        -- ASSIGN  |PARTITIONED|
+                                                                          -- ASSIGN  |PARTITIONED|
+                                                                            -- STREAM_PROJECT  |PARTITIONED|
+                                                                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                                -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                                  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                                    -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
diff --git a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_push.plan b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_push.plan
index ebf5624..6134bd4 100644
--- a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_push.plan
+++ b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/new_tweets_push.plan
@@ -1,13 +1,13 @@
 -- NOTIFY_BROKERS  |PARTITIONED|
   -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-    -- PRE_CLUSTERED_GROUP_BY[$$93, $$94, $$channelExecutionTime]  |PARTITIONED|
+    -- PRE_CLUSTERED_GROUP_BY[$$91, $$92, $$channelExecutionTime]  |PARTITIONED|
             {
               -- AGGREGATE  |LOCAL|
                 -- NESTED_TUPLE_SOURCE  |LOCAL|
             }
       -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-        -- STABLE_SORT [$$93(ASC), $$94(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
-          -- HASH_PARTITION_EXCHANGE [$$93, $$94, $$channelExecutionTime]  |PARTITIONED|
+        -- STABLE_SORT [$$91(ASC), $$92(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
+          -- HASH_PARTITION_EXCHANGE [$$91, $$92, $$channelExecutionTime]  |PARTITIONED|
             -- PRE_SORTED_DISTINCT_BY  |PARTITIONED|
               -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                 -- STABLE_SORT [$$79(ASC)]  |PARTITIONED|
@@ -55,12 +55,11 @@
                                                               -- ASSIGN  |PARTITIONED|
                                                                 -- STREAM_PROJECT  |PARTITIONED|
                                                                   -- STREAM_SELECT  |PARTITIONED|
-                                                                    -- ASSIGN  |PARTITIONED|
+                                                                    -- STREAM_PROJECT  |PARTITIONED|
                                                                       -- ASSIGN  |PARTITIONED|
                                                                         -- STREAM_PROJECT  |PARTITIONED|
-                                                                          -- ASSIGN  |PARTITIONED|
-                                                                            -- STREAM_PROJECT  |PARTITIONED|
-                                                                              -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                                                                -- DATASOURCE_SCAN  |PARTITIONED|
-                                                                                  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                                                                    -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
+                                                                          -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
+                                                                            -- DATASOURCE_SCAN  |PARTITIONED|
+                                                                              -- BROADCAST_EXCHANGE  |PARTITIONED|
+                                                                                -- ASSIGN  |PARTITIONED|
+                                                                                  -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
diff --git a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/unseen_nearby_tweets.plan b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/unseen_nearby_tweets.plan
index 5bf7f10..a6a17de 100644
--- a/asterix-bad/src/test/resources/optimizerts/results/bad_cq/unseen_nearby_tweets.plan
+++ b/asterix-bad/src/test/resources/optimizerts/results/bad_cq/unseen_nearby_tweets.plan
@@ -1,13 +1,13 @@
 -- NOTIFY_BROKERS  |PARTITIONED|
   -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-    -- PRE_CLUSTERED_GROUP_BY[$$125, $$126, $$channelExecutionTime]  |PARTITIONED|
+    -- PRE_CLUSTERED_GROUP_BY[$$128, $$129, $$channelExecutionTime]  |PARTITIONED|
             {
               -- AGGREGATE  |LOCAL|
                 -- NESTED_TUPLE_SOURCE  |LOCAL|
             }
       -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-        -- STABLE_SORT [$$125(ASC), $$126(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
-          -- HASH_PARTITION_EXCHANGE [$$125, $$126, $$channelExecutionTime]  |PARTITIONED|
+        -- STABLE_SORT [$$128(ASC), $$129(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
+          -- HASH_PARTITION_EXCHANGE [$$128, $$129, $$channelExecutionTime]  |PARTITIONED|
             -- PRE_SORTED_DISTINCT_BY  |PARTITIONED|
               -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                 -- STABLE_SORT [$$99(ASC)]  |PARTITIONED|
diff --git a/asterix-bad/src/test/resources/optimizerts/results/bad_rq/channel-subscribe.plan b/asterix-bad/src/test/resources/optimizerts/results/bad_rq/channel-subscribe.plan
index a5a40c1..017bd9e 100644
--- a/asterix-bad/src/test/resources/optimizerts/results/bad_rq/channel-subscribe.plan
+++ b/asterix-bad/src/test/resources/optimizerts/results/bad_rq/channel-subscribe.plan
@@ -1,17 +1,17 @@
 -- NOTIFY_BROKERS  |PARTITIONED|
   -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-    -- PRE_CLUSTERED_GROUP_BY[$$100, $$101, $$channelExecutionTime]  |PARTITIONED|
+    -- PRE_CLUSTERED_GROUP_BY[$$106, $$107, $$channelExecutionTime]  |PARTITIONED|
             {
               -- AGGREGATE  |LOCAL|
                 -- NESTED_TUPLE_SOURCE  |LOCAL|
             }
       -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-        -- STABLE_SORT [$$100(ASC), $$101(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
-          -- HASH_PARTITION_EXCHANGE [$$100, $$101, $$channelExecutionTime]  |PARTITIONED|
+        -- STABLE_SORT [$$106(ASC), $$107(ASC), $$channelExecutionTime(ASC)]  |PARTITIONED|
+          -- HASH_PARTITION_EXCHANGE [$$106, $$107, $$channelExecutionTime]  |PARTITIONED|
             -- PRE_SORTED_DISTINCT_BY  |PARTITIONED|
               -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                -- STABLE_SORT [$$85(ASC)]  |PARTITIONED|
-                  -- HASH_PARTITION_EXCHANGE [$$85]  |PARTITIONED|
+                -- STABLE_SORT [$$91(ASC)]  |PARTITIONED|
+                  -- HASH_PARTITION_EXCHANGE [$$91]  |PARTITIONED|
                     -- STREAM_PROJECT  |PARTITIONED|
                       -- COMMIT  |PARTITIONED|
                         -- STREAM_PROJECT  |PARTITIONED|
@@ -21,7 +21,7 @@
                                 -- STREAM_PROJECT  |PARTITIONED|
                                   -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                     -- INSERT_DELETE  |PARTITIONED|
-                                      -- HASH_PARTITION_EXCHANGE [$$79]  |PARTITIONED|
+                                      -- HASH_PARTITION_EXCHANGE [$$85]  |PARTITIONED|
                                         -- ASSIGN  |PARTITIONED|
                                           -- STREAM_PROJECT  |PARTITIONED|
                                             -- ASSIGN  |PARTITIONED|
@@ -35,8 +35,8 @@
                                                             -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                                               -- STREAM_PROJECT  |PARTITIONED|
                                                                 -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
-                                                                  -- HYBRID_HASH_JOIN [$$91, $$89][$$86, $$87]  |PARTITIONED|
-                                                                    -- HASH_PARTITION_EXCHANGE [$$91, $$89]  |PARTITIONED|
+                                                                  -- HYBRID_HASH_JOIN [$$97, $$95][$$92, $$93]  |PARTITIONED|
+                                                                    -- HASH_PARTITION_EXCHANGE [$$97, $$95]  |PARTITIONED|
                                                                       -- STREAM_PROJECT  |PARTITIONED|
                                                                         -- ASSIGN  |PARTITIONED|
                                                                           -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
@@ -44,7 +44,7 @@
                                                                               -- BROADCAST_EXCHANGE  |PARTITIONED|
                                                                                 -- ASSIGN  |UNPARTITIONED|
                                                                                   -- EMPTY_TUPLE_SOURCE  |UNPARTITIONED|
-                                                                    -- HASH_PARTITION_EXCHANGE [$$86, $$87]  |PARTITIONED|
+                                                                    -- HASH_PARTITION_EXCHANGE [$$92, $$93]  |PARTITIONED|
                                                                       -- STREAM_PROJECT  |PARTITIONED|
                                                                         -- ASSIGN  |PARTITIONED|
                                                                           -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
@@ -65,7 +65,7 @@
       -- STREAM_PROJECT  |PARTITIONED|
         -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
           -- INSERT_DELETE  |PARTITIONED|
-            -- HASH_PARTITION_EXCHANGE [$$13]  |PARTITIONED|
+            -- HASH_PARTITION_EXCHANGE [$$14]  |PARTITIONED|
               -- ASSIGN  |UNPARTITIONED|
                 -- STREAM_PROJECT  |UNPARTITIONED|
                   -- ASSIGN  |UNPARTITIONED|
diff --git a/asterix-bad/src/test/resources/runtimets/queries/bad_rq/add_index/add_index.5.query.sqlpp b/asterix-bad/src/test/resources/runtimets/queries/bad_rq/add_index/add_index.5.query.sqlpp
index eebfbc4..f5b5a03 100644
--- a/asterix-bad/src/test/resources/runtimets/queries/bad_rq/add_index/add_index.5.query.sqlpp
+++ b/asterix-bad/src/test/resources/runtimets/queries/bad_rq/add_index/add_index.5.query.sqlpp
@@ -19,4 +19,4 @@
 use channels;
 
 select value array_count(
-(select * from EmergencyChannelResults where deliveryTime > datetime("2017-05-02T17:52:59.570Z")));
\ No newline at end of file
+(select * from EmergencyChannelResults where deliveryTime > datetime("2017-05-02T17:52:59.570Z"))) > 1000;
\ No newline at end of file
diff --git a/asterix-bad/src/test/resources/runtimets/results/bad_rq/add_index/add_index.1.adm b/asterix-bad/src/test/resources/runtimets/results/bad_rq/add_index/add_index.1.adm
index 2e9bdd9..f32a580 100644
--- a/asterix-bad/src/test/resources/runtimets/results/bad_rq/add_index/add_index.1.adm
+++ b/asterix-bad/src/test/resources/runtimets/results/bad_rq/add_index/add_index.1.adm
@@ -1 +1 @@
-1074
\ No newline at end of file
+true
\ No newline at end of file