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 0467f6e..db4fc2d 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
@@ -19,7 +19,8 @@
 package org.apache.asterix.bad;
 
 public interface BADConstants {
-    String SubscriptionId = "subscriptionId";
+    String ChannelSubscriptionId = "channelSubId";
+    String BrokerSubscriptionId = "brokerSubId";
     String BrokerName = "BrokerName";
     String ChannelName = "ChannelName";
     String ProcedureName = "ProcedureName";
@@ -29,9 +30,11 @@
     String ResultId = "resultId";
     String ChannelExecutionTime = "channelExecutionTime";
     String ChannelSubscriptionsType = "ChannelSubscriptionsType";
+    String BrokerSubscriptionsType = "BrokerSubscriptionsType";
     String ChannelResultsType = "ChannelResultsType";
     String ResultsDatasetName = "ResultsDatasetName";
-    String SubscriptionsDatasetName = "SubscriptionsDatasetName";
+    String ChannelSubscriptionsDatasetName = "ChannelSubscriptionsDatasetName";
+    String BrokerSubscriptionsDatasetName = "BrokerSubscriptionsDatasetName";
     String CHANNEL_EXTENSION_NAME = "Channel";
     String PROCEDURE_KEYWORD = "Procedure";
     String BROKER_KEYWORD = "Broker";
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelDropStatement.java b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelDropStatement.java
index 8403fcc..5d9d971 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelDropStatement.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelDropStatement.java
@@ -142,7 +142,11 @@
             ((QueryTranslator) statementExecutor).handleDatasetDropStatement(tempMdProvider, dropStmt, hcc, null);
             tempMdProvider.getLocks().reset();
             dropStmt = new DropDatasetStatement(new Identifier(dataverse),
-                    new Identifier(channel.getSubscriptionsDataset()), true);
+                    new Identifier(channel.getChannelSubscriptionsDataset()), true);
+            ((QueryTranslator) statementExecutor).handleDatasetDropStatement(tempMdProvider, dropStmt, hcc, null);
+            tempMdProvider.getLocks().reset();
+            dropStmt = new DropDatasetStatement(new Identifier(dataverse),
+                    new Identifier(channel.getBrokerSubscriptionsDataset()), true);
             ((QueryTranslator) statementExecutor).handleDatasetDropStatement(tempMdProvider, dropStmt, hcc, null);
             MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
         } catch (Exception e) {
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelSubscribeStatement.java b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelSubscribeStatement.java
index 7583f0b..5803f60 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelSubscribeStatement.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelSubscribeStatement.java
@@ -138,7 +138,7 @@
                 throw new AsterixException("There is no broker with this name " + brokerName + ".");
             }
 
-            String subscriptionsDatasetName = channel.getSubscriptionsDataset();
+            String subscriptionsDatasetName = channel.getChannelSubscriptionsDataset();
 
             if (argList.size() != channel.getFunction().getArity()) {
                 throw new AsterixException("Channel expected " + channel.getFunction().getArity()
@@ -157,7 +157,7 @@
             fb.add(new FieldBinding(leftExpr, rightExpr));
 
             if (subscriptionId != null) {
-                leftExpr = new LiteralExpr(new StringLiteral(BADConstants.SubscriptionId));
+                leftExpr = new LiteralExpr(new StringLiteral(BADConstants.ChannelSubscriptionId));
 
                 List<Expression> UUIDList = new ArrayList<>();
                 UUIDList.add(new LiteralExpr(new StringLiteral(subscriptionId)));
@@ -190,7 +190,8 @@
                 VariableExpr resultVar = new VariableExpr(new VarIdentifier("$result", 0));
                 VariableExpr useResultVar = new VariableExpr(new VarIdentifier("$result", 0));
                 useResultVar.setIsNewVar(false);
-                FieldAccessor accessor = new FieldAccessor(useResultVar, new Identifier(BADConstants.SubscriptionId));
+                FieldAccessor accessor =
+                        new FieldAccessor(useResultVar, new Identifier(BADConstants.ChannelSubscriptionId));
 
                 metadataProvider.setResultSetId(new ResultSetId(resultSetId));
                 boolean resultsAsync =
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelUnsubscribeStatement.java b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelUnsubscribeStatement.java
index b23bf3b..e6ba69c 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelUnsubscribeStatement.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelUnsubscribeStatement.java
@@ -113,11 +113,11 @@
                 throw new AsterixException("There is no channel with this name " + channelName + ".");
             }
 
-            String subscriptionsDatasetName = channel.getSubscriptionsDataset();
+            String subscriptionsDatasetName = channel.getChannelSubscriptionsDataset();
 
             //Need a condition to say subscription-id = sid
             OperatorExpr condition = new OperatorExpr();
-            FieldAccessor fa = new FieldAccessor(vars, new Identifier(BADConstants.SubscriptionId));
+            FieldAccessor fa = new FieldAccessor(vars, new Identifier(BADConstants.ChannelSubscriptionId));
             condition.addOperand(fa);
             condition.setCurrentop(true);
             condition.addOperator("=");
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/CreateChannelStatement.java b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/CreateChannelStatement.java
index a28666a..b2ffa1b 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/CreateChannelStatement.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/CreateChannelStatement.java
@@ -85,7 +85,8 @@
     private Identifier dataverseName;
     private String duration;
     private String body;
-    private String subscriptionsTableName;
+    private String channelSubscriptionsTableName;
+    private String brokerSubscriptionsTableName;
     private String resultsTableName;
     private String dataverse;
     private final boolean push;
@@ -113,7 +114,7 @@
     }
 
     public String getSubscriptionsName() {
-        return subscriptionsTableName;
+        return channelSubscriptionsTableName;
     }
 
     public String getDuration() {
@@ -159,29 +160,53 @@
     private void createDatasets(IStatementExecutor statementExecutor, MetadataProvider metadataProvider,
             IHyracksClientConnection hcc) throws AsterixException, Exception {
 
+        //Create channel subscriptions dataset
         Identifier subscriptionsTypeName = new Identifier(BADConstants.ChannelSubscriptionsType);
         Identifier resultsTypeName = new Identifier(BADConstants.ChannelResultsType);
         //Setup the subscriptions dataset
         List<List<String>> partitionFields = new ArrayList<>();
         List<Integer> keyIndicators = new ArrayList<>();
         keyIndicators.add(0);
-        List<String> fieldNames = new ArrayList<>();
-        fieldNames.add(BADConstants.SubscriptionId);
-        partitionFields.add(fieldNames);
-        IDatasetDetailsDecl idd = new InternalDetailsDecl(partitionFields, keyIndicators, true, null);
-        DatasetDecl createSubscriptionsDataset = new DatasetDecl(dataverseName, new Identifier(subscriptionsTableName),
-                new Identifier(BADConstants.BAD_DATAVERSE_NAME), subscriptionsTypeName, null, null, null,
-                new HashMap<String, String>(), DatasetType.INTERNAL, idd, null, true);
+        List<String> fieldName = new ArrayList<>();
+        fieldName.add(BADConstants.ChannelSubscriptionId);
+        partitionFields.add(fieldName);
+        IDatasetDetailsDecl idd = new InternalDetailsDecl(partitionFields, keyIndicators, false, null);
+        DatasetDecl createChannelSubscriptionsDataset =
+                new DatasetDecl(dataverseName, new Identifier(channelSubscriptionsTableName),
+                        new Identifier(BADConstants.BAD_DATAVERSE_NAME), subscriptionsTypeName, null, null, null,
+                        new HashMap<String, String>(), DatasetType.INTERNAL, idd, null, true);
 
-        ((QueryTranslator) statementExecutor).handleCreateDatasetStatement(metadataProvider, createSubscriptionsDataset,
+        ((QueryTranslator) statementExecutor).handleCreateDatasetStatement(metadataProvider,
+                createChannelSubscriptionsDataset, hcc, null);
+
+        //Create broker subscriptions dataset
+        Identifier brokerSubscriptionsTypeName = new Identifier(BADConstants.BrokerSubscriptionsType);
+        partitionFields = new ArrayList<>();
+        keyIndicators = new ArrayList<>();
+        keyIndicators.add(0);
+        keyIndicators.add(0);
+        fieldName = new ArrayList<>();
+        List<String> fieldName2 = new ArrayList<>();
+        fieldName.add(BADConstants.ChannelSubscriptionId);
+        fieldName2.add(BADConstants.BrokerSubscriptionId);
+        partitionFields.add(fieldName);
+        partitionFields.add(fieldName2);
+        idd = new InternalDetailsDecl(partitionFields, keyIndicators, false, null);
+        DatasetDecl createBrokerSubscriptionsDataset =
+                new DatasetDecl(dataverseName, new Identifier(brokerSubscriptionsTableName),
+                        new Identifier(BADConstants.BAD_DATAVERSE_NAME), brokerSubscriptionsTypeName, null, null, null,
+                        new HashMap<String, String>(), DatasetType.INTERNAL, idd, null, true);
+        metadataProvider.getLocks().reset();
+        ((QueryTranslator) statementExecutor).handleCreateDatasetStatement(metadataProvider,
+                createBrokerSubscriptionsDataset,
                 hcc, null);
 
         if (!push) {
             //Setup the results dataset
             partitionFields = new ArrayList<>();
-            fieldNames = new ArrayList<>();
-            fieldNames.add(BADConstants.ResultId);
-            partitionFields.add(fieldNames);
+            fieldName = new ArrayList<>();
+            fieldName.add(BADConstants.ResultId);
+            partitionFields.add(fieldName);
             idd = new InternalDetailsDecl(partitionFields, keyIndicators, true, null);
             DatasetDecl createResultsDataset = new DatasetDecl(dataverseName, new Identifier(resultsTableName),
                     new Identifier(BADConstants.BAD_DATAVERSE_NAME), resultsTypeName, null, null, null, new HashMap<>(),
@@ -225,9 +250,10 @@
         builder.append("with " + BADConstants.ChannelExecutionTime + " as current_datetime() \n");
         builder.append("select result, ");
         builder.append(BADConstants.ChannelExecutionTime + ", ");
-        builder.append("sub." + BADConstants.SubscriptionId + " as " + BADConstants.SubscriptionId + ",");
+        builder.append("sub." + BADConstants.ChannelSubscriptionId + " as " + BADConstants.ChannelSubscriptionId + ",");
         builder.append("current_datetime() as " + BADConstants.DeliveryTime + "\n");
-        builder.append("from " + dataverse + "." + subscriptionsTableName + " sub,\n");
+        builder.append("from " + dataverse + "." + channelSubscriptionsTableName + " sub,\n");
+        builder.append(dataverse + "." + brokerSubscriptionsTableName + " bs,\n");
         builder.append(BADConstants.BAD_DATAVERSE_NAME + "." + BADConstants.BROKER_KEYWORD + " b, \n");
         builder.append(function.getNamespace() + "." + function.getName() + "(");
         int i = 0;
@@ -235,8 +261,6 @@
             builder.append("sub.param" + i + ",");
         }
         builder.append("sub.param" + i + ") result \n");
-        builder.append("where b." + BADConstants.BrokerName + " = sub." + BADConstants.BrokerName + "\n");
-        builder.append("and b." + BADConstants.DataverseName + " = sub." + BADConstants.DataverseName + "\n");
         if (!push) {
             builder.append(")");
             builder.append(" returning a");
@@ -269,7 +293,9 @@
 
         dataverseName = new Identifier(((QueryTranslator) statementExecutor).getActiveDataverse(dataverseName));
         dataverse = dataverseName.getValue();
-        subscriptionsTableName = channelName + BADConstants.subscriptionEnding;
+        channelSubscriptionsTableName =
+                channelName + BADConstants.CHANNEL_EXTENSION_NAME + BADConstants.subscriptionEnding;
+        brokerSubscriptionsTableName = channelName + BADConstants.BROKER_KEYWORD + BADConstants.subscriptionEnding;
         resultsTableName = push ? "" : channelName + BADConstants.resultsEnding;
 
         EntityId entityId = new EntityId(BADConstants.CHANNEL_EXTENSION_NAME, dataverse, channelName.getValue());
@@ -297,7 +323,10 @@
             initialize(mdTxnCtx);
 
             //check if names are available before creating anything
-            if (MetadataManager.INSTANCE.getDataset(mdTxnCtx, dataverse, subscriptionsTableName) != null) {
+            if (MetadataManager.INSTANCE.getDataset(mdTxnCtx, dataverse, channelSubscriptionsTableName) != null) {
+                throw new AsterixException("The channel name:" + channelName + " is not available.");
+            }
+            if (MetadataManager.INSTANCE.getDataset(mdTxnCtx, dataverse, brokerSubscriptionsTableName) != null) {
                 throw new AsterixException("The channel name:" + channelName + " is not available.");
             }
             if (!push && MetadataManager.INSTANCE.getDataset(mdTxnCtx, dataverse, resultsTableName) != null) {
@@ -323,7 +352,8 @@
 
             BADJobService.setupExecutorJob(entityId, channeljobSpec, hcc, listener, metadataProvider.getTxnIdFactory(),
                     duration);
-            channel = new Channel(dataverse, channelName.getValue(), subscriptionsTableName, resultsTableName, function,
+            channel = new Channel(dataverse, channelName.getValue(), channelSubscriptionsTableName,
+                    brokerSubscriptionsTableName, resultsTableName, function,
                     duration, null, body);
 
             MetadataManager.INSTANCE.addEntity(mdTxnCtx, channel);
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataExtension.java b/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataExtension.java
index cd2ff86..5be2e84 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataExtension.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataExtension.java
@@ -47,8 +47,12 @@
     public static final Dataverse BAD_DATAVERSE = new Dataverse(BADConstants.BAD_DATAVERSE_NAME,
             NonTaggedDataFormat.class.getName(), MetadataUtil.PENDING_NO_OP);
 
-    public static final Datatype BAD_SUBSCRIPTION_DATATYPE = new Datatype(BADConstants.BAD_DATAVERSE_NAME,
+    public static final Datatype BAD_CHANNEL_SUBSCRIPTION_DATATYPE = new Datatype(BADConstants.BAD_DATAVERSE_NAME,
             BADConstants.ChannelSubscriptionsType, BADMetadataRecordTypes.channelSubscriptionsType, false);
+
+    public static final Datatype BAD_BROKER_SUBSCRIPTION_DATATYPE = new Datatype(BADConstants.BAD_DATAVERSE_NAME,
+            BADConstants.BrokerSubscriptionsType, BADMetadataRecordTypes.brokerSubscriptionsType, false);
+
     public static final Datatype BAD_RESULT_DATATYPE = new Datatype(BADConstants.BAD_DATAVERSE_NAME,
             BADConstants.ChannelResultsType, BADMetadataRecordTypes.channelResultsType, false);
 
@@ -76,7 +80,6 @@
         return new MetadataTupleTranslatorProvider();
     }
 
-    @SuppressWarnings("rawtypes")
     @Override
     public List<ExtensionMetadataDataset> getExtensionIndexes() {
         try {
@@ -107,7 +110,8 @@
                 // MetadataManager.INSTANCE.addDataverse(mdTxnCtx, BAD_DATAVERSE);
                 // insert default data type
                 MetadataManager.INSTANCE.addDatatype(mdTxnCtx, BAD_RESULT_DATATYPE);
-                MetadataManager.INSTANCE.addDatatype(mdTxnCtx, BAD_SUBSCRIPTION_DATATYPE);
+                MetadataManager.INSTANCE.addDatatype(mdTxnCtx, BAD_CHANNEL_SUBSCRIPTION_DATATYPE);
+                MetadataManager.INSTANCE.addDatatype(mdTxnCtx, BAD_BROKER_SUBSCRIPTION_DATATYPE);
                 MetadataManager.INSTANCE.addDatatype(mdTxnCtx, BAD_BROKER_DATATYPE);
                 MetadataManager.INSTANCE.addDatatype(mdTxnCtx, BAD_CHANNEL_DATATYPE);
                 MetadataManager.INSTANCE.addDatatype(mdTxnCtx, BAD_PROCEDURE_DATATYPE);
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataRecordTypes.java b/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataRecordTypes.java
index a764a5a..f6e2e1f 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataRecordTypes.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/BADMetadataRecordTypes.java
@@ -27,16 +27,24 @@
 
 public class BADMetadataRecordTypes {
 
-    // -------------------------------------- Subscriptions --------------------------------------//
-    private static final String[] subTypeFieldNames =
-            { BADConstants.DataverseName, BADConstants.BrokerName, BADConstants.SubscriptionId };
-    private static final IAType[] subTypeFieldTypes = { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.AUUID };
+    // -------------------------------------- Channel Subscriptions --------------------------------------//
+    private static final String[] channelSubTypeFieldNames = { BADConstants.ChannelSubscriptionId };
+    private static final IAType[] channelSubTypeFieldTypes = { BuiltinType.AUUID };
     public static final ARecordType channelSubscriptionsType =
-            new ARecordType(BADConstants.ChannelSubscriptionsType, subTypeFieldNames, subTypeFieldTypes, true);
+            new ARecordType(BADConstants.ChannelSubscriptionsType, channelSubTypeFieldNames, channelSubTypeFieldTypes,
+                    true);
+
+    // -------------------------------------- Broker Subscriptions --------------------------------------//
+    private static final String[] brokerSubTypeFieldNames = { BADConstants.ChannelSubscriptionId,
+            BADConstants.BrokerSubscriptionId, BADConstants.DataverseName, BADConstants.BrokerName };
+    private static final IAType[] brokerSubTypeFieldTypes =
+            { BuiltinType.AUUID, BuiltinType.AUUID, BuiltinType.ASTRING, BuiltinType.ASTRING };
+    public static final ARecordType brokerSubscriptionsType = new ARecordType(BADConstants.BrokerSubscriptionsType,
+            brokerSubTypeFieldNames, brokerSubTypeFieldTypes, true);
 
     // ---------------------------------------- Results --------------------------------------------//
     private static final String[] resultTypeFieldNames = { BADConstants.ResultId, BADConstants.ChannelExecutionTime,
-            BADConstants.SubscriptionId, BADConstants.DeliveryTime };
+            BADConstants.ChannelSubscriptionId, BADConstants.DeliveryTime };
     private static final IAType[] resultTypeFieldTypes =
             { BuiltinType.AUUID, BuiltinType.ADATETIME, BuiltinType.AUUID, BuiltinType.ADATETIME };
     public static final ARecordType channelResultsType =
@@ -45,25 +53,28 @@
     //------------------------------------------ Channel ----------------------------------------//     
     public static final int CHANNEL_ARECORD_DATAVERSE_NAME_FIELD_INDEX = 0;
     public static final int CHANNEL_ARECORD_CHANNEL_NAME_FIELD_INDEX = 1;
-    public static final int CHANNEL_ARECORD_SUBSCRIPTIONS_NAME_FIELD_INDEX = 2;
+    public static final int CHANNEL_ARECORD_CHANNEL_SUBSCRIPTIONS_NAME_FIELD_INDEX = 2;
     public static final int CHANNEL_ARECORD_RESULTS_NAME_FIELD_INDEX = 3;
     public static final int CHANNEL_ARECORD_FUNCTION_FIELD_INDEX = 4;
     public static final int CHANNEL_ARECORD_DURATION_FIELD_INDEX = 5;
     public static final int CHANNEL_ARECORD_DEPENDENCIES_FIELD_INDEX = 6;
     public static final int CHANNEL_ARECORD_BODY_FIELD_INDEX = 7;
+    public static final int CHANNEL_ARECORD_BROKER_SUBSCRIPTIONS_NAME_FIELD_INDEX = 8;
     public static final ARecordType CHANNEL_RECORDTYPE = MetadataRecordTypes.createRecordType(
             // RecordTypeName
             BADConstants.RECORD_TYPENAME_CHANNEL,
             // FieldNames
-            new String[] { BADConstants.DataverseName, BADConstants.ChannelName, BADConstants.SubscriptionsDatasetName,
+            new String[] { BADConstants.DataverseName, BADConstants.ChannelName,
+                    BADConstants.ChannelSubscriptionsDatasetName,
                     BADConstants.ResultsDatasetName, BADConstants.Function, BADConstants.Duration,
-                    BADConstants.FIELD_NAME_DEPENDENCIES, BADConstants.FIELD_NAME_BODY },
+                    BADConstants.FIELD_NAME_DEPENDENCIES, BADConstants.FIELD_NAME_BODY,
+                    BADConstants.BrokerSubscriptionsDatasetName },
             // FieldTypes
             new IAType[] { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING,
                     new AOrderedListType(BuiltinType.ASTRING, null), BuiltinType.ASTRING,
                     new AOrderedListType(new AOrderedListType(new AOrderedListType(BuiltinType.ASTRING, null), null),
                             null),
-                    BuiltinType.ASTRING },
+                    BuiltinType.ASTRING, BuiltinType.ASTRING },
             //IsOpen?
             true);
     //------------------------------------------ Broker ----------------------------------------//
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/Channel.java b/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/Channel.java
index ed9346c..61df9af 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/Channel.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/Channel.java
@@ -34,7 +34,8 @@
 
     /** A unique identifier for the channel */
     protected final EntityId channelId;
-    private final String subscriptionsDatasetName;
+    private final String channelSubscriptionsDatasetName;
+    private final String brokerSubscriptionsDatasetName;
     private final String resultsDatasetName;
     private final String duration;
     private final String channelBody;
@@ -49,13 +50,15 @@
     */
     private final List<List<List<String>>> dependencies;
 
-    public Channel(String dataverseName, String channelName, String subscriptionsDataset, String resultsDataset,
+    public Channel(String dataverseName, String channelName, String channelSubscriptionsDatasetName,
+            String brokerSubscriptionsDatasetName, String resultsDataset,
             FunctionSignature function, String duration, List<List<List<String>>> dependencies, String channelBody) {
         this.channelId = new EntityId(BADConstants.CHANNEL_EXTENSION_NAME, dataverseName, channelName);
         this.function = function;
         this.duration = duration;
         this.resultsDatasetName = resultsDataset;
-        this.subscriptionsDatasetName = subscriptionsDataset;
+        this.channelSubscriptionsDatasetName = channelSubscriptionsDatasetName;
+        this.brokerSubscriptionsDatasetName = brokerSubscriptionsDatasetName;
         this.channelBody = channelBody;
         if (this.function.getNamespace() == null) {
             this.function.setNamespace(dataverseName);
@@ -67,7 +70,7 @@
             this.dependencies.add(new ArrayList<>());
             this.dependencies.add(new ArrayList<>());
             List<String> resultsList = Arrays.asList(dataverseName, resultsDatasetName);
-            List<String> subscriptionList = Arrays.asList(dataverseName, subscriptionsDatasetName);
+            List<String> subscriptionList = Arrays.asList(dataverseName, channelSubscriptionsDatasetName);
             this.dependencies.get(0).add(resultsList);
             this.dependencies.get(0).add(subscriptionList);
             this.dependencies.get(1).add(functionAsPath);
@@ -84,8 +87,12 @@
         return dependencies;
     }
 
-    public String getSubscriptionsDataset() {
-        return subscriptionsDatasetName;
+    public String getChannelSubscriptionsDataset() {
+        return channelSubscriptionsDatasetName;
+    }
+
+    public String getBrokerSubscriptionsDataset() {
+        return brokerSubscriptionsDatasetName;
     }
 
     public String getResultsDatasetName() {
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/ChannelTupleTranslator.java b/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/ChannelTupleTranslator.java
index 175280e..6cf7f50 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/ChannelTupleTranslator.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/metadata/ChannelTupleTranslator.java
@@ -82,8 +82,9 @@
         String channelName =
                 ((AString) channelRecord.getValueByPos(BADMetadataRecordTypes.CHANNEL_ARECORD_CHANNEL_NAME_FIELD_INDEX))
                         .getStringValue();
-        String subscriptionsName = ((AString) channelRecord
-                .getValueByPos(BADMetadataRecordTypes.CHANNEL_ARECORD_SUBSCRIPTIONS_NAME_FIELD_INDEX)).getStringValue();
+        String channelSubscriptionsName = ((AString) channelRecord
+                .getValueByPos(BADMetadataRecordTypes.CHANNEL_ARECORD_CHANNEL_SUBSCRIPTIONS_NAME_FIELD_INDEX))
+                        .getStringValue();
         String resultsName =
                 ((AString) channelRecord.getValueByPos(BADMetadataRecordTypes.CHANNEL_ARECORD_RESULTS_NAME_FIELD_INDEX))
                         .getStringValue();
@@ -127,10 +128,15 @@
                 ((AString) channelRecord.getValueByPos(BADMetadataRecordTypes.CHANNEL_ARECORD_BODY_FIELD_INDEX))
                         .getStringValue();
 
+        String brokerSubscriptionsName = ((AString) channelRecord
+                .getValueByPos(BADMetadataRecordTypes.CHANNEL_ARECORD_BROKER_SUBSCRIPTIONS_NAME_FIELD_INDEX))
+                        .getStringValue();
+
         FunctionSignature signature = new FunctionSignature(functionSignature.get(0), functionSignature.get(1),
                 Integer.parseInt(functionSignature.get(2)));
 
-        channel = new Channel(dataverseName, channelName, subscriptionsName, resultsName, signature, duration,
+        channel = new Channel(dataverseName, channelName, channelSubscriptionsName, brokerSubscriptionsName,
+                resultsName, signature, duration,
                 dependencies, channelBody);
         return channel;
     }
@@ -164,9 +170,10 @@
 
         // write field 2
         fieldValue.reset();
-        aString.setValue(channel.getSubscriptionsDataset());
+        aString.setValue(channel.getChannelSubscriptionsDataset());
         stringSerde.serialize(aString, fieldValue.getDataOutput());
-        recordBuilder.addField(BADMetadataRecordTypes.CHANNEL_ARECORD_SUBSCRIPTIONS_NAME_FIELD_INDEX, fieldValue);
+        recordBuilder.addField(BADMetadataRecordTypes.CHANNEL_ARECORD_CHANNEL_SUBSCRIPTIONS_NAME_FIELD_INDEX,
+                fieldValue);
 
         // write field 3
         fieldValue.reset();
@@ -227,6 +234,13 @@
         stringSerde.serialize(aString, fieldValue.getDataOutput());
         recordBuilder.addField(BADMetadataRecordTypes.CHANNEL_ARECORD_BODY_FIELD_INDEX, fieldValue);
 
+        // write field 8
+        fieldValue.reset();
+        aString.setValue(channel.getBrokerSubscriptionsDataset());
+        stringSerde.serialize(aString, fieldValue.getDataOutput());
+        recordBuilder.addField(BADMetadataRecordTypes.CHANNEL_ARECORD_BROKER_SUBSCRIPTIONS_NAME_FIELD_INDEX,
+                fieldValue);
+
         // write record
         recordBuilder.write(tupleBuilder.getDataOutput(), true);
 
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/rules/InsertBrokerNotifierForChannelRule.java b/asterix-bad/src/main/java/org/apache/asterix/bad/rules/InsertBrokerNotifierForChannelRule.java
index 9ead7f0..4ec6d2f 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/rules/InsertBrokerNotifierForChannelRule.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/rules/InsertBrokerNotifierForChannelRule.java
@@ -28,9 +28,11 @@
 import org.apache.asterix.bad.runtime.NotifyBrokerPOperator;
 import org.apache.asterix.lang.common.util.FunctionUtil;
 import org.apache.asterix.metadata.declared.DatasetDataSource;
+import org.apache.asterix.om.base.AInt32;
 import org.apache.asterix.om.base.AString;
 import org.apache.asterix.om.constants.AsterixConstantValue;
 import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.asterix.om.functions.FunctionInfo;
 import org.apache.asterix.om.types.IAType;
 import org.apache.commons.lang3.mutable.Mutable;
 import org.apache.commons.lang3.mutable.MutableObject;
@@ -47,6 +49,7 @@
 import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
 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.AbstractLogicalOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AggregateOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AssignOperator;
@@ -58,6 +61,7 @@
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.InsertDeleteUpsertOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.NestedTupleSourceOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.ProjectOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.SelectOperator;
 import org.apache.hyracks.algebricks.core.algebra.plan.ALogicalPlanImpl;
 import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
 
@@ -76,25 +80,19 @@
         if (op1.getOperatorTag() != LogicalOperatorTag.DISTRIBUTE_RESULT) {
             return false;
         }
-        boolean push = false;
 
-        AbstractLogicalOperator op = (AbstractLogicalOperator) op1.getInputs().get(0).getValue();
-        if (op.getOperatorTag() != LogicalOperatorTag.DELEGATE_OPERATOR) {
-            if (op.getOperatorTag() != LogicalOperatorTag.PROJECT) {
-                return false;
-            }
+        boolean push = false;
+        AbstractLogicalOperator op = findOp(op1, 4, "", "");
+        if (op == null) {
             push = true;
         }
+
         DataSourceScanOperator subscriptionsScan;
         String channelDataverse;
         String channelName;
 
         if (!push) {
-            DelegateOperator eOp = (DelegateOperator) op;
-            if (!(eOp.getDelegate() instanceof CommitOperator)) {
-                return false;
-            }
-            AbstractLogicalOperator descendantOp = (AbstractLogicalOperator) eOp.getInputs().get(0).getValue();
+            AbstractLogicalOperator descendantOp = (AbstractLogicalOperator) op.getInputs().get(0).getValue();
             if (descendantOp.getOperatorTag() != LogicalOperatorTag.INSERT_DELETE_UPSERT) {
                 return false;
             }
@@ -113,15 +111,20 @@
             //Now we know that we are inserting into results
 
             channelName = datasetName.substring(0, datasetName.length() - 7);
-            String subscriptionsName = channelName + "Subscriptions";
-            subscriptionsScan = (DataSourceScanOperator) findOp(op, subscriptionsName);
+            String subscriptionsName = channelName + "ChannelSubscriptions";
+            subscriptionsScan =
+                    (DataSourceScanOperator) findOp(op, 3, subscriptionsName, BADConstants.ChannelSubscriptionsType);
             if (subscriptionsScan == null) {
                 return false;
             }
 
         } else {
+            op = (AbstractLogicalOperator) op1.getInputs().get(0).getValue();
+            if (op.getOperatorTag() != LogicalOperatorTag.PROJECT) {
+                return false;
+            }
             //if push, get the channel name here instead
-            subscriptionsScan = (DataSourceScanOperator) findOp(op, "");
+            subscriptionsScan = (DataSourceScanOperator) findOp(op, 3, "", BADConstants.ChannelSubscriptionsType);
             if (subscriptionsScan == null) {
                 return false;
             }
@@ -131,15 +134,8 @@
             channelName = datasetName.substring(0, datasetName.length() - 13);
         }
 
-        //Now we need to get the broker EndPoint
-        LogicalVariable brokerEndpointVar = context.newVar();
-        AbstractLogicalOperator opAboveBrokersScan = findOp(op, "brokers");
-        if (opAboveBrokersScan == null) {
-            return false;
-        }
-
-        //get subscriptionIdVar
-        LogicalVariable subscriptionIdVar = subscriptionsScan.getVariables().get(0);
+        //get channelSubscriptionIdVar
+        LogicalVariable channelSubscriptionIdVar = subscriptionsScan.getVariables().get(0);
 
         //The channelExecutionTime is created just before the scan
         ILogicalOperator channelExecutionAssign = subscriptionsScan.getInputs().get(0).getValue();
@@ -155,25 +151,48 @@
             ((CommitOperator) ((DelegateOperator) op).getDelegate()).setSink(false);
         }
 
-        AssignOperator assignOp = createbrokerEndPointAssignOperator(brokerEndpointVar, opAboveBrokersScan);
-        //now brokerNameVar holds the brokerName for use farther up in the plan
+        //move broker scan
+        AbstractLogicalOperator opAboveBrokersScan = findOp(op, 1, "", "");
+        if (opAboveBrokersScan == null) {
+            return false;
+        }
+        DataSourceScanOperator brokerScan =
+                moveScans(opAboveBrokersScan, op1, context, channelName + "BrokerSubscriptions");
 
-        context.computeAndSetTypeEnvironmentForOperator(assignOp);
-        context.computeAndSetTypeEnvironmentForOperator(opAboveBrokersScan);
-        context.computeAndSetTypeEnvironmentForOperator(op);
+        if (brokerScan == null) {
+            return false;
+        }
+        DataSourceScanOperator brokerSubscriptionScan =
+                (DataSourceScanOperator) brokerScan.getInputs().get(0).getValue();
 
-        ProjectOperator badProject = (ProjectOperator) findOp(op1, "project");
-        badProject.getVariables().add(subscriptionIdVar);
-        badProject.getVariables().add(brokerEndpointVar);
+        //Add select to join subscriptions and broker and assign to get endpoint
+        LogicalVariable brokerDataverseVar = brokerScan.getVariables().get(0);
+        LogicalVariable brokerNameVar = brokerScan.getVariables().get(1);
+        LogicalVariable brokerVar = brokerScan.getVariables().get(2);
+        LogicalVariable brokerSubscriptionVar = brokerSubscriptionScan.getVariables().get(2);
+        LogicalVariable brokerSubscriptionIdVar = brokerSubscriptionScan.getVariables().get(1);
+        LogicalVariable brokerSubscriptionChannelIdVar = brokerSubscriptionScan.getVariables().get(0);
+
+        LogicalVariable brokerEndpointVar = context.newVar();
+        AssignOperator assign = createAssignAndSelect(brokerDataverseVar, brokerNameVar, brokerVar,
+                brokerSubscriptionVar, brokerSubscriptionIdVar, brokerEndpointVar, channelSubscriptionIdVar,
+                brokerSubscriptionChannelIdVar, context, op1);
+
+        context.computeAndSetTypeEnvironmentForOperator(op1);
+
+        //Maybe we need to add a project???
+        ProjectOperator badProject = (ProjectOperator) findOp(op1, 2, "", "");
         badProject.getVariables().add(channelExecutionVar);
+        badProject.getVariables().add(channelSubscriptionIdVar);
         context.computeAndSetTypeEnvironmentForOperator(badProject);
 
 
         //Create my brokerNotify plan above the extension Operator
         DelegateOperator dOp = push
                 ? createNotifyBrokerPushPlan(brokerEndpointVar, badProject.getVariables().get(0), channelExecutionVar,
-                        context, op, (DistributeResultOperator) op1, channelDataverse, channelName)
-                : createNotifyBrokerPullPlan(brokerEndpointVar, subscriptionIdVar, channelExecutionVar, context, op,
+                        context, assign, (DistributeResultOperator) op1, channelDataverse, channelName)
+                : createNotifyBrokerPullPlan(brokerEndpointVar, channelSubscriptionIdVar, channelExecutionVar, context,
+                        assign,
                         (DistributeResultOperator) op1, channelDataverse, channelName);
 
         opRef.setValue(dOp);
@@ -181,6 +200,98 @@
         return true;
     }
 
+    private AssignOperator createAssignAndSelect(LogicalVariable brokerDataverseVar, LogicalVariable brokerNameVar,
+            LogicalVariable brokerVar, LogicalVariable brokerSubscriptionVar, LogicalVariable brokerSubscriptionIdVar,
+            LogicalVariable brokerEndpointVar, LogicalVariable channelSubscriptionIdVar,
+            LogicalVariable brokerSubscriptionChannelIdVar, IOptimizationContext context, AbstractLogicalOperator op1)
+            throws AlgebricksException {
+
+        FunctionInfo finfoGetField =
+                (FunctionInfo) FunctionUtil.getFunctionInfo(BuiltinFunctions.FIELD_ACCESS_BY_INDEX);
+        FunctionInfo finfoGetEquality = (FunctionInfo) FunctionUtil.getFunctionInfo(AlgebricksBuiltinFunctions.EQ);
+        FunctionInfo finfoGetAnd = (FunctionInfo) FunctionUtil.getFunctionInfo(AlgebricksBuiltinFunctions.AND);
+
+        //Create Select Operator
+        //The operator matches (A) the broker dataverse and name between broker subscriptions and brokers
+        // and (B) the channel subscription Id between broker subscriptions and channel subscriptions
+        ScalarFunctionCallExpression getBrokerName = new ScalarFunctionCallExpression(finfoGetField,
+                new MutableObject<>(new VariableReferenceExpression(brokerSubscriptionVar)),
+                new MutableObject<>(new ConstantExpression(new AsterixConstantValue(new AInt32(3)))));
+        ScalarFunctionCallExpression getBrokerDataverse = new ScalarFunctionCallExpression(finfoGetField,
+                new MutableObject<>(new VariableReferenceExpression(brokerSubscriptionVar)),
+                new MutableObject<>(new ConstantExpression(new AsterixConstantValue(new AInt32(2)))));
+
+        VariableReferenceExpression BrokerBrokerNameReference = new VariableReferenceExpression(brokerNameVar);
+        VariableReferenceExpression BrokerBrokerDataverseReference =
+                new VariableReferenceExpression(brokerDataverseVar);
+        VariableReferenceExpression brokerSubscriptionChannelIdVarReference =
+                new VariableReferenceExpression(brokerSubscriptionChannelIdVar);
+
+        ScalarFunctionCallExpression brokerNameCheck = new ScalarFunctionCallExpression(finfoGetEquality,
+                new MutableObject<>(getBrokerName), new MutableObject<>(BrokerBrokerNameReference));
+        ScalarFunctionCallExpression brokerDataverseCheck = new ScalarFunctionCallExpression(finfoGetEquality,
+                new MutableObject<>(getBrokerDataverse), new MutableObject<>(BrokerBrokerDataverseReference));
+        ScalarFunctionCallExpression channelSubCheck = new ScalarFunctionCallExpression(finfoGetEquality,
+                new MutableObject<>(brokerSubscriptionChannelIdVarReference),
+                new MutableObject<>(new VariableReferenceExpression(channelSubscriptionIdVar)));
+
+        ScalarFunctionCallExpression andExpression =
+                new ScalarFunctionCallExpression(finfoGetAnd, new MutableObject<>(channelSubCheck),
+                        new MutableObject<>(brokerDataverseCheck), new MutableObject<>(brokerNameCheck));
+
+        SelectOperator select = new SelectOperator(new MutableObject<>(andExpression), false, null);
+        select.getInputs().addAll(op1.getInputs());
+
+        //Create Assign Operator
+        ScalarFunctionCallExpression getEndPoint = new ScalarFunctionCallExpression(finfoGetField,
+                new MutableObject<>(new VariableReferenceExpression(brokerVar)),
+                new MutableObject<>(new ConstantExpression(new AsterixConstantValue(new AInt32(2)))));
+        AssignOperator assign = new AssignOperator(brokerEndpointVar, new MutableObject<>(getEndPoint));
+        assign.getInputs().add(new MutableObject<>(select));
+
+        op1.getInputs().set(0, new MutableObject<>(assign));
+        context.computeAndSetTypeEnvironmentForOperator(select);
+        context.computeAndSetTypeEnvironmentForOperator(assign);
+        context.computeAndSetTypeEnvironmentForOperator(op1);
+
+        return assign;
+
+    }
+
+    private DataSourceScanOperator moveScans(AbstractLogicalOperator opAboveScan, AbstractLogicalOperator op1,
+            IOptimizationContext context, String subscriptionsName) throws AlgebricksException {
+
+        DataSourceScanOperator brokerScan = null;
+        int i = 0;
+        for (Mutable<ILogicalOperator> subOp : opAboveScan.getInputs()) {
+            if (isBrokerScan((AbstractLogicalOperator) subOp.getValue())) {
+                brokerScan = (DataSourceScanOperator) subOp.getValue();
+                break;
+            }
+            i++;
+        }
+        if (brokerScan == null) {
+            return null;
+        }
+
+        AbstractLogicalOperator brokerSubcriptionsScan =
+                (AbstractLogicalOperator) brokerScan.getInputs().get(0).getValue();
+
+        if (!isSubscriptionsScan(brokerSubcriptionsScan, subscriptionsName, BADConstants.BrokerSubscriptionsType)) {
+            return null;
+        }
+        opAboveScan.getInputs().set(i, brokerSubcriptionsScan.getInputs().get(0));
+        context.computeAndSetTypeEnvironmentForOperator(opAboveScan);
+
+        brokerSubcriptionsScan.getInputs().set(0, op1.getInputs().get(0));
+        op1.getInputs().set(0, new MutableObject<>(brokerScan));
+        context.computeAndSetTypeEnvironmentForOperator(brokerSubcriptionsScan);
+        context.computeAndSetTypeEnvironmentForOperator(brokerScan);
+        context.computeAndSetTypeEnvironmentForOperator(op1);
+        return brokerScan;
+
+    }
+
     private DelegateOperator createBrokerOp(LogicalVariable brokerEndpointVar, LogicalVariable sendVar,
             LogicalVariable channelExecutionVar, String channelDataverse, String channelName, boolean push,
             IAType resultType) {
@@ -310,49 +421,66 @@
     }
 
     /*This function is used to find specific operators within the plan, either
-     * A. The brokers dataset scan
-     * B. The subscriptions scan
-     * C. The highest project of the plan
+     * 1. The brokers dataset scan
+     * 2. The highest project of the plan
+     * 3. The subscriptions scan
+     * 4. Commit operator
+     *
+     * param1 is the name of the expected subscriptions dataset when searching for a subscriptions scan
+     * and param2 is the type of the subscriptions dataset (channel or broker)
      */
-    private AbstractLogicalOperator findOp(AbstractLogicalOperator op, String lookingForString) {
+    private AbstractLogicalOperator findOp(AbstractLogicalOperator op, int searchId, String param1, String param2) {
         if (!op.hasInputs()) {
             return null;
         }
         for (Mutable<ILogicalOperator> subOp : op.getInputs()) {
-            if (lookingForString.equals("brokers")) {
+            if (searchId == 1) {
                 if (isBrokerScan((AbstractLogicalOperator) subOp.getValue())) {
                     return op;
                 } else {
                     AbstractLogicalOperator nestedOp = findOp((AbstractLogicalOperator) subOp.getValue(),
-                            lookingForString);
+                            searchId, param1, param2);
                     if (nestedOp != null) {
                         return nestedOp;
                     }
                 }
 
-            } else if (lookingForString.equals("project")) {
+            } else if (searchId == 2) {
                 if (subOp.getValue().getOperatorTag() == LogicalOperatorTag.PROJECT) {
                     return (AbstractLogicalOperator) subOp.getValue();
                 } else {
                     AbstractLogicalOperator nestedOp = findOp((AbstractLogicalOperator) subOp.getValue(),
-                            lookingForString);
+                            searchId, param1, param2);
                     if (nestedOp != null) {
                         return nestedOp;
                     }
                 }
             }
 
-            else {
-                if (isSubscriptionsScan((AbstractLogicalOperator) subOp.getValue(), lookingForString)) {
+            else if (searchId == 3) {
+                if (isSubscriptionsScan((AbstractLogicalOperator) subOp.getValue(), param1, param2)) {
                     return (AbstractLogicalOperator) subOp.getValue();
                 } else {
                     AbstractLogicalOperator nestedOp = findOp((AbstractLogicalOperator) subOp.getValue(),
-                            lookingForString);
+                            searchId, param1, param2);
                     if (nestedOp != null) {
                         return nestedOp;
                     }
                 }
 
+            } else if (searchId == 4) {
+                if (subOp.getValue().getOperatorTag() == LogicalOperatorTag.DELEGATE_OPERATOR) {
+                    DelegateOperator dOp = (DelegateOperator) subOp.getValue();
+                    if (dOp.getDelegate() instanceof CommitOperator) {
+                        return (AbstractLogicalOperator) subOp.getValue();
+                    }
+                } else {
+                    AbstractLogicalOperator nestedOp =
+                            findOp((AbstractLogicalOperator) subOp.getValue(), searchId, param1, param2);
+                    if (nestedOp != null) {
+                        return nestedOp;
+                    }
+                }
             }
         }
         return null;
@@ -371,12 +499,12 @@
         return false;
     }
 
-    private boolean isSubscriptionsScan(AbstractLogicalOperator op, String subscriptionsName) {
+    private boolean isSubscriptionsScan(AbstractLogicalOperator op, String subscriptionsName, String subscriptionType) {
         if (op instanceof DataSourceScanOperator) {
             if (((DataSourceScanOperator) op).getDataSource() instanceof DatasetDataSource) {
                 DatasetDataSource dds = (DatasetDataSource) ((DataSourceScanOperator) op).getDataSource();
                 if (dds.getDataset().getItemTypeDataverseName().equals("Metadata")
-                        && dds.getDataset().getItemTypeName().equals("ChannelSubscriptionsType")) {
+                        && dds.getDataset().getItemTypeName().equals(subscriptionType)) {
                     if (subscriptionsName.equals("") || dds.getDataset().getDatasetName().equals(subscriptionsName)) {
                         return true;
                     }