diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/BTreeSearchPOperator.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/BTreeSearchPOperator.java
index 249173a..fd54891 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/BTreeSearchPOperator.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/BTreeSearchPOperator.java
@@ -143,14 +143,13 @@
                 }
                 DatasetFormatInfo formatInfo = dataset.getDatasetFormatInfo();
                 if (isPrimaryIndex && formatInfo.getFormat() == DatasetConfig.DatasetFormat.COLUMN) {
-                    IProjectionFiltrationInfo<?> projectionInfo = unnestMapOp.getDatasetProjectionInfo();
-                    IProjectionFiltrationInfo<?> metaProjectionInfo = unnestMapOp.getMetaProjectionInfo();
+                    IProjectionFiltrationInfo projectionFiltrationInfo = unnestMapOp.getProjectionFiltrationInfo();
                     ARecordType datasetType = (ARecordType) metadataProvider.findType(dataset);
                     ARecordType metaItemType = (ARecordType) metadataProvider.findMetaType(dataset);
                     datasetType = (ARecordType) metadataProvider.findTypeForDatasetWithoutType(datasetType,
                             metaItemType, dataset);
-                    tupleProjectorFactory = IndexUtil.createTupleProjectorFactory(context, formatInfo, projectionInfo,
-                            metaProjectionInfo, datasetType, metaItemType, dataset.getPrimaryKeys().size());
+                    tupleProjectorFactory = IndexUtil.createTupleProjectorFactory(context, typeEnv, formatInfo,
+                            projectionFiltrationInfo, datasetType, metaItemType, dataset.getPrimaryKeys().size());
                 }
                 break;
             case LEFT_OUTER_UNNEST_MAP:
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/RuleCollections.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/RuleCollections.java
index 68fb146..f757295 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/RuleCollections.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/RuleCollections.java
@@ -79,7 +79,7 @@
 import org.apache.asterix.optimizer.rules.PushLimitIntoPrimarySearchRule;
 import org.apache.asterix.optimizer.rules.PushProperJoinThroughProduct;
 import org.apache.asterix.optimizer.rules.PushSimilarityFunctionsBelowJoin;
-import org.apache.asterix.optimizer.rules.PushValueAccessToDataScanRule;
+import org.apache.asterix.optimizer.rules.PushValueAccessAndFilterDownRule;
 import org.apache.asterix.optimizer.rules.RemoveDuplicateFieldsRule;
 import org.apache.asterix.optimizer.rules.RemoveLeftOuterUnnestForLeftOuterJoinRule;
 import org.apache.asterix.optimizer.rules.RemoveOrReplaceDefaultNullCastRule;
@@ -427,7 +427,7 @@
          * Must run IntroduceProjectsRule before PushValueAccessToDataScanRule to ensure that no entire records are
          * returned if they are projected out
          */
-        physicalRewritesTopLevel.add(new PushValueAccessToDataScanRule());
+        physicalRewritesTopLevel.add(new PushValueAccessAndFilterDownRule());
         physicalRewritesTopLevel.add(new SetAsterixPhysicalOperatorsRule());
         physicalRewritesTopLevel.add(new IntroduceRapidFrameFlushProjectAssignRule());
         physicalRewritesTopLevel.add(new SetExecutionModeRule());
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/PushValueAccessAndFilterDownRule.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/PushValueAccessAndFilterDownRule.java
new file mode 100644
index 0000000..b8f66f7
--- /dev/null
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/PushValueAccessAndFilterDownRule.java
@@ -0,0 +1,149 @@
+/*
+ * 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.optimizer.rules;
+
+import java.util.Set;
+
+import org.apache.asterix.common.metadata.DataverseName;
+import org.apache.asterix.metadata.declared.DataSource;
+import org.apache.asterix.metadata.declared.MetadataProvider;
+import org.apache.asterix.metadata.entities.Dataset;
+import org.apache.asterix.metadata.utils.DatasetUtil;
+import org.apache.asterix.optimizer.base.AsterixOptimizationContext;
+import org.apache.asterix.optimizer.rules.pushdown.PushdownContext;
+import org.apache.asterix.optimizer.rules.pushdown.processor.ColumnFilterPushdownProcessor;
+import org.apache.asterix.optimizer.rules.pushdown.processor.ColumnRangeFilterPushdownProcessor;
+import org.apache.asterix.optimizer.rules.pushdown.processor.ColumnValueAccessPushdownProcessor;
+import org.apache.asterix.optimizer.rules.pushdown.processor.InlineFilterExpressionsProcessor;
+import org.apache.asterix.optimizer.rules.pushdown.processor.PushdownProcessorsExecutor;
+import org.apache.asterix.optimizer.rules.pushdown.visitor.PushdownOperatorVisitor;
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+
+import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
+import it.unimi.dsi.fastutil.objects.ObjectSet;
+
+/**
+ * Pushes value access expressions to datasets' scans (if they permit) to minimize the size of the record.
+ * This rule currently does not remove the value access expression. Instead, it adds the requested field names to
+ * data-scan operator to produce records that only contain the requested values. The rule also pushes down filter
+ * expressions to data-scans if scanned datasets permit. This rule does not change the plan's structure after firing.
+ * Example:
+ * Before plan:
+ * ...
+ * select (and(gt($$00, 20), gt($$r.getField("salary"), 70000)))
+ * ...
+ * assign [$$00] <- [$$r.getField("personalInfo").getField("age")]
+ * ...
+ * data-scan []<-[$$0, $$r] <- ColumnDataverse.ColumnDataset
+ * <p>
+ * After plan:
+ * ...
+ * select (and(gt($$00, 20), gt($$r.getField("salary"), 70000)))
+ * ...
+ * assign [$$00] <- [$$r.getField("personalInfo").getField("age")]
+ * ...
+ * data-scan []<-[$$0, $$r] <- ColumnDataverse.ColumnDataset
+ * project ({personalInfo:{age: any},salary: any})
+ * filter on: and(gt($r.getField("personalInfo").getField("age"), 20), gt($$r.getField("salary"), 70000))
+ * range-filter on: and(gt($r.getField("personalInfo").getField("age"), 20), gt($$r.getField("salary"), 70000))
+ * <p>
+ * The resulting record $$r will be {"personalInfo":{"age": *AGE*}, "salary": *SALARY*}
+ * and other fields will not be included in $$r.
+ */
+public class PushValueAccessAndFilterDownRule implements IAlgebraicRewriteRule {
+    private boolean run = true;
+
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
+        if (!context.getPhysicalOptimizationConfig().isExternalFieldPushdown() || !run) {
+            //The rule was fired, or value access pushdown is disabled
+            return false;
+        }
+
+        /*
+         * Only run this rewrite rule once and only if the plan contains a data-scan on a dataset that
+         * supports value-access, filter, and/or range-filter.
+         */
+        run = shouldRun(context);
+        if (run) {
+            // Context holds all the necessary information to perform pushdowns
+            PushdownContext pushdownContext = new PushdownContext();
+            // Compute all the necessary pushdown information and performs inter-operator pushdown optimizations
+            PushdownOperatorVisitor pushdownInfoComputer = new PushdownOperatorVisitor(pushdownContext, context);
+            opRef.getValue().accept(pushdownInfoComputer, null);
+            // Execute several optimization passes to perform the pushdown
+            PushdownProcessorsExecutor pushdownProcessorsExecutor = new PushdownProcessorsExecutor();
+            addProcessors(pushdownProcessorsExecutor, pushdownContext, context);
+            pushdownProcessorsExecutor.execute();
+            pushdownProcessorsExecutor.finalizePushdown(pushdownContext, context);
+            run = false;
+        }
+        return false;
+    }
+
+    private void addProcessors(PushdownProcessorsExecutor pushdownProcessorsExecutor, PushdownContext pushdownContext,
+            IOptimizationContext context) {
+        // Performs value-access pushdowns
+        pushdownProcessorsExecutor.add(new ColumnValueAccessPushdownProcessor(pushdownContext, context));
+        if (context.getPhysicalOptimizationConfig().isColumnFilterEnabled()) {
+            // Performs filter pushdowns
+            pushdownProcessorsExecutor.add(new ColumnFilterPushdownProcessor(pushdownContext, context));
+            // Perform range-filter pushdowns
+            pushdownProcessorsExecutor.add(new ColumnRangeFilterPushdownProcessor(pushdownContext, context));
+            // Inline AND/OR expression
+            pushdownProcessorsExecutor.add(new InlineFilterExpressionsProcessor(pushdownContext, context));
+        }
+    }
+
+    /**
+     * Check whether the plan contains a dataset that supports pushdown
+     *
+     * @param context optimization context
+     * @return true if the plan contains such dataset, false otherwise
+     */
+    private boolean shouldRun(IOptimizationContext context) throws AlgebricksException {
+        ObjectSet<Int2ObjectMap.Entry<Set<DataSource>>> entrySet =
+                ((AsterixOptimizationContext) context).getDataSourceMap().int2ObjectEntrySet();
+        MetadataProvider metadataProvider = (MetadataProvider) context.getMetadataProvider();
+        for (Int2ObjectMap.Entry<Set<DataSource>> dataSources : entrySet) {
+            for (DataSource dataSource : dataSources.getValue()) {
+                if (supportPushdown(metadataProvider, dataSource)) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
+    private boolean supportPushdown(MetadataProvider metadataProvider, DataSource dataSource)
+            throws AlgebricksException {
+        DataverseName dataverse = dataSource.getId().getDataverseName();
+        String datasetName = dataSource.getId().getDatasourceName();
+        Dataset dataset = metadataProvider.findDataset(dataverse, datasetName);
+
+        return dataset != null && (DatasetUtil.isFieldAccessPushdownSupported(dataset)
+                || DatasetUtil.isFilterPushdownSupported(dataset)
+                || DatasetUtil.isRangeFilterPushdownSupported(dataset));
+    }
+}
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/PushValueAccessToDataScanRule.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/PushValueAccessToDataScanRule.java
deleted file mode 100644
index d43a5d1..0000000
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/PushValueAccessToDataScanRule.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/*
- * 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.optimizer.rules;
-
-import java.util.Set;
-
-import org.apache.asterix.common.config.DatasetConfig;
-import org.apache.asterix.common.metadata.DataverseName;
-import org.apache.asterix.external.util.ExternalDataUtils;
-import org.apache.asterix.metadata.declared.DataSource;
-import org.apache.asterix.metadata.declared.MetadataProvider;
-import org.apache.asterix.metadata.entities.Dataset;
-import org.apache.asterix.metadata.entities.ExternalDatasetDetails;
-import org.apache.asterix.optimizer.base.AsterixOptimizationContext;
-import org.apache.asterix.optimizer.rules.pushdown.OperatorValueAccessPushdownVisitor;
-import org.apache.commons.lang3.mutable.Mutable;
-import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
-import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
-import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
-import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
-
-import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
-import it.unimi.dsi.fastutil.objects.ObjectSet;
-
-/**
- * Pushes value access expressions to the external dataset scan to minimize the size of the record.
- * This rule currently does not remove the value access expression. Instead, it adds the requested field names to
- * external dataset details to produce records that only contain the requested values. Thus, no changes would occur
- * to the plan's structure after firing this rule.
- * Example:
- * Before plan:
- * ...
- * select (and(gt($$00, 20), gt($$r.getField("salary"), 70000)))
- * ...
- * assign [$$00] <- [$$r.getField("personalInfo").getField("age")]
- * ...
- * data-scan []<-[$$r] <- ParquetDataverse.ParquetDataset
- * <p>
- * After plan:
- * ...
- * select (and(gt($$00, 20), gt($$r.getField("salary"), 70000)))
- * ...
- * assign [$$00] <- [$$r.getField("personalInfo").getField("age")]
- * ...
- * data-scan []<-[$$r] <- ParquetDataverse.ParquetDataset project ({personalInfo:{age: VALUE},salary:VALUE})
- * <p>
- * The resulting record $$r will be {"personalInfo":{"age": *AGE*}, "salary": *SALARY*}
- * and other fields will not be included in $$r.
- */
-public class PushValueAccessToDataScanRule implements IAlgebraicRewriteRule {
-    //Initially, assume we need to run the rule
-    private boolean run = true;
-
-    @Override
-    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
-            throws AlgebricksException {
-        if (!context.getPhysicalOptimizationConfig().isExternalFieldPushdown() || !run) {
-            //The rule was fired, or value access pushdown is disabled
-            return false;
-        }
-
-        /*
-         * Only run the rewrite rule once and only if the plan contains a data-scan on a dataset that
-         * support value access pushdown.
-         */
-        run = shouldRun(context);
-        if (run) {
-            run = false;
-            OperatorValueAccessPushdownVisitor visitor = new OperatorValueAccessPushdownVisitor(context);
-            opRef.getValue().accept(visitor, null);
-            visitor.finish();
-        }
-
-        //This rule does not do any actual structural changes to the plan
-        return false;
-    }
-
-    /**
-     * Check whether the plan contains a dataset that supports pushdown
-     *
-     * @param context optimization context
-     * @return true if the plan contains such dataset, false otherwise
-     */
-    private boolean shouldRun(IOptimizationContext context) throws AlgebricksException {
-        ObjectSet<Int2ObjectMap.Entry<Set<DataSource>>> entrySet =
-                ((AsterixOptimizationContext) context).getDataSourceMap().int2ObjectEntrySet();
-        MetadataProvider metadataProvider = (MetadataProvider) context.getMetadataProvider();
-        for (Int2ObjectMap.Entry<Set<DataSource>> dataSources : entrySet) {
-            for (DataSource dataSource : dataSources.getValue()) {
-                if (supportPushdown(metadataProvider, dataSource)) {
-                    return true;
-                }
-            }
-        }
-        return false;
-    }
-
-    private boolean supportPushdown(MetadataProvider metadataProvider, DataSource dataSource)
-            throws AlgebricksException {
-        DataverseName dataverse = dataSource.getId().getDataverseName();
-        String datasetName = dataSource.getId().getDatasourceName();
-        Dataset dataset = metadataProvider.findDataset(dataverse, datasetName);
-
-        return dataset != null && ((dataset.getDatasetType() == DatasetConfig.DatasetType.EXTERNAL && ExternalDataUtils
-                .supportsPushdown(((ExternalDatasetDetails) dataset.getDatasetDetails()).getProperties()))
-                || dataset.getDatasetFormatInfo().getFormat() == DatasetConfig.DatasetFormat.COLUMN);
-    }
-}
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/ExpressionValueFilterPushdown.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/ExpressionValueFilterPushdown.java
deleted file mode 100644
index 29a2465..0000000
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/ExpressionValueFilterPushdown.java
+++ /dev/null
@@ -1,314 +0,0 @@
-/*
- * 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.optimizer.rules.pushdown;
-
-import static org.apache.asterix.metadata.utils.filter.NormalizedColumnFilterBuilder.COMPARE_FUNCTIONS;
-import static org.apache.asterix.metadata.utils.filter.NormalizedColumnFilterBuilder.NORMALIZED_PUSHABLE_FUNCTIONS;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.asterix.common.config.DatasetConfig;
-import org.apache.asterix.metadata.declared.DatasetDataSource;
-import org.apache.asterix.metadata.entities.Dataset;
-import org.apache.asterix.metadata.utils.filter.ColumnFilterBuilder;
-import org.apache.asterix.om.base.IAObject;
-import org.apache.asterix.om.constants.AsterixConstantValue;
-import org.apache.asterix.om.functions.BuiltinFunctions;
-import org.apache.asterix.om.types.ARecordType;
-import org.apache.asterix.optimizer.rules.pushdown.processor.ColumnRangeFilterPushdownProcessor;
-import org.apache.asterix.optimizer.rules.pushdown.schema.AnyExpectedSchemaNode;
-import org.apache.asterix.optimizer.rules.pushdown.schema.ExpectedSchemaBuilder;
-import org.apache.asterix.optimizer.rules.pushdown.schema.ExpectedSchemaNodeType;
-import org.apache.asterix.optimizer.rules.pushdown.schema.IExpectedSchemaNode;
-import org.apache.asterix.optimizer.rules.pushdown.visitor.ColumnFilterPathBuilderVisitor;
-import org.apache.asterix.runtime.projection.FunctionCallInformation;
-import org.apache.asterix.runtime.projection.ProjectionFiltrationWarningFactoryProvider;
-import org.apache.commons.lang3.mutable.Mutable;
-import org.apache.commons.lang3.mutable.MutableObject;
-import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
-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.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.functions.AlgebricksBuiltinFunctions;
-import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
-import org.apache.hyracks.algebricks.core.algebra.functions.IFunctionInfo;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractScanOperator;
-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.api.exceptions.SourceLocation;
-
-/**
- * Pushdown {@link SelectOperator} condition to the dataset to allow filtering mega leaf nodes.
- * This is currently only allowed for {@link DatasetConfig.DatasetFormat#COLUMN}
- * TODO Filter could prevent REPLICATE (i.e., we can scan a dataset twice due to the fact one scan is filtered and
- * TODO the other is not or both have different filters)
- * TODO part of this class could potentially be used for external data dynamic prefixes
- *
- * @deprecated use {@link ColumnRangeFilterPushdownProcessor} and {@link ColumnFilterPathBuilderVisitor}
- */
-class ExpressionValueFilterPushdown {
-    private final ExpectedSchemaBuilder builder;
-    private final ColumnFilterPathBuilderVisitor pathBuilder;
-    private final Map<AbstractScanOperator, Map<ILogicalExpression, ARecordType>> normalizedFilterPaths;
-    private final Map<AbstractScanOperator, Map<ILogicalExpression, ARecordType>> actualFilterPaths;
-    private final Map<AbstractScanOperator, ILogicalExpression> datasetFilterExpression;
-    private final Map<AbstractScanOperator, Map<String, FunctionCallInformation>> scanSourceInformationMaps;
-    private final Set<AbstractScanOperator> registeredScans;
-    private final HashMap<LogicalVariable, ILogicalExpression> registeredExpressions;
-    private final boolean columnFilterEnabled;
-
-    ExpressionValueFilterPushdown(ExpectedSchemaBuilder builder, boolean columnFilterEnabled) {
-        this.builder = builder;
-        pathBuilder = new ColumnFilterPathBuilderVisitor();
-        normalizedFilterPaths = new HashMap<>();
-        actualFilterPaths = new HashMap<>();
-        datasetFilterExpression = new HashMap<>();
-        scanSourceInformationMaps = new HashMap<>();
-        registeredScans = new HashSet<>();
-        registeredExpressions = new HashMap<>();
-        this.columnFilterEnabled = columnFilterEnabled;
-    }
-
-    public void registerDataset(AbstractScanOperator op, DatasetDataSource source) {
-        if (!columnFilterEnabled) {
-            return;
-        }
-
-        Dataset dataset = source.getDataset();
-        if (dataset.getDatasetType() == DatasetConfig.DatasetType.INTERNAL
-                && dataset.getDatasetFormatInfo().getFormat() == DatasetConfig.DatasetFormat.COLUMN) {
-            registeredScans.add(op);
-        }
-    }
-
-    public void registerExpression(LogicalVariable producedVar, ILogicalExpression expr) {
-        if (builder.getNode(expr) == null) {
-            // we only register expressions that do not correspond to a schema node
-            registeredExpressions.put(producedVar, expr);
-        }
-    }
-
-    /**
-     * Try to push the condition down to dataset scan/access
-     *
-     * @param selectOp the select operator
-     */
-    public void addFilterExpression(IOptimizationContext context, SelectOperator selectOp,
-            AbstractScanOperator scanOp) {
-        Map<ILogicalExpression, ARecordType> normalizedPaths = new HashMap<>();
-        Map<ILogicalExpression, ARecordType> actualPaths = new HashMap<>();
-        Map<String, FunctionCallInformation> sourceInformationMap = new HashMap<>();
-        ILogicalExpression conditionClone = selectOp.getCondition().getValue().cloneExpression();
-        Mutable<ILogicalExpression> conditionRef = new MutableObject<>(conditionClone);
-        if (addPaths(conditionRef, normalizedPaths, actualPaths, sourceInformationMap, true)) {
-            // Normalized paths
-            Map<ILogicalExpression, ARecordType> allNormalizedPaths =
-                    normalizedFilterPaths.computeIfAbsent(scanOp, k -> new HashMap<>());
-            allNormalizedPaths.putAll(normalizedPaths);
-
-            // Actual paths
-            Map<ILogicalExpression, ARecordType> allActualPaths =
-                    actualFilterPaths.computeIfAbsent(scanOp, k -> new HashMap<>());
-            allActualPaths.putAll(actualPaths);
-
-            // OR the previous condition with the current condition (if any)
-            // This might bring more than what's actually satisfies the predicate
-            putExpression(context, scanOp, conditionClone);
-            scanSourceInformationMaps.put(scanOp, sourceInformationMap);
-        }
-    }
-
-    private void putExpression(IOptimizationContext context, AbstractScanOperator scanOp,
-            ILogicalExpression conditionExpr) {
-        ILogicalExpression filterExpr = datasetFilterExpression.get(scanOp);
-        if (filterExpr != null) {
-            AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) filterExpr;
-            if (!BuiltinFunctions.OR.equals(funcExpr.getFunctionIdentifier())) {
-                IFunctionInfo fInfo = context.getMetadataProvider().lookupFunction(AlgebricksBuiltinFunctions.OR);
-                List<Mutable<ILogicalExpression>> args = new ArrayList<>();
-                args.add(new MutableObject<>(filterExpr));
-                funcExpr = new ScalarFunctionCallExpression(fInfo, args);
-                filterExpr = funcExpr;
-            }
-            funcExpr.getArguments().add(new MutableObject<>(conditionExpr));
-        } else {
-            filterExpr = conditionExpr;
-        }
-        datasetFilterExpression.put(scanOp, filterExpr);
-    }
-
-    public Map<ILogicalExpression, ARecordType> getNormalizedFilterPaths(AbstractScanOperator scanOp) {
-        return normalizedFilterPaths.getOrDefault(scanOp, Collections.emptyMap());
-    }
-
-    public Map<ILogicalExpression, ARecordType> getActualFilterPaths(AbstractScanOperator scanOp) {
-        return actualFilterPaths.getOrDefault(scanOp, Collections.emptyMap());
-    }
-
-    public ILogicalExpression getFilterExpression(AbstractScanOperator scanOp) {
-        return datasetFilterExpression.get(scanOp);
-    }
-
-    public Map<String, FunctionCallInformation> getSourceInformationMap(AbstractScanOperator scanOp) {
-        return scanSourceInformationMaps.getOrDefault(scanOp, new HashMap<>());
-    }
-
-    private boolean addPaths(Mutable<ILogicalExpression> exprRef, Map<ILogicalExpression, ARecordType> normalizedPaths,
-            Map<ILogicalExpression, ARecordType> actualPaths, Map<String, FunctionCallInformation> sourceInformationMap,
-            boolean includeNormalized) {
-        ILogicalExpression expr = exprRef.getValue();
-        if (expr.getExpressionTag() == LogicalExpressionTag.CONSTANT) {
-            IAObject constantValue = ((AsterixConstantValue) ((ConstantExpression) expr).getValue()).getObject();
-            // Continue if a primitive constant is encountered
-            return !constantValue.getType().getTypeTag().isDerivedType();
-        }
-
-        LogicalVariable variable = VariableUtilities.getVariable(expr);
-        if (variable != null && registeredExpressions.containsKey(variable)) {
-            // Inline the expression
-            ILogicalExpression currentExpr = registeredExpressions.get(variable);
-            exprRef.setValue(currentExpr);
-            return addPaths(exprRef, normalizedPaths, actualPaths, sourceInformationMap, false);
-        }
-
-        IExpectedSchemaNode node = builder.getNode(expr);
-        if (node != null) {
-            return addPath(node, expr, actualPaths);
-        }
-
-        if (!isFunctionExpression(expr)) {
-            return false;
-        }
-
-        AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) expr;
-        FunctionIdentifier fid = ((AbstractFunctionCallExpression) expr).getFunctionIdentifier();
-        if (!ColumnFilterBuilder.isPushable(fid)) {
-            return false;
-        }
-
-        boolean normalizedIncluded = includeNormalized && NORMALIZED_PUSHABLE_FUNCTIONS.contains(fid);
-
-        if (COMPARE_FUNCTIONS.contains(fid)) {
-            return handleCompare(funcExpr, normalizedPaths, actualPaths, sourceInformationMap, normalizedIncluded);
-        }
-        //AND/OR descend to the expression tree
-        return handleArgs(funcExpr, normalizedPaths, actualPaths, sourceInformationMap, includeNormalized);
-    }
-
-    private boolean handleCompare(AbstractFunctionCallExpression funcExpr,
-            Map<ILogicalExpression, ARecordType> normalizedPaths, Map<ILogicalExpression, ARecordType> actualPaths,
-            Map<String, FunctionCallInformation> sourceInformationMap, boolean includeNormalized) {
-        List<Mutable<ILogicalExpression>> args = funcExpr.getArguments();
-
-        Mutable<ILogicalExpression> leftRef = args.get(0);
-        Mutable<ILogicalExpression> rightRef = args.get(1);
-
-        ILogicalExpression left = leftRef.getValue();
-        ILogicalExpression right = rightRef.getValue();
-
-        if (isConstantExpression(left)) {
-            return handleCompare(funcExpr, rightRef, left, normalizedPaths, actualPaths, sourceInformationMap,
-                    includeNormalized, true);
-        } else if (isConstantExpression(right)) {
-            return handleCompare(funcExpr, leftRef, right, normalizedPaths, actualPaths, sourceInformationMap,
-                    includeNormalized, false);
-        }
-        // No constants, return false
-        return false;
-    }
-
-    private boolean handleCompare(AbstractFunctionCallExpression funcExpr, Mutable<ILogicalExpression> columnExprRef,
-            ILogicalExpression constExpr, Map<ILogicalExpression, ARecordType> normalizedPaths,
-            Map<ILogicalExpression, ARecordType> actualPaths, Map<String, FunctionCallInformation> sourceInformationMap,
-            boolean includeNormalized, boolean leftConstant) {
-
-        IAObject constantValue = ((AsterixConstantValue) ((ConstantExpression) constExpr).getValue()).getObject();
-        if (constantValue.getType().getTypeTag().isDerivedType()) {
-            // Cannot compare against nested values
-            return false;
-        }
-
-        ILogicalExpression columnExpr = columnExprRef.getValue();
-        IExpectedSchemaNode node = builder.getNode(columnExpr);
-        if (node == null || node.getType() != ExpectedSchemaNodeType.ANY) {
-            // Handle as a nested function call (e.g., numeric-add($$x, 1) > 10) where $$x is a value path
-            return addPaths(columnExprRef, normalizedPaths, actualPaths, null, false);
-        }
-
-        AnyExpectedSchemaNode leafNode = (AnyExpectedSchemaNode) node;
-
-        String functionName = funcExpr.getFunctionIdentifier().getName();
-        SourceLocation sourceLocation = funcExpr.getSourceLocation();
-        FunctionCallInformation functionCallInfo = new FunctionCallInformation(functionName, sourceLocation,
-                ProjectionFiltrationWarningFactoryProvider.getIncomparableTypesFactory(leftConstant));
-
-        ARecordType path = pathBuilder.buildPath(leafNode, constantValue, sourceInformationMap, functionCallInfo);
-        if (includeNormalized
-                && (!normalizedPaths.containsKey(columnExpr) || path.equals(normalizedPaths.get(columnExpr)))) {
-            normalizedPaths.put(columnExpr, path);
-        } else {
-            normalizedPaths.clear();
-        }
-        actualPaths.put(columnExpr, path);
-        return true;
-    }
-
-    private boolean addPath(IExpectedSchemaNode node, ILogicalExpression columnExpr,
-            Map<ILogicalExpression, ARecordType> actualPaths) {
-        if (node.getType() != ExpectedSchemaNodeType.ANY) {
-            return false;
-        }
-        AnyExpectedSchemaNode leafNode = (AnyExpectedSchemaNode) node;
-        ARecordType path = pathBuilder.buildPath(leafNode, null, null, null);
-        actualPaths.put(columnExpr, path);
-        return true;
-    }
-
-    private boolean handleArgs(AbstractFunctionCallExpression funcExpr,
-            Map<ILogicalExpression, ARecordType> normalizedPaths, Map<ILogicalExpression, ARecordType> actualPaths,
-            Map<String, FunctionCallInformation> sourceInformationMap, boolean includeNormalized) {
-        List<Mutable<ILogicalExpression>> args = funcExpr.getArguments();
-        boolean add = true;
-        for (int i = 0; add && i < args.size(); i++) {
-            add = addPaths(args.get(i), normalizedPaths, actualPaths, sourceInformationMap, includeNormalized);
-        }
-        return add;
-    }
-
-    private static boolean isFunctionExpression(ILogicalExpression expr) {
-        return expr.getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL;
-    }
-
-    private static boolean isConstantExpression(ILogicalExpression expr) {
-        return expr.getExpressionTag() == LogicalExpressionTag.CONSTANT;
-    }
-
-    public boolean allowsPushdown(AbstractScanOperator lastSeenScan) {
-        return columnFilterEnabled && lastSeenScan != null && registeredScans.contains(lastSeenScan);
-    }
-}
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/OperatorValueAccessPushdownVisitor.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/OperatorValueAccessPushdownVisitor.java
deleted file mode 100644
index 8e49bbb..0000000
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/OperatorValueAccessPushdownVisitor.java
+++ /dev/null
@@ -1,626 +0,0 @@
-/*
- * 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.optimizer.rules.pushdown;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.asterix.common.config.DatasetConfig;
-import org.apache.asterix.common.config.DatasetConfig.DatasetFormat;
-import org.apache.asterix.common.metadata.DataverseName;
-import org.apache.asterix.external.util.ExternalDataUtils;
-import org.apache.asterix.metadata.declared.DataSource;
-import org.apache.asterix.metadata.declared.DataSourceId;
-import org.apache.asterix.metadata.declared.DatasetDataSource;
-import org.apache.asterix.metadata.declared.MetadataProvider;
-import org.apache.asterix.metadata.entities.Dataset;
-import org.apache.asterix.metadata.entities.ExternalDatasetDetails;
-import org.apache.asterix.om.functions.BuiltinFunctions;
-import org.apache.asterix.om.types.ARecordType;
-import org.apache.asterix.om.utils.ConstantExpressionUtil;
-import org.apache.asterix.optimizer.rules.pushdown.schema.ExpectedSchemaBuilder;
-import org.apache.asterix.optimizer.rules.pushdown.schema.RootExpectedSchemaNode;
-import org.apache.asterix.runtime.projection.FunctionCallInformation;
-import org.apache.commons.lang3.mutable.Mutable;
-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.ILogicalPlan;
-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.IVariableTypeEnvironment;
-import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractScanOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.AggregateOperator;
-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.DelegateOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.DistinctOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.DistributeResultOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.EmptyTupleSourceOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.ExchangeOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.ForwardOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.GroupByOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.IndexInsertDeleteUpsertOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.InnerJoinOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.InsertDeleteUpsertOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.IntersectOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.LeftOuterJoinOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.LeftOuterUnnestMapOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.LeftOuterUnnestOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.LimitOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.MaterializeOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.NestedTupleSourceOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.OrderOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.ProjectOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.ReplicateOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.RunningAggregateOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.ScriptOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.SelectOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.SinkOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.SplitOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.SubplanOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.SwitchOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.TokenizeOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.UnionAllOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.UnnestMapOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.UnnestOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.WindowOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.WriteOperator;
-import org.apache.hyracks.algebricks.core.algebra.visitors.ILogicalOperatorVisitor;
-
-/**
- * This visitor visits the entire plan and tries to build the information of the required values from all dataset
- *
- * @deprecated use {@link org.apache.asterix.optimizer.rules.pushdown.visitor.PushdownOperatorVisitor}
- */
-@Deprecated
-public class OperatorValueAccessPushdownVisitor implements ILogicalOperatorVisitor<Void, Void> {
-    private static final List<LogicalVariable> EMPTY_VARIABLES = Collections.emptyList();
-
-    private final IOptimizationContext context;
-    //Requested schema builder. It is only expected schema not a definite one
-    private final ExpectedSchemaBuilder builder;
-    //To visit every expression in each operator
-    private final ExpressionValueAccessPushdownVisitor pushdownVisitor;
-    private final ExpressionValueFilterPushdown filterPushdown;
-    //Datasets that allow pushdowns
-    private final Map<LogicalVariable, AbstractScanOperator> registeredDatasets;
-    //Datasets that allow pushdowns and has meta
-    private final Map<LogicalVariable, AbstractScanOperator> registeredMetas;
-    //visitedOperators so we do not visit the same operator twice (in case of REPLICATE)
-    private final Set<ILogicalOperator> visitedOperators;
-    //Last scan operator seen
-    private AbstractScanOperator lastSeenScan;
-
-    public OperatorValueAccessPushdownVisitor(IOptimizationContext context) {
-        this.context = context;
-        builder = new ExpectedSchemaBuilder();
-        registeredDatasets = new HashMap<>();
-        registeredMetas = new HashMap<>();
-        pushdownVisitor = new ExpressionValueAccessPushdownVisitor(builder);
-        filterPushdown = new ExpressionValueFilterPushdown(builder,
-                context.getPhysicalOptimizationConfig().isColumnFilterEnabled());
-        visitedOperators = new HashSet<>();
-    }
-
-    public void finish() {
-        for (Map.Entry<LogicalVariable, AbstractScanOperator> entry : registeredDatasets.entrySet()) {
-            AbstractScanOperator scanOp = entry.getValue();
-            Map<ILogicalExpression, ARecordType> normalizedPaths = filterPushdown.getNormalizedFilterPaths(scanOp);
-            Map<ILogicalExpression, ARecordType> actualPaths = filterPushdown.getActualFilterPaths(scanOp);
-            ILogicalExpression filterExpression = filterPushdown.getFilterExpression(scanOp);
-            Map<String, FunctionCallInformation> sourceInformationMap = filterPushdown.getSourceInformationMap(scanOp);
-            if (scanOp.getOperatorTag() == LogicalOperatorTag.DATASOURCESCAN) {
-                DataSourceScanOperator scan = (DataSourceScanOperator) scanOp;
-                scan.setDatasetProjectionInfo(builder.createProjectionInfo(entry.getKey(), normalizedPaths, actualPaths,
-                        filterExpression, sourceInformationMap));
-            } else {
-                UnnestMapOperator unnest = (UnnestMapOperator) scanOp;
-                unnest.setDatasetProjectionInfo(builder.createProjectionInfo(entry.getKey(), normalizedPaths,
-                        actualPaths, filterExpression, sourceInformationMap));
-            }
-        }
-
-        for (Map.Entry<LogicalVariable, AbstractScanOperator> entry : registeredMetas.entrySet()) {
-            AbstractScanOperator abstractScan = entry.getValue();
-            if (abstractScan.getOperatorTag() == LogicalOperatorTag.DATASOURCESCAN) {
-                DataSourceScanOperator scan = (DataSourceScanOperator) abstractScan;
-                scan.setMetaProjectionInfo(builder.createProjectionInfo(entry.getKey()));
-            } else {
-                UnnestMapOperator unnest = (UnnestMapOperator) abstractScan;
-                unnest.setMetaProjectionInfo(builder.createProjectionInfo(entry.getKey()));
-            }
-        }
-    }
-
-    /**
-     * Visit every input of an operator. Then, start pushdown any value expression that the operator has
-     *
-     * @param op                the operator to process
-     * @param producedVariables any produced variables by the operator. We only care about the {@link AssignOperator}
-     *                          and {@link UnnestOperator} variables for now.
-     */
-    private void visitInputs(ILogicalOperator op, List<LogicalVariable> producedVariables) throws AlgebricksException {
-        if (visitedOperators.contains(op)) {
-            return;
-        }
-        for (Mutable<ILogicalOperator> child : op.getInputs()) {
-            child.getValue().accept(this, null);
-        }
-        IVariableTypeEnvironment typeEnv = op.computeOutputTypeEnvironment(context);
-        visitedOperators.add(op);
-        //Initiate the pushdown visitor
-        pushdownVisitor.init(producedVariables, typeEnv);
-        //pushdown any expression the operator has
-        op.acceptExpressionTransform(pushdownVisitor);
-
-        if (filterPushdown.allowsPushdown(lastSeenScan) && op.getOperatorTag() == LogicalOperatorTag.SELECT) {
-            //Push filters down
-            filterPushdown.addFilterExpression(context, (SelectOperator) op, lastSeenScan);
-        }
-
-        pushdownVisitor.end();
-    }
-
-    /*
-     * ******************************************************************************
-     * Operators that need to handle special cases
-     * ******************************************************************************
-     */
-
-    @Override
-    public Void visitProjectOperator(ProjectOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        //Set as empty records for data-scan or unnest-map if certain variables are projected out
-        setEmptyRecord(op.getInputs().get(0).getValue(), op.getVariables());
-        return null;
-    }
-
-    /**
-     * From the {@link DataSourceScanOperator}, we need to register the payload variable (record variable) to check
-     * which expression in the plan is using it.
-     */
-    @Override
-    public Void visitDataScanOperator(DataSourceScanOperator op, Void arg) throws AlgebricksException {
-        DatasetDataSource datasetDataSource = getDatasetDataSourceIfApplicable((DataSource) op.getDataSource());
-        registerDatasetIfApplicable(datasetDataSource, op);
-        visitInputs(op);
-        return null;
-    }
-
-    /**
-     * From the {@link UnnestMapOperator}, we need to register the payload variable (record variable) to check
-     * which expression in the plan is using it.
-     */
-    @Override
-    public Void visitUnnestMapOperator(UnnestMapOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        DatasetDataSource datasetDataSource = getDatasetDataSourceIfApplicable(getDataSourceFromUnnestMapOperator(op));
-        registerDatasetIfApplicable(datasetDataSource, op);
-        return null;
-    }
-
-    @Override
-    public Void visitAggregateOperator(AggregateOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        if (!op.isGlobal() && isCountConstant(op.getExpressions())) {
-            /*
-             * Optimize the SELECT COUNT(*) case
-             * It is local aggregate and has agg-sql-count function with a constant argument. Set empty record if the
-             * input operator is DataSourceScanOperator
-             */
-            setEmptyRecord(op.getInputs().get(0).getValue(), EMPTY_VARIABLES);
-        }
-        return null;
-    }
-
-    @Override
-    public Void visitAssignOperator(AssignOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op, op.getVariables());
-        if (filterPushdown.allowsPushdown(lastSeenScan)) {
-            List<LogicalVariable> variables = op.getVariables();
-            List<Mutable<ILogicalExpression>> exprs = op.getExpressions();
-            for (int i = 0; i < variables.size(); i++) {
-                // Register any potential expression that can be used by the pushed down to filter
-                filterPushdown.registerExpression(variables.get(i), exprs.get(i).getValue());
-            }
-        }
-        return null;
-    }
-
-    /*
-     * ******************************************************************************
-     * Helper methods
-     * ******************************************************************************
-     */
-
-    /**
-     * The role of this method is:
-     * 1- Check whether the datasource allows value access pushdowns
-     * 2- return the actual DatasetDataSource
-     */
-    private DatasetDataSource getDatasetDataSourceIfApplicable(DataSource dataSource) throws AlgebricksException {
-        if (dataSource == null || dataSource.getDatasourceType() == DataSource.Type.SAMPLE) {
-            return null;
-        }
-
-        MetadataProvider mp = (MetadataProvider) context.getMetadataProvider();
-        DataverseName dataverse = dataSource.getId().getDataverseName();
-        String datasetName = dataSource.getId().getDatasourceName();
-        Dataset dataset = mp.findDataset(dataverse, datasetName);
-
-        //Only external dataset can have pushed down expressions
-        if (dataset.getDatasetType() == DatasetConfig.DatasetType.EXTERNAL
-                && !ExternalDataUtils
-                        .supportsPushdown(((ExternalDatasetDetails) dataset.getDatasetDetails()).getProperties())
-                || dataset.getDatasetType() == DatasetConfig.DatasetType.INTERNAL
-                        && dataset.getDatasetFormatInfo().getFormat() == DatasetFormat.ROW) {
-            return null;
-        }
-
-        return (DatasetDataSource) dataSource;
-    }
-
-    /**
-     * Find datasource from {@link UnnestMapOperator}
-     *
-     * @param unnest unnest map operator
-     * @return datasource
-     */
-    private DataSource getDataSourceFromUnnestMapOperator(UnnestMapOperator unnest) throws AlgebricksException {
-        AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) unnest.getExpressionRef().getValue();
-        String dataverse = ConstantExpressionUtil.getStringArgument(funcExpr, 2);
-        String dataset = ConstantExpressionUtil.getStringArgument(funcExpr, 3);
-        if (!ConstantExpressionUtil.getStringArgument(funcExpr, 0).equals(dataset)) {
-            return null;
-        }
-
-        DataSourceId dsid = new DataSourceId(DataverseName.createFromCanonicalForm(dataverse), dataset);
-        MetadataProvider metadataProvider = (MetadataProvider) context.getMetadataProvider();
-        return metadataProvider.findDataSource(dsid);
-    }
-
-    private void registerDatasetIfApplicable(DatasetDataSource datasetDataSource, AbstractScanOperator op) {
-        if (datasetDataSource != null) {
-            LogicalVariable recordVar = datasetDataSource.getDataRecordVariable(op.getVariables());
-            if (!builder.isVariableRegistered(recordVar)) {
-                /*
-                 * This is the first time we see the dataset, and we know we might only need part of the record.
-                 * Register the dataset to prepare for value access expression pushdowns.
-                 * Initially, we will request the entire record.
-                 */
-                builder.registerRoot(recordVar, RootExpectedSchemaNode.ALL_FIELDS_ROOT_NODE);
-                if (op.getOperatorTag() == LogicalOperatorTag.DATASOURCESCAN) {
-                    // Not needed for secondary indexes
-                    filterPushdown.registerDataset(op, datasetDataSource);
-                }
-                registeredDatasets.put(recordVar, op);
-
-                if (datasetDataSource.hasMeta()) {
-                    /*
-                     * The dataset has meta. Register the meta root variable as another root for the dataset and add
-                     * it the metaVar to the registered metas
-                     */
-                    LogicalVariable metaVar = datasetDataSource.getMetaVariable(op.getVariables());
-                    builder.registerRoot(metaVar, RootExpectedSchemaNode.ALL_FIELDS_ROOT_NODE);
-                    registeredMetas.put(metaVar, op);
-                }
-            }
-            lastSeenScan = op;
-        }
-    }
-
-    /**
-     * If the inputOp is a {@link DataSourceScanOperator} or {@link UnnestMapOperator}, then set the projected value
-     * needed as empty record if any variable originated from either operators are not in {@code retainedVariables}
-     *
-     * @param inputOp           an operator that is potentially a {@link DataSourceScanOperator} or a {@link
-     *                          UnnestMapOperator}
-     * @param retainedVariables variables that should be retained
-     * @see #visitAggregateOperator(AggregateOperator, Void)
-     * @see #visitProjectOperator(ProjectOperator, Void)
-     */
-    private void setEmptyRecord(ILogicalOperator inputOp, List<LogicalVariable> retainedVariables)
-            throws AlgebricksException {
-        LogicalOperatorTag tag = inputOp.getOperatorTag();
-        if (tag != LogicalOperatorTag.DATASOURCESCAN && tag != LogicalOperatorTag.UNNEST_MAP) {
-            return;
-        }
-
-        DataSource dataSource;
-        List<LogicalVariable> variables;
-        Mutable<ILogicalExpression> selectCondition;
-        if (inputOp.getOperatorTag() == LogicalOperatorTag.DATASOURCESCAN) {
-            DataSourceScanOperator scan = (DataSourceScanOperator) inputOp;
-            dataSource = (DataSource) scan.getDataSource();
-            variables = scan.getVariables();
-            selectCondition = scan.getSelectCondition();
-        } else {
-            UnnestMapOperator unnest = (UnnestMapOperator) inputOp;
-            dataSource = getDataSourceFromUnnestMapOperator(unnest);
-            variables = unnest.getVariables();
-            selectCondition = unnest.getSelectCondition();
-        }
-
-        DatasetDataSource datasetDataSource = getDatasetDataSourceIfApplicable(dataSource);
-
-        if (datasetDataSource == null) {
-            //Does not support pushdown
-            return;
-        }
-
-        Set<LogicalVariable> selectConditionVariables = new HashSet<>();
-        if (selectCondition != null) {
-            //Get the used variables for a select condition
-            selectCondition.getValue().getUsedVariables(selectConditionVariables);
-        }
-
-        //We know that we only need the count of objects. So return empty objects only
-        LogicalVariable recordVar = datasetDataSource.getDataRecordVariable(variables);
-
-        /*
-         * If the recordVar is not retained by an upper operator and not used by a select condition, then return empty
-         * record instead of the entire record.
-         */
-        if (!retainedVariables.contains(recordVar) && !selectConditionVariables.contains(recordVar)) {
-            /*
-             * Set the root node as EMPTY_ROOT_NODE (i.e., no fields will be read from disk). We register the
-             * dataset with EMPTY_ROOT_NODE so that we skip pushdowns on empty node.
-             */
-            builder.registerRoot(recordVar, RootExpectedSchemaNode.EMPTY_ROOT_NODE);
-        }
-
-        if (datasetDataSource.hasMeta()) {
-            //Do the same for meta
-            LogicalVariable metaVar = datasetDataSource.getMetaVariable(variables);
-            if (!retainedVariables.contains(metaVar)) {
-                builder.registerRoot(metaVar, RootExpectedSchemaNode.EMPTY_ROOT_NODE);
-            }
-        }
-    }
-
-    private boolean isCountConstant(List<Mutable<ILogicalExpression>> expressions) {
-        if (expressions.size() != 1) {
-            return false;
-        }
-        ILogicalExpression expression = expressions.get(0).getValue();
-        if (expression.getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL) {
-            return false;
-        }
-        AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) expression;
-        FunctionIdentifier fid = funcExpr.getFunctionIdentifier();
-        return BuiltinFunctions.SQL_COUNT.equals(fid)
-                && funcExpr.getArguments().get(0).getValue().getExpressionTag() == LogicalExpressionTag.CONSTANT;
-    }
-
-    private void visitSubplans(List<ILogicalPlan> nestedPlans) throws AlgebricksException {
-        for (ILogicalPlan plan : nestedPlans) {
-            for (Mutable<ILogicalOperator> root : plan.getRoots()) {
-                visitInputs(root.getValue());
-            }
-        }
-    }
-
-    /*
-     * ******************************************************************************
-     * Pushdown when possible for each operator
-     * ******************************************************************************
-     */
-
-    @Override
-    public Void visitSelectOperator(SelectOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitSubplanOperator(SubplanOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        visitSubplans(op.getNestedPlans());
-        return null;
-    }
-
-    @Override
-    public Void visitUnnestOperator(UnnestOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op, op.getVariables());
-        return null;
-    }
-
-    @Override
-    public Void visitRunningAggregateOperator(RunningAggregateOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitEmptyTupleSourceOperator(EmptyTupleSourceOperator op, Void arg) throws AlgebricksException {
-        return null;
-    }
-
-    @Override
-    public Void visitGroupByOperator(GroupByOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        visitSubplans(op.getNestedPlans());
-        return null;
-    }
-
-    @Override
-    public Void visitLimitOperator(LimitOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitInnerJoinOperator(InnerJoinOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitLeftOuterJoinOperator(LeftOuterJoinOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitNestedTupleSourceOperator(NestedTupleSourceOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitOrderOperator(OrderOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitDelegateOperator(DelegateOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitReplicateOperator(ReplicateOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitSplitOperator(SplitOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitSwitchOperator(SwitchOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitMaterializeOperator(MaterializeOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitScriptOperator(ScriptOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitSinkOperator(SinkOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitUnionOperator(UnionAllOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitIntersectOperator(IntersectOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitLeftOuterUnnestOperator(LeftOuterUnnestOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitLeftOuterUnnestMapOperator(LeftOuterUnnestMapOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitDistinctOperator(DistinctOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitExchangeOperator(ExchangeOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitWriteOperator(WriteOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitDistributeResultOperator(DistributeResultOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitInsertDeleteUpsertOperator(InsertDeleteUpsertOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitIndexInsertDeleteUpsertOperator(IndexInsertDeleteUpsertOperator op, Void arg)
-            throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitTokenizeOperator(TokenizeOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitForwardOperator(ForwardOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        return null;
-    }
-
-    @Override
-    public Void visitWindowOperator(WindowOperator op, Void arg) throws AlgebricksException {
-        visitInputs(op);
-        visitSubplans(op.getNestedPlans());
-        return null;
-    }
-
-    private void visitInputs(ILogicalOperator op) throws AlgebricksException {
-        visitInputs(op, null);
-    }
-}
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/processor/ColumnValueAccessPushdownProcessor.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/processor/ColumnValueAccessPushdownProcessor.java
new file mode 100644
index 0000000..0715dc4
--- /dev/null
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/processor/ColumnValueAccessPushdownProcessor.java
@@ -0,0 +1,96 @@
+/*
+ * 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.optimizer.rules.pushdown.processor;
+
+import java.util.List;
+
+import org.apache.asterix.metadata.utils.DatasetUtil;
+import org.apache.asterix.optimizer.rules.pushdown.PushdownContext;
+import org.apache.asterix.optimizer.rules.pushdown.descriptor.DefineDescriptor;
+import org.apache.asterix.optimizer.rules.pushdown.descriptor.ScanDefineDescriptor;
+import org.apache.asterix.optimizer.rules.pushdown.descriptor.UseDescriptor;
+import org.apache.asterix.optimizer.rules.pushdown.schema.ExpectedSchemaBuilder;
+import org.apache.asterix.optimizer.rules.pushdown.schema.RootExpectedSchemaNode;
+import org.apache.asterix.optimizer.rules.pushdown.visitor.ExpressionValueAccessPushdownVisitor;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
+
+/**
+ * Computes the expected schema for columnar datasets (whether internal or external). The expected schema is then
+ * used to project only the values that were requested by the query.
+ */
+public class ColumnValueAccessPushdownProcessor extends AbstractPushdownProcessor {
+    private final ExpectedSchemaBuilder builder;
+    private final ExpressionValueAccessPushdownVisitor expressionVisitor;
+
+    public ColumnValueAccessPushdownProcessor(PushdownContext pushdownContext, IOptimizationContext context) {
+        super(pushdownContext, context);
+        builder = new ExpectedSchemaBuilder();
+        expressionVisitor = new ExpressionValueAccessPushdownVisitor(builder);
+    }
+
+    @Override
+    public void process() throws AlgebricksException {
+        List<ScanDefineDescriptor> scanDefineDescriptors = pushdownContext.getRegisteredScans();
+        for (ScanDefineDescriptor scanDefineDescriptor : scanDefineDescriptors) {
+            if (!DatasetUtil.isFieldAccessPushdownSupported(scanDefineDescriptor.getDataset())) {
+                continue;
+            }
+            pushdownFieldAccessForDataset(scanDefineDescriptor);
+            scanDefineDescriptor
+                    .setRecordNode((RootExpectedSchemaNode) builder.getNode(scanDefineDescriptor.getVariable()));
+            if (scanDefineDescriptor.hasMeta()) {
+                scanDefineDescriptor.setMetaNode(
+                        (RootExpectedSchemaNode) builder.getNode(scanDefineDescriptor.getMetaRecordVariable()));
+            }
+        }
+    }
+
+    private void pushdownFieldAccessForDataset(ScanDefineDescriptor scanDefineDescriptor) throws AlgebricksException {
+        builder.registerRoot(scanDefineDescriptor.getVariable(), scanDefineDescriptor.getRecordNode());
+        if (scanDefineDescriptor.hasMeta()) {
+            builder.registerRoot(scanDefineDescriptor.getMetaRecordVariable(), scanDefineDescriptor.getMetaNode());
+        }
+        pushdownFieldAccess(scanDefineDescriptor);
+    }
+
+    private void pushdownFieldAccess(DefineDescriptor defineDescriptor) throws AlgebricksException {
+        List<UseDescriptor> useDescriptors = pushdownContext.getUseDescriptors(defineDescriptor);
+        for (UseDescriptor useDescriptor : useDescriptors) {
+            LogicalVariable producedVariable = useDescriptor.getProducedVariable();
+            IVariableTypeEnvironment typeEnv = useDescriptor.getOperator().computeOutputTypeEnvironment(context);
+            expressionVisitor.transform(useDescriptor.getExpression(), producedVariable, typeEnv);
+        }
+
+        /*
+         * Two loops are needed as we need first to build the schemas for all useDescriptors expressions and then
+         * follow through (if the useDescriptor expression was assigned to a variable). In other words, the traversal
+         * of the expression tree has to be BFS and not DFS to prevent building a schema for undeclared variable.
+         * 'Undeclared variable' means we don't have a schema for a variable as we didn't visit it.
+         */
+        for (UseDescriptor useDescriptor : useDescriptors) {
+            DefineDescriptor nextDefineDescriptor = pushdownContext.getDefineDescriptor(useDescriptor);
+            if (nextDefineDescriptor != null) {
+                pushdownFieldAccess(nextDefineDescriptor);
+            }
+        }
+    }
+}
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/processor/PushdownProcessorsExecutor.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/processor/PushdownProcessorsExecutor.java
new file mode 100644
index 0000000..110cd29
--- /dev/null
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/processor/PushdownProcessorsExecutor.java
@@ -0,0 +1,129 @@
+/*
+ * 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.optimizer.rules.pushdown.processor;
+
+import static org.apache.asterix.om.utils.ProjectionFiltrationTypeUtil.ALL_FIELDS_TYPE;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.asterix.common.config.DatasetConfig;
+import org.apache.asterix.metadata.entities.Dataset;
+import org.apache.asterix.metadata.utils.DatasetUtil;
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.optimizer.rules.pushdown.PushdownContext;
+import org.apache.asterix.optimizer.rules.pushdown.descriptor.ScanDefineDescriptor;
+import org.apache.asterix.optimizer.rules.pushdown.visitor.ExpectedSchemaNodeToIATypeTranslatorVisitor;
+import org.apache.asterix.runtime.projection.ColumnDatasetProjectionFiltrationInfo;
+import org.apache.asterix.runtime.projection.ExternalDatasetProjectionInfo;
+import org.apache.asterix.runtime.projection.FunctionCallInformation;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.DefaultProjectionFiltrationInfo;
+import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import org.apache.hyracks.algebricks.core.algebra.metadata.IProjectionFiltrationInfo;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractScanOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.DataSourceScanOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.UnnestMapOperator;
+
+public class PushdownProcessorsExecutor {
+    private final List<IPushdownProcessor> processors;
+
+    public PushdownProcessorsExecutor() {
+        this.processors = new ArrayList<>();
+    }
+
+    public void add(IPushdownProcessor processor) {
+        processors.add(processor);
+    }
+
+    public void execute() throws AlgebricksException {
+        for (IPushdownProcessor processor : processors) {
+            processor.process();
+        }
+    }
+
+    public void finalizePushdown(PushdownContext pushdownContext, IOptimizationContext context) {
+        for (ScanDefineDescriptor scanDefineDescriptor : pushdownContext.getRegisteredScans()) {
+            Dataset dataset = scanDefineDescriptor.getDataset();
+            AbstractScanOperator scanOp = (AbstractScanOperator) scanDefineDescriptor.getOperator();
+            IProjectionFiltrationInfo info = null;
+            if (dataset.getDatasetFormatInfo().getFormat() == DatasetConfig.DatasetFormat.COLUMN) {
+                info = createInternalColumnarDatasetInfo(scanDefineDescriptor, context);
+            } else if (dataset.getDatasetType() == DatasetConfig.DatasetType.EXTERNAL
+                    && DatasetUtil.isFieldAccessPushdownSupported(dataset)) {
+                info = createExternalDatasetProjectionInfo(scanDefineDescriptor, context);
+            }
+            setInfoToDataScan(scanOp, info);
+        }
+    }
+
+    private IProjectionFiltrationInfo createInternalColumnarDatasetInfo(ScanDefineDescriptor scanDefineDescriptor,
+            IOptimizationContext context) {
+        Map<String, FunctionCallInformation> pathLocations = scanDefineDescriptor.getPathLocations();
+        ARecordType recordRequestedType = ALL_FIELDS_TYPE;
+        ARecordType metaRequestedType = scanDefineDescriptor.hasMeta() ? ALL_FIELDS_TYPE : null;
+
+        // Pushdown field access only if it is enabled
+        if (context.getPhysicalOptimizationConfig().isExternalFieldPushdown()) {
+            ExpectedSchemaNodeToIATypeTranslatorVisitor converter =
+                    new ExpectedSchemaNodeToIATypeTranslatorVisitor(pathLocations);
+            recordRequestedType = (ARecordType) scanDefineDescriptor.getRecordNode().accept(converter,
+                    scanDefineDescriptor.getDataset().getDatasetName());
+            if (metaRequestedType != null) {
+                metaRequestedType = (ARecordType) scanDefineDescriptor.getMetaNode().accept(converter,
+                        scanDefineDescriptor.getDataset().getDatasetName());
+            }
+        }
+
+        // Still allow for filter pushdowns even if value access pushdown is disabled
+        return new ColumnDatasetProjectionFiltrationInfo(recordRequestedType, metaRequestedType, pathLocations,
+                scanDefineDescriptor.getFilterPaths(), scanDefineDescriptor.getFilterExpression(),
+                scanDefineDescriptor.getRangeFilterExpression());
+    }
+
+    private IProjectionFiltrationInfo createExternalDatasetProjectionInfo(ScanDefineDescriptor scanDefineDescriptor,
+            IOptimizationContext context) {
+        if (!context.getPhysicalOptimizationConfig().isExternalFieldPushdown()) {
+            return DefaultProjectionFiltrationInfo.INSTANCE;
+        }
+
+        Map<String, FunctionCallInformation> pathLocations = scanDefineDescriptor.getPathLocations();
+        ExpectedSchemaNodeToIATypeTranslatorVisitor converter =
+                new ExpectedSchemaNodeToIATypeTranslatorVisitor(pathLocations);
+        ARecordType recordRequestedType = (ARecordType) scanDefineDescriptor.getRecordNode().accept(converter,
+                scanDefineDescriptor.getDataset().getDatasetName());
+        return new ExternalDatasetProjectionInfo(recordRequestedType, pathLocations);
+    }
+
+    private void setInfoToDataScan(AbstractScanOperator scanOp, IProjectionFiltrationInfo info) {
+        if (info == null) {
+            return;
+        }
+
+        if (scanOp.getOperatorTag() == LogicalOperatorTag.DATASOURCESCAN) {
+            DataSourceScanOperator dataScanOp = (DataSourceScanOperator) scanOp;
+            dataScanOp.setProjectionFiltrationInfo(info);
+        } else if (scanOp.getOperatorTag() == LogicalOperatorTag.UNNEST_MAP) {
+            UnnestMapOperator unnestMapOp = (UnnestMapOperator) scanOp;
+            unnestMapOp.setProjectionFiltrationInfo(info);
+        }
+    }
+}
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/schema/ExpectedSchemaBuilder.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/schema/ExpectedSchemaBuilder.java
index 345b545..71be18e 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/schema/ExpectedSchemaBuilder.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/schema/ExpectedSchemaBuilder.java
@@ -21,17 +21,12 @@
 import static org.apache.asterix.metadata.utils.PushdownUtil.ARRAY_FUNCTIONS;
 import static org.apache.asterix.metadata.utils.PushdownUtil.SUPPORTED_FUNCTIONS;
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 import org.apache.asterix.metadata.utils.PushdownUtil;
 import org.apache.asterix.om.functions.BuiltinFunctions;
-import org.apache.asterix.om.types.ARecordType;
-import org.apache.asterix.optimizer.rules.pushdown.visitor.ExpectedSchemaNodeToIATypeTranslatorVisitor;
-import org.apache.asterix.runtime.projection.DataProjectionFiltrationInfo;
-import org.apache.asterix.runtime.projection.FunctionCallInformation;
 import org.apache.commons.lang3.mutable.Mutable;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
@@ -54,30 +49,9 @@
 public class ExpectedSchemaBuilder {
     //Registered Variables
     private final Map<LogicalVariable, IExpectedSchemaNode> varToNode;
-    //Inferred type for expressions
-    private final Map<AbstractFunctionCallExpression, IExpectedSchemaNode> exprToNode;
 
     public ExpectedSchemaBuilder() {
         varToNode = new HashMap<>();
-        exprToNode = new HashMap<>();
-    }
-
-    public DataProjectionFiltrationInfo createProjectionInfo(LogicalVariable recordVariable) {
-        return createProjectionInfo(recordVariable, Collections.emptyMap(), Collections.emptyMap(), null,
-                Collections.emptyMap());
-    }
-
-    public DataProjectionFiltrationInfo createProjectionInfo(LogicalVariable recordVariable,
-            Map<ILogicalExpression, ARecordType> normalizedPaths, Map<ILogicalExpression, ARecordType> actualPaths,
-            ILogicalExpression filterExpression, Map<String, FunctionCallInformation> sourceInformationMap) {
-        IExpectedSchemaNode rootNode = varToNode.get(recordVariable);
-
-        ExpectedSchemaNodeToIATypeTranslatorVisitor typeBuilder =
-                new ExpectedSchemaNodeToIATypeTranslatorVisitor(sourceInformationMap);
-        ARecordType recordType = (ARecordType) rootNode.accept(typeBuilder, null);
-
-        return new DataProjectionFiltrationInfo(recordType, sourceInformationMap, normalizedPaths, actualPaths,
-                filterExpression);
     }
 
     public boolean setSchemaFromExpression(AbstractFunctionCallExpression expr, LogicalVariable producedVar,
@@ -88,9 +62,6 @@
             IExpectedSchemaNode leaf =
                     new AnyExpectedSchemaNode(parent, expr.getSourceLocation(), expr.getFunctionIdentifier().getName());
             addChild(expr, typeEnv, parent, leaf);
-
-            //Associate expression to node
-            exprToNode.put(expr, leaf);
             if (producedVar != null) {
                 //Register the node if a variable is produced
                 varToNode.put(producedVar, leaf);
@@ -124,21 +95,14 @@
         return varToNode.containsKey(variable);
     }
 
-    public boolean containsRegisteredDatasets() {
-        return !varToNode.isEmpty();
+    public boolean isEmpty() {
+        return varToNode.isEmpty();
     }
 
-    IExpectedSchemaNode getNode(LogicalVariable variable) {
+    public IExpectedSchemaNode getNode(LogicalVariable variable) {
         return varToNode.get(variable);
     }
 
-    public IExpectedSchemaNode getNode(ILogicalExpression expr) {
-        if (expr.getExpressionTag() == LogicalExpressionTag.VARIABLE) {
-            return getNode(VariableUtilities.getVariable(expr));
-        }
-        return exprToNode.get((AbstractFunctionCallExpression) expr);
-    }
-
     private IExpectedSchemaNode buildNestedNode(ILogicalExpression expr, IVariableTypeEnvironment typeEnv)
             throws AlgebricksException {
         //The current node expression
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/schema/ObjectExpectedSchemaNode.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/schema/ObjectExpectedSchemaNode.java
index 82a7214..ff89398 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/schema/ObjectExpectedSchemaNode.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/schema/ObjectExpectedSchemaNode.java
@@ -59,14 +59,6 @@
         children.replace(fieldName, newNode);
     }
 
-    protected IAType getType(IAType childType, IExpectedSchemaNode childNode, String typeName) {
-        String key = getChildFieldName(childNode);
-        IAType[] fieldTypes = { childType };
-        String[] fieldNames = { key };
-
-        return new ARecordType("typeName", fieldNames, fieldTypes, false);
-    }
-
     public String getChildFieldName(IExpectedSchemaNode requestedChild) {
         String key = null;
         for (Map.Entry<String, IExpectedSchemaNode> child : children.entrySet()) {
@@ -82,4 +74,12 @@
         }
         return key;
     }
+
+    protected IAType getType(IAType childType, IExpectedSchemaNode childNode, String typeName) {
+        String key = getChildFieldName(childNode);
+        IAType[] fieldTypes = { childType };
+        String[] fieldNames = { key };
+
+        return new ARecordType("typeName", fieldNames, fieldTypes, false);
+    }
 }
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/ExpressionValueAccessPushdownVisitor.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/visitor/ExpressionValueAccessPushdownVisitor.java
similarity index 75%
rename from asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/ExpressionValueAccessPushdownVisitor.java
rename to asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/visitor/ExpressionValueAccessPushdownVisitor.java
index c57f653..15c37d3 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/ExpressionValueAccessPushdownVisitor.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/pushdown/visitor/ExpressionValueAccessPushdownVisitor.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.asterix.optimizer.rules.pushdown;
+package org.apache.asterix.optimizer.rules.pushdown.visitor;
 
 import static org.apache.asterix.metadata.utils.PushdownUtil.ALLOWED_FUNCTIONS;
 import static org.apache.asterix.metadata.utils.PushdownUtil.SUPPORTED_FUNCTIONS;
@@ -32,53 +32,30 @@
 import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
 import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.visitors.VariableUtilities;
-import org.apache.hyracks.algebricks.core.algebra.visitors.ILogicalExpressionReferenceTransform;
 
-class ExpressionValueAccessPushdownVisitor implements ILogicalExpressionReferenceTransform {
+public class ExpressionValueAccessPushdownVisitor {
     private final ExpectedSchemaBuilder builder;
-    private List<LogicalVariable> producedVariables;
-    private IVariableTypeEnvironment typeEnv;
-    private int producedVariableIndex;
 
     public ExpressionValueAccessPushdownVisitor(ExpectedSchemaBuilder builder) {
         this.builder = builder;
-        end();
     }
 
-    public void init(List<LogicalVariable> producedVariables, IVariableTypeEnvironment typeEnv) {
-        this.producedVariables = producedVariables;
-        this.typeEnv = typeEnv;
-        producedVariableIndex = 0;
-    }
-
-    @Override
-    public boolean transform(Mutable<ILogicalExpression> expression) throws AlgebricksException {
-        if (producedVariableIndex == -1) {
-            //This for ensuring that the produced variables (if any) should be set
-            throw new IllegalStateException("init must be called first");
-        }
-        pushValueAccessExpression(expression, getNextProducedVariable(), typeEnv);
+    public boolean transform(ILogicalExpression expression, LogicalVariable producedVariable,
+            IVariableTypeEnvironment typeEnv) throws AlgebricksException {
+        pushValueAccessExpression(expression, producedVariable, typeEnv);
         return false;
     }
 
-    public void end() {
-        producedVariables = null;
-        typeEnv = null;
-        producedVariableIndex = -1;
-    }
-
-    private LogicalVariable getNextProducedVariable() {
-        LogicalVariable variable = producedVariables != null ? producedVariables.get(producedVariableIndex) : null;
-        producedVariableIndex++;
-        return variable;
+    private void pushValueAccessExpression(Mutable<ILogicalExpression> exprRef, LogicalVariable producedVar,
+            IVariableTypeEnvironment typeEnv) throws AlgebricksException {
+        pushValueAccessExpression(exprRef.getValue(), producedVar, typeEnv);
     }
 
     /**
      * Pushdown field access expressions and array access expressions down
      */
-    private void pushValueAccessExpression(Mutable<ILogicalExpression> exprRef, LogicalVariable producedVar,
+    private void pushValueAccessExpression(ILogicalExpression expr, LogicalVariable producedVar,
             IVariableTypeEnvironment typeEnv) throws AlgebricksException {
-        final ILogicalExpression expr = exprRef.getValue();
         if (skipPushdown(expr)) {
             return;
         }
@@ -91,11 +68,11 @@
         }
 
         //Check nested arguments if contains any pushable value access
-        pushValueAccessExpressionArg(funcExpr.getArguments(), producedVar);
+        pushValueAccessExpressionArg(funcExpr.getArguments(), producedVar, typeEnv);
     }
 
     /**
-     * Check if we can pushdown an expression. Also, unregister a variable if we found that a common expression value is
+     * Check if we can push down an expression. Also, unregister a variable if we found that a common expression value is
      * required in its entirety.
      */
     private boolean skipPushdown(ILogicalExpression expr) {
@@ -104,7 +81,7 @@
             unregisterVariableIfNeeded(variable);
             return true;
         }
-        return expr.getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL || !builder.containsRegisteredDatasets()
+        return expr.getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL || builder.isEmpty()
                 || isTypeCheckOnVariable(expr);
     }
 
@@ -139,8 +116,8 @@
                 && funcExpr.getArguments().get(0).getValue().getExpressionTag() == LogicalExpressionTag.VARIABLE;
     }
 
-    private void pushValueAccessExpressionArg(List<Mutable<ILogicalExpression>> exprList, LogicalVariable producedVar)
-            throws AlgebricksException {
+    private void pushValueAccessExpressionArg(List<Mutable<ILogicalExpression>> exprList, LogicalVariable producedVar,
+            IVariableTypeEnvironment typeEnv) throws AlgebricksException {
         for (Mutable<ILogicalExpression> exprRef : exprList) {
             /*
              * We need to set the produced variable as null here as the produced variable will not correspond to the
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/function/QueryIndexDatasource.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/function/QueryIndexDatasource.java
index de493ac..5178c3e 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/function/QueryIndexDatasource.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/function/QueryIndexDatasource.java
@@ -104,8 +104,7 @@
             List<LogicalVariable> minFilterVars, List<LogicalVariable> maxFilterVars,
             ITupleFilterFactory tupleFilterFactory, long outputLimit, IOperatorSchema opSchema,
             IVariableTypeEnvironment typeEnv, JobGenContext context, JobSpecification jobSpec, Object implConfig,
-            IProjectionFiltrationInfo<?> projectionInfo, IProjectionFiltrationInfo<?> metaProjectionInfo)
-            throws AlgebricksException {
+            IProjectionFiltrationInfo projectionInfo) throws AlgebricksException {
         return metadataProvider.getBtreeSearchRuntime(jobSpec, opSchema, typeEnv, context, true, false, null, ds,
                 indexName, null, null, true, true, false, null, null, null, tupleFilterFactory, outputLimit, false,
                 false, DefaultTupleProjectorFactory.INSTANCE, false);
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp.xml
index fc0ceca..5e09fa6 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp.xml
@@ -16226,45 +16226,46 @@
         <output-dir compare="Text">upsert/002</output-dir>
       </compilation-unit>
     </test-case>
-    <test-case FilePath="column" check-warnings="true">
-      <compilation-unit name="filter/001">
-        <output-dir compare="Text">filter/001</output-dir>
-        <expected-warn>ASX0051: Incomparable input types: string and bigint (in line 30, at column 23)</expected-warn>
-        <expected-warn>ASX0051: Incomparable input types: bigint and string (in line 29, at column 38)</expected-warn>
-        <expected-warn>ASX0051: Incomparable input types: array and bigint (in line 28, at column 15)</expected-warn>
-      </compilation-unit>
-    </test-case>
-    <test-case FilePath="column" check-warnings="true">
-      <compilation-unit name="filter/002">
-        <output-dir compare="Text">filter/002</output-dir>
-        <expected-warn>ASX0051: Incomparable input types: string and bigint (in line 29, at column 23)</expected-warn>
-      </compilation-unit>
-    </test-case>
-    <test-case FilePath="column">
-      <compilation-unit name="filter/003">
-        <output-dir compare="Text">filter/003</output-dir>
-      </compilation-unit>
-    </test-case>
-    <test-case FilePath="column">
-      <compilation-unit name="filter/004">
-        <output-dir compare="Text">filter/004</output-dir>
-      </compilation-unit>
-    </test-case>
-    <test-case FilePath="column">
-      <compilation-unit name="filter/005">
-        <output-dir compare="Text">filter/005</output-dir>
-      </compilation-unit>
-    </test-case>
-    <test-case FilePath="column">
-      <compilation-unit name="filter/006">
-        <output-dir compare="Text">filter/006</output-dir>
-      </compilation-unit>
-    </test-case>
-    <test-case FilePath="column">
-      <compilation-unit name="filter/007">
-        <output-dir compare="Text">filter/007</output-dir>
-      </compilation-unit>
-    </test-case>
+    <!--  Disabled until fully merge the remaining filter patches  -->
+<!--    <test-case FilePath="column" check-warnings="true">-->
+<!--      <compilation-unit name="filter/001">-->
+<!--        <output-dir compare="Text">filter/001</output-dir>-->
+<!--        <expected-warn>ASX0051: Incomparable input types: string and bigint (in line 30, at column 23)</expected-warn>-->
+<!--        <expected-warn>ASX0051: Incomparable input types: bigint and string (in line 29, at column 38)</expected-warn>-->
+<!--        <expected-warn>ASX0051: Incomparable input types: array and bigint (in line 28, at column 15)</expected-warn>-->
+<!--      </compilation-unit>-->
+<!--    </test-case>-->
+<!--    <test-case FilePath="column" check-warnings="true">-->
+<!--      <compilation-unit name="filter/002">-->
+<!--        <output-dir compare="Text">filter/002</output-dir>-->
+<!--        <expected-warn>ASX0051: Incomparable input types: string and bigint (in line 29, at column 23)</expected-warn>-->
+<!--      </compilation-unit>-->
+<!--    </test-case>-->
+<!--    <test-case FilePath="column">-->
+<!--      <compilation-unit name="filter/003">-->
+<!--        <output-dir compare="Text">filter/003</output-dir>-->
+<!--      </compilation-unit>-->
+<!--    </test-case>-->
+<!--    <test-case FilePath="column">-->
+<!--      <compilation-unit name="filter/004">-->
+<!--        <output-dir compare="Text">filter/004</output-dir>-->
+<!--      </compilation-unit>-->
+<!--    </test-case>-->
+<!--    <test-case FilePath="column">-->
+<!--      <compilation-unit name="filter/005">-->
+<!--        <output-dir compare="Text">filter/005</output-dir>-->
+<!--      </compilation-unit>-->
+<!--    </test-case>-->
+<!--    <test-case FilePath="column">-->
+<!--      <compilation-unit name="filter/006">-->
+<!--        <output-dir compare="Text">filter/006</output-dir>-->
+<!--      </compilation-unit>-->
+<!--    </test-case>-->
+<!--    <test-case FilePath="column">-->
+<!--      <compilation-unit name="filter/007">-->
+<!--        <output-dir compare="Text">filter/007</output-dir>-->
+<!--      </compilation-unit>-->
+<!--    </test-case>-->
     <test-case FilePath="column">
       <compilation-unit name="big-object">
         <output-dir compare="Text">big-object</output-dir>
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
index 084e1ae..02653f3 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
@@ -75,7 +75,7 @@
 import org.apache.asterix.om.types.IAType;
 import org.apache.asterix.om.types.TypeTagUtil;
 import org.apache.asterix.runtime.evaluators.common.NumberUtils;
-import org.apache.asterix.runtime.projection.DataProjectionFiltrationInfo;
+import org.apache.asterix.runtime.projection.ExternalDatasetProjectionInfo;
 import org.apache.asterix.runtime.projection.FunctionCallInformation;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -891,10 +891,10 @@
                 || ExternalDataConstants.FORMAT_PARQUET.equals(properties.get(ExternalDataConstants.KEY_FORMAT));
     }
 
-    public static void setExternalDataProjectionInfo(DataProjectionFiltrationInfo projectionInfo,
+    public static void setExternalDataProjectionInfo(ExternalDatasetProjectionInfo projectionInfo,
             Map<String, String> properties) throws IOException {
         properties.put(ExternalDataConstants.KEY_REQUESTED_FIELDS,
-                serializeExpectedTypeToString(projectionInfo.getProjectionInfo()));
+                serializeExpectedTypeToString(projectionInfo.getProjectedType()));
         properties.put(ExternalDataConstants.KEY_HADOOP_ASTERIX_FUNCTION_CALL_INFORMATION,
                 serializeFunctionCallInfoToString(projectionInfo.getFunctionCallInfoMap()));
     }
@@ -913,7 +913,7 @@
         ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
         DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
         Base64.Encoder encoder = Base64.getEncoder();
-        DataProjectionFiltrationInfo.writeTypeField(expectedType, dataOutputStream);
+        ExternalDatasetProjectionInfo.writeTypeField(expectedType, dataOutputStream);
         return encoder.encodeToString(byteArrayOutputStream.toByteArray());
     }
 
@@ -929,7 +929,7 @@
         ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
         DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
         Base64.Encoder encoder = Base64.getEncoder();
-        DataProjectionFiltrationInfo.writeFunctionCallInformationMapField(functionCallInfoMap, dataOutputStream);
+        ExternalDatasetProjectionInfo.writeFunctionCallInformationMapField(functionCallInfoMap, dataOutputStream);
         return encoder.encodeToString(byteArrayOutputStream.toByteArray());
     }
 
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
index 7b7b227..2618ae6 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
@@ -43,7 +43,7 @@
 import org.apache.asterix.external.input.stream.HDFSInputStream;
 import org.apache.asterix.external.util.ExternalDataConstants.ParquetOptions;
 import org.apache.asterix.om.types.ARecordType;
-import org.apache.asterix.runtime.projection.DataProjectionFiltrationInfo;
+import org.apache.asterix.runtime.projection.ExternalDatasetProjectionInfo;
 import org.apache.asterix.runtime.projection.FunctionCallInformation;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
@@ -271,7 +271,7 @@
         Base64.Decoder decoder = Base64.getDecoder();
         byte[] typeBytes = decoder.decode(encoded);
         DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(typeBytes));
-        return DataProjectionFiltrationInfo.createTypeField(dataInputStream);
+        return ExternalDatasetProjectionInfo.createTypeField(dataInputStream);
     }
 
     public static void setFunctionCallInformationMap(Map<String, FunctionCallInformation> funcCallInfoMap,
@@ -287,7 +287,7 @@
             Base64.Decoder decoder = Base64.getDecoder();
             byte[] functionCallInfoMapBytes = decoder.decode(encoded);
             DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(functionCallInfoMapBytes));
-            return DataProjectionFiltrationInfo.createFunctionCallInformationMap(dataInputStream);
+            return ExternalDatasetProjectionInfo.createFunctionCallInformationMap(dataInputStream);
         }
         return null;
     }
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/DataSource.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/DataSource.java
index 4e6119f..1fa84f6 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/DataSource.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/DataSource.java
@@ -166,6 +166,5 @@
             List<LogicalVariable> minFilterVars, List<LogicalVariable> maxFilterVars,
             ITupleFilterFactory tupleFilterFactory, long outputLimit, IOperatorSchema opSchema,
             IVariableTypeEnvironment typeEnv, JobGenContext context, JobSpecification jobSpec, Object implConfig,
-            IProjectionFiltrationInfo<?> datasetProjectionInfo, IProjectionFiltrationInfo<?> metaProjectionInfo)
-            throws AlgebricksException;
+            IProjectionFiltrationInfo projectionFiltrationInfo) throws AlgebricksException;
 }
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/DatasetDataSource.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/DatasetDataSource.java
index 9f5437b..9d4d508 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/DatasetDataSource.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/DatasetDataSource.java
@@ -41,10 +41,11 @@
 import org.apache.asterix.metadata.utils.KeyFieldTypeUtil;
 import org.apache.asterix.om.types.ARecordType;
 import org.apache.asterix.om.types.IAType;
-import org.apache.asterix.runtime.projection.DataProjectionFiltrationInfo;
+import org.apache.asterix.runtime.projection.ExternalDatasetProjectionInfo;
 import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.common.utils.Pair;
+import org.apache.hyracks.algebricks.core.algebra.base.DefaultProjectionFiltrationInfo;
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
 import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
 import org.apache.hyracks.algebricks.core.algebra.metadata.IDataSource;
@@ -121,8 +122,7 @@
             List<LogicalVariable> minFilterVars, List<LogicalVariable> maxFilterVars,
             ITupleFilterFactory tupleFilterFactory, long outputLimit, IOperatorSchema opSchema,
             IVariableTypeEnvironment typeEnv, JobGenContext context, JobSpecification jobSpec, Object implConfig,
-            IProjectionFiltrationInfo<?> projectionInfo, IProjectionFiltrationInfo<?> metaProjectionInfo)
-            throws AlgebricksException {
+            IProjectionFiltrationInfo projectionFiltrationInfo) throws AlgebricksException {
         String itemTypeName = dataset.getItemTypeName();
         IAType itemType = MetadataManager.INSTANCE
                 .getDatatype(metadataProvider.getMetadataTxnContext(), dataset.getItemTypeDataverseName(), itemTypeName)
@@ -134,7 +134,8 @@
                 ExternalDatasetDetails edd = (ExternalDatasetDetails) externalDataset.getDatasetDetails();
                 PhysicalOptimizationConfig physicalOptimizationConfig = context.getPhysicalOptimizationConfig();
                 int externalScanBufferSize = physicalOptimizationConfig.getExternalScanBufferSize();
-                Map<String, String> properties = addExternalProjectionInfo(projectionInfo, edd.getProperties());
+                Map<String, String> properties =
+                        addExternalProjectionInfo(projectionFiltrationInfo, edd.getProperties());
                 properties = addSubPath(externalDataSource.getProperties(), properties);
                 properties.put(KEY_EXTERNAL_SCAN_BUFFER_SIZE, String.valueOf(externalScanBufferSize));
                 ITypedAdapterFactory adapterFactory = metadataProvider.getConfiguredAdapterFactory(externalDataset,
@@ -158,8 +159,8 @@
                 }
                 int numberOfPrimaryKeys = dataset.getPrimaryKeys().size();
                 ITupleProjectorFactory tupleProjectorFactory =
-                        IndexUtil.createTupleProjectorFactory(context, dataset.getDatasetFormatInfo(), projectionInfo,
-                                metaProjectionInfo, datasetType, metaItemType, numberOfPrimaryKeys);
+                        IndexUtil.createTupleProjectorFactory(context, typeEnv, dataset.getDatasetFormatInfo(),
+                                projectionFiltrationInfo, datasetType, metaItemType, numberOfPrimaryKeys);
 
                 int[] minFilterFieldIndexes = createFilterIndexes(minFilterVars, opSchema);
                 int[] maxFilterFieldIndexes = createFilterIndexes(maxFilterVars, opSchema);
@@ -172,14 +173,14 @@
         }
     }
 
-    private Map<String, String> addExternalProjectionInfo(IProjectionFiltrationInfo<?> projectionInfo,
+    private Map<String, String> addExternalProjectionInfo(IProjectionFiltrationInfo projectionInfo,
             Map<String, String> properties) {
         Map<String, String> propertiesCopy = properties;
-        if (projectionInfo != null) {
+        if (projectionInfo != DefaultProjectionFiltrationInfo.INSTANCE) {
             //properties could be cached and reused, so we make a copy per query
             propertiesCopy = new HashMap<>(properties);
             try {
-                DataProjectionFiltrationInfo externalProjectionInfo = (DataProjectionFiltrationInfo) projectionInfo;
+                ExternalDatasetProjectionInfo externalProjectionInfo = (ExternalDatasetProjectionInfo) projectionInfo;
                 ExternalDataUtils.setExternalDataProjectionInfo(externalProjectionInfo, propertiesCopy);
             } catch (IOException e) {
                 throw new IllegalStateException(e);
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FeedDataSource.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FeedDataSource.java
index 0d65083..d975404 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FeedDataSource.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FeedDataSource.java
@@ -164,8 +164,7 @@
             List<LogicalVariable> minFilterVars, List<LogicalVariable> maxFilterVars,
             ITupleFilterFactory tupleFilterFactory, long outputLimit, IOperatorSchema opSchema,
             IVariableTypeEnvironment typeEnv, JobGenContext context, JobSpecification jobSpec, Object implConfig,
-            IProjectionFiltrationInfo<?> projectionInfo, IProjectionFiltrationInfo<?> metaProjectionInfo)
-            throws AlgebricksException {
+            IProjectionFiltrationInfo projectionFiltrationInfo) throws AlgebricksException {
         try {
             if (tupleFilterFactory != null || outputLimit >= 0) {
                 throw CompilationException.create(ErrorCode.COMPILATION_ILLEGAL_STATE,
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FunctionDataSource.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FunctionDataSource.java
index 2c57162..690338c 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FunctionDataSource.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FunctionDataSource.java
@@ -107,8 +107,7 @@
             List<LogicalVariable> minFilterVars, List<LogicalVariable> maxFilterVars,
             ITupleFilterFactory tupleFilterFactory, long outputLimit, IOperatorSchema opSchema,
             IVariableTypeEnvironment typeEnv, JobGenContext context, JobSpecification jobSpec, Object implConfig,
-            IProjectionFiltrationInfo<?> projectionInfo, IProjectionFiltrationInfo<?> metaProjectionInfo)
-            throws AlgebricksException {
+            IProjectionFiltrationInfo projectionFiltrationInfo) throws AlgebricksException {
         GenericAdapterFactory adapterFactory = new GenericAdapterFactory();
         adapterFactory.setOutputType(RecordUtil.FULLY_OPEN_RECORD_TYPE);
         IClusterStateManager csm = metadataProvider.getApplicationContext().getClusterStateManager();
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/LoadableDataSource.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/LoadableDataSource.java
index fc65c4b..ef95511 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/LoadableDataSource.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/LoadableDataSource.java
@@ -136,8 +136,7 @@
             List<LogicalVariable> minFilterVars, List<LogicalVariable> maxFilterVars,
             ITupleFilterFactory tupleFilterFactory, long outputLimit, IOperatorSchema opSchema,
             IVariableTypeEnvironment typeEnv, JobGenContext context, JobSpecification jobSpec, Object implConfig,
-            IProjectionFiltrationInfo<?> projectionInfo, IProjectionFiltrationInfo<?> metaProjectionInfo)
-            throws AlgebricksException {
+            IProjectionFiltrationInfo projectionFiltrationInfo) throws AlgebricksException {
         if (tupleFilterFactory != null || outputLimit >= 0) {
             throw CompilationException.create(ErrorCode.COMPILATION_ILLEGAL_STATE,
                     "tuple filter and limit are not supported by LoadableDataSource");
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
index e85e10a..c5b8ea3 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
@@ -490,11 +490,10 @@
             List<LogicalVariable> projectVariables, boolean projectPushed, List<LogicalVariable> minFilterVars,
             List<LogicalVariable> maxFilterVars, ITupleFilterFactory tupleFilterFactory, long outputLimit,
             IOperatorSchema opSchema, IVariableTypeEnvironment typeEnv, JobGenContext context, JobSpecification jobSpec,
-            Object implConfig, IProjectionFiltrationInfo<?> projectionInfo,
-            IProjectionFiltrationInfo<?> metaProjectionInfo) throws AlgebricksException {
+            Object implConfig, IProjectionFiltrationInfo projectionFiltrationInfo) throws AlgebricksException {
         return ((DataSource) dataSource).buildDatasourceScanRuntime(this, dataSource, scanVariables, projectVariables,
                 projectPushed, minFilterVars, maxFilterVars, tupleFilterFactory, outputLimit, opSchema, typeEnv,
-                context, jobSpec, implConfig, projectionInfo, metaProjectionInfo);
+                context, jobSpec, implConfig, projectionFiltrationInfo);
     }
 
     protected Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getLoadableDatasetScanRuntime(
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/SampleDataSource.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/SampleDataSource.java
index c1858af..3b2937b 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/SampleDataSource.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/SampleDataSource.java
@@ -60,8 +60,7 @@
             List<LogicalVariable> minFilterVars, List<LogicalVariable> maxFilterVars,
             ITupleFilterFactory tupleFilterFactory, long outputLimit, IOperatorSchema opSchema,
             IVariableTypeEnvironment typeEnv, JobGenContext context, JobSpecification jobSpec, Object implConfig,
-            IProjectionFiltrationInfo<?> projectionInfo, IProjectionFiltrationInfo<?> metaProjectionInfo)
-            throws AlgebricksException {
+            IProjectionFiltrationInfo projectionInfo) throws AlgebricksException {
         return metadataProvider.getBtreeSearchRuntime(jobSpec, opSchema, typeEnv, context, true, false, null, dataset,
                 sampleIndexName, null, null, true, true, false, null, null, null, tupleFilterFactory, outputLimit,
                 false, false, DefaultTupleProjectorFactory.INSTANCE, false);
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/IndexUtil.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/IndexUtil.java
index 136fb37..46b4a7b 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/IndexUtil.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/IndexUtil.java
@@ -50,11 +50,13 @@
 import org.apache.asterix.om.types.ARecordType;
 import org.apache.asterix.om.types.IAType;
 import org.apache.asterix.runtime.job.listener.JobEventListenerFactory;
-import org.apache.asterix.runtime.projection.DataProjectionFiltrationInfo;
+import org.apache.asterix.runtime.projection.ColumnDatasetProjectionFiltrationInfo;
 import org.apache.asterix.runtime.projection.FunctionCallInformation;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.common.utils.Pair;
 import org.apache.hyracks.algebricks.common.utils.Triple;
+import org.apache.hyracks.algebricks.core.algebra.base.DefaultProjectionFiltrationInfo;
+import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
 import org.apache.hyracks.algebricks.core.algebra.metadata.IProjectionFiltrationInfo;
 import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenContext;
@@ -270,42 +272,36 @@
     }
 
     public static ITupleProjectorFactory createTupleProjectorFactory(JobGenContext context,
-            DatasetFormatInfo datasetFormatInfo, IProjectionFiltrationInfo<?> projectionInfo,
-            IProjectionFiltrationInfo<?> metaProjectionInfo, ARecordType datasetType, ARecordType metaItemType,
+            IVariableTypeEnvironment typeEnv, DatasetFormatInfo datasetFormatInfo,
+            IProjectionFiltrationInfo projectionFiltrationInfo, ARecordType datasetType, ARecordType metaItemType,
             int numberOfPrimaryKeys) throws AlgebricksException {
         if (datasetFormatInfo.getFormat() == DatasetConfig.DatasetFormat.ROW) {
             return DefaultTupleProjectorFactory.INSTANCE;
         }
-        DataProjectionFiltrationInfo dataProjectionInfo = (DataProjectionFiltrationInfo) projectionInfo;
-        if (dataProjectionInfo == null) {
-            // projecting pushdown is disabled
+
+        if (projectionFiltrationInfo == DefaultProjectionFiltrationInfo.INSTANCE) {
+            // pushdown is disabled
             ARecordType metaType = metaItemType == null ? null : ALL_FIELDS_TYPE;
             return new QueryColumnTupleProjectorFactory(datasetType, metaItemType, numberOfPrimaryKeys, ALL_FIELDS_TYPE,
                     Collections.emptyMap(), metaType, Collections.emptyMap(), NoOpColumnFilterEvaluatorFactory.INSTANCE,
                     NoOpColumnFilterEvaluatorFactory.INSTANCE);
         }
+        ColumnDatasetProjectionFiltrationInfo columnInfo =
+                (ColumnDatasetProjectionFiltrationInfo) projectionFiltrationInfo;
 
-        DataProjectionFiltrationInfo metaDataProjectionInfo = (DataProjectionFiltrationInfo) metaProjectionInfo;
+        ARecordType recordRequestedType = columnInfo.getProjectedType();
+        ARecordType metaRequestedType = columnInfo.getMetaProjectedType();
+        Map<String, FunctionCallInformation> callInfo = columnInfo.getFunctionCallInfoMap();
 
-        ARecordType datasetRequestedType = dataProjectionInfo.getProjectionInfo();
-        Map<String, FunctionCallInformation> datasetFunctionCallInfo = dataProjectionInfo.getFunctionCallInfoMap();
-
-        ARecordType metaRequestedType =
-                metaDataProjectionInfo == null ? null : metaDataProjectionInfo.getProjectionInfo();
-        Map<String, FunctionCallInformation> metaFunctionCallInfo =
-                metaProjectionInfo == null ? null : metaDataProjectionInfo.getFunctionCallInfoMap();
-
-        NormalizedColumnFilterBuilder normalizedColumnFilterBuilder =
-                new NormalizedColumnFilterBuilder(dataProjectionInfo);
+        NormalizedColumnFilterBuilder normalizedColumnFilterBuilder = new NormalizedColumnFilterBuilder(columnInfo);
         IColumnNormalizedFilterEvaluatorFactory normalizedFilterEvaluatorFactory =
                 normalizedColumnFilterBuilder.build();
 
-        ColumnFilterBuilder columnFilterBuilder = new ColumnFilterBuilder(dataProjectionInfo, context);
+        ColumnFilterBuilder columnFilterBuilder = new ColumnFilterBuilder(columnInfo, context);
         IColumnIterableFilterEvaluatorFactory columnFilterEvaluatorFactory = columnFilterBuilder.build();
 
-        return new QueryColumnTupleProjectorFactory(datasetType, metaItemType, numberOfPrimaryKeys,
-                datasetRequestedType, datasetFunctionCallInfo, metaRequestedType, metaFunctionCallInfo,
-                normalizedFilterEvaluatorFactory, columnFilterEvaluatorFactory);
+        return new QueryColumnTupleProjectorFactory(datasetType, metaItemType, numberOfPrimaryKeys, recordRequestedType,
+                callInfo, metaRequestedType, callInfo, normalizedFilterEvaluatorFactory, columnFilterEvaluatorFactory);
     }
 
     public static ITupleProjectorFactory createUpsertTupleProjectorFactory(DatasetFormatInfo datasetFormatInfo,
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/ColumnFilterBuilder.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/ColumnFilterBuilder.java
index 5c770fd..2b8847b 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/ColumnFilterBuilder.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/ColumnFilterBuilder.java
@@ -35,7 +35,7 @@
 import org.apache.asterix.om.functions.IFunctionManager;
 import org.apache.asterix.om.functions.IFunctionTypeInferer;
 import org.apache.asterix.om.types.ARecordType;
-import org.apache.asterix.runtime.projection.DataProjectionFiltrationInfo;
+import org.apache.asterix.runtime.projection.ColumnDatasetProjectionFiltrationInfo;
 import org.apache.commons.lang3.mutable.Mutable;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
@@ -54,9 +54,9 @@
     private final JobGenContext context;
     private final ArrayPathCheckerVisitor checkerVisitor;
 
-    public ColumnFilterBuilder(DataProjectionFiltrationInfo projectionFiltrationInfo, JobGenContext context) {
+    public ColumnFilterBuilder(ColumnDatasetProjectionFiltrationInfo projectionFiltrationInfo, JobGenContext context) {
         typeEnv = new FilterVariableTypeEnvironment();
-        this.filterPaths = projectionFiltrationInfo.getActualPaths();
+        this.filterPaths = projectionFiltrationInfo.getFilterPaths();
         this.filterExpression = projectionFiltrationInfo.getFilterExpression();
         this.context = context;
         checkerVisitor = new ArrayPathCheckerVisitor();
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/NormalizedColumnFilterBuilder.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/NormalizedColumnFilterBuilder.java
index d667e68..0994664 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/NormalizedColumnFilterBuilder.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/NormalizedColumnFilterBuilder.java
@@ -40,7 +40,7 @@
 import org.apache.asterix.om.constants.AsterixConstantValue;
 import org.apache.asterix.om.functions.BuiltinFunctions;
 import org.apache.asterix.om.types.ARecordType;
-import org.apache.asterix.runtime.projection.DataProjectionFiltrationInfo;
+import org.apache.asterix.runtime.projection.ColumnDatasetProjectionFiltrationInfo;
 import org.apache.commons.lang3.mutable.Mutable;
 import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalExpressionTag;
@@ -56,8 +56,8 @@
     private final Map<ILogicalExpression, ARecordType> filterPaths;
     private final ILogicalExpression filterExpression;
 
-    public NormalizedColumnFilterBuilder(DataProjectionFiltrationInfo projectionFiltrationInfo) {
-        this.filterPaths = projectionFiltrationInfo.getNormalizedPaths();
+    public NormalizedColumnFilterBuilder(ColumnDatasetProjectionFiltrationInfo projectionFiltrationInfo) {
+        this.filterPaths = projectionFiltrationInfo.getFilterPaths();
         this.filterExpression = projectionFiltrationInfo.getFilterExpression();
     }
 
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/ColumnDatasetProjectionFiltrationInfo.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/ColumnDatasetProjectionFiltrationInfo.java
new file mode 100644
index 0000000..0a5ef7a
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/ColumnDatasetProjectionFiltrationInfo.java
@@ -0,0 +1,156 @@
+/*
+ * 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.runtime.projection;
+
+import static org.apache.asterix.om.utils.ProjectionFiltrationTypeUtil.ALL_FIELDS_TYPE;
+import static org.apache.asterix.om.utils.ProjectionFiltrationTypeUtil.EMPTY_TYPE;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.AlgebricksStringBuilderWriter;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+
+public class ColumnDatasetProjectionFiltrationInfo extends ExternalDatasetProjectionInfo {
+    private final ARecordType metaProjectedType;
+    private final ILogicalExpression filterExpression;
+    private final Map<ILogicalExpression, ARecordType> filterPaths;
+    private final ILogicalExpression rangeFilterExpression;
+
+    public ColumnDatasetProjectionFiltrationInfo(ARecordType recordRequestedType, ARecordType metaProjectedType,
+            Map<String, FunctionCallInformation> sourceInformationMap, Map<ILogicalExpression, ARecordType> filterPaths,
+            ILogicalExpression filterExpression, ILogicalExpression rangeFilterExpression) {
+        super(recordRequestedType, sourceInformationMap);
+        this.metaProjectedType = metaProjectedType;
+
+        this.filterExpression = filterExpression;
+        this.rangeFilterExpression = rangeFilterExpression;
+        this.filterPaths = filterPaths;
+    }
+
+    private ColumnDatasetProjectionFiltrationInfo(ColumnDatasetProjectionFiltrationInfo other) {
+        super(other.projectedType, other.functionCallInfoMap);
+        metaProjectedType = other.metaProjectedType;
+
+        filterExpression = other.filterExpression;
+        filterPaths = new HashMap<>(other.filterPaths);
+
+        rangeFilterExpression = other.rangeFilterExpression;
+    }
+
+    @Override
+    public ColumnDatasetProjectionFiltrationInfo createCopy() {
+        return new ColumnDatasetProjectionFiltrationInfo(this);
+    }
+
+    @Override
+    public void print(AlgebricksStringBuilderWriter writer) {
+        StringBuilder builder = new StringBuilder();
+        if (projectedType != ALL_FIELDS_TYPE) {
+            writer.append(" project (");
+            if (projectedType == EMPTY_TYPE) {
+                writer.append(projectedType.getTypeName());
+            } else {
+                writer.append(getOnelinerSchema(projectedType, builder));
+            }
+            writer.append(')');
+        }
+
+        if (metaProjectedType != null && metaProjectedType != ALL_FIELDS_TYPE) {
+            writer.append(" project-meta (");
+            writer.append(getOnelinerSchema(metaProjectedType, builder));
+            writer.append(')');
+        }
+
+        if (filterExpression != null) {
+            writer.append(" filter on: ");
+            writer.append(filterExpression.toString());
+        }
+
+        if (rangeFilterExpression != null) {
+            writer.append(" range-filter on: ");
+            writer.append(rangeFilterExpression.toString());
+        }
+    }
+
+    @Override
+    public void print(JsonGenerator generator) throws IOException {
+        if (projectedType == ALL_FIELDS_TYPE) {
+            return;
+        }
+
+        StringBuilder builder = new StringBuilder();
+        if (projectedType == EMPTY_TYPE) {
+            generator.writeStringField("project", projectedType.getTypeName());
+        } else {
+            generator.writeStringField("project", getOnelinerSchema(projectedType, builder));
+        }
+
+        if (metaProjectedType != null && metaProjectedType != ALL_FIELDS_TYPE) {
+            generator.writeStringField("project-meta", getOnelinerSchema(projectedType, builder));
+        }
+
+        if (filterExpression != null) {
+            generator.writeStringField("filter-on", filterExpression.toString());
+        }
+
+        if (rangeFilterExpression != null) {
+            generator.writeStringField("range-filter-on", rangeFilterExpression.toString());
+        }
+    }
+
+    public ARecordType getMetaProjectedType() {
+        return metaProjectedType;
+    }
+
+    public ILogicalExpression getFilterExpression() {
+        return filterExpression;
+    }
+
+    public Map<ILogicalExpression, ARecordType> getFilterPaths() {
+        return filterPaths;
+    }
+
+    public ILogicalExpression getRangeFilterExpression() {
+        return rangeFilterExpression;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ColumnDatasetProjectionFiltrationInfo otherInfo = (ColumnDatasetProjectionFiltrationInfo) o;
+        return projectedType.deepEqual(otherInfo.projectedType)
+                && Objects.equals(metaProjectedType, otherInfo.metaProjectedType)
+                && Objects.equals(functionCallInfoMap, otherInfo.functionCallInfoMap)
+                && Objects.equals(filterExpression, otherInfo.filterExpression)
+                && Objects.equals(filterPaths, otherInfo.filterPaths)
+                && Objects.equals(rangeFilterExpression, otherInfo.rangeFilterExpression);
+    }
+
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/DataProjectionFiltrationInfo.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/ExternalDatasetProjectionInfo.java
similarity index 60%
rename from asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/DataProjectionFiltrationInfo.java
rename to asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/ExternalDatasetProjectionInfo.java
index 4444fff..55919b3 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/DataProjectionFiltrationInfo.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/ExternalDatasetProjectionInfo.java
@@ -29,70 +29,47 @@
 import java.util.Objects;
 
 import org.apache.asterix.om.types.ARecordType;
-import org.apache.asterix.om.types.IAType;
 import org.apache.asterix.om.types.visitor.SimpleStringBuilderForIATypeVisitor;
 import org.apache.commons.lang3.SerializationUtils;
-import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
 import org.apache.hyracks.algebricks.core.algebra.metadata.IProjectionFiltrationInfo;
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.AlgebricksStringBuilderWriter;
 
-public class DataProjectionFiltrationInfo implements IProjectionFiltrationInfo<ARecordType> {
-    private final ARecordType root;
-    private final Map<String, FunctionCallInformation> functionCallInfoMap;
-    private final Map<ILogicalExpression, ARecordType> normalizedPaths;
-    private final Map<ILogicalExpression, ARecordType> actualPaths;
-    private final ILogicalExpression filterExpression;
+import com.fasterxml.jackson.core.JsonGenerator;
 
-    public DataProjectionFiltrationInfo(ARecordType root, Map<String, FunctionCallInformation> sourceInformationMap,
-            Map<ILogicalExpression, ARecordType> normalizedPaths, Map<ILogicalExpression, ARecordType> actualPaths,
-            ILogicalExpression filterExpression) {
-        this.root = root;
+public class ExternalDatasetProjectionInfo implements IProjectionFiltrationInfo {
+    protected final ARecordType projectedType;
+    protected final Map<String, FunctionCallInformation> functionCallInfoMap;
+
+    public ExternalDatasetProjectionInfo(ARecordType projectedType,
+            Map<String, FunctionCallInformation> sourceInformationMap) {
+        this.projectedType = projectedType;
         this.functionCallInfoMap = sourceInformationMap;
-        this.normalizedPaths = normalizedPaths;
-        this.actualPaths = actualPaths;
-        this.filterExpression = filterExpression;
     }
 
-    private DataProjectionFiltrationInfo(DataProjectionFiltrationInfo other) {
-        if (other.root == ALL_FIELDS_TYPE) {
-            root = ALL_FIELDS_TYPE;
-        } else if (other.root == EMPTY_TYPE) {
-            root = EMPTY_TYPE;
+    private ExternalDatasetProjectionInfo(ExternalDatasetProjectionInfo other) {
+        if (other.projectedType == ALL_FIELDS_TYPE) {
+            projectedType = ALL_FIELDS_TYPE;
+        } else if (other.projectedType == EMPTY_TYPE) {
+            projectedType = EMPTY_TYPE;
         } else {
-            root = other.root.deepCopy(other.root);
+            projectedType = other.projectedType.deepCopy(other.projectedType);
         }
         functionCallInfoMap = new HashMap<>(other.functionCallInfoMap);
-        normalizedPaths = new HashMap<>(other.normalizedPaths);
-        actualPaths = new HashMap<>(other.actualPaths);
-        filterExpression = other.filterExpression;
     }
 
     @Override
-    public ARecordType getProjectionInfo() {
-        return root;
+    public ExternalDatasetProjectionInfo createCopy() {
+        return new ExternalDatasetProjectionInfo(this);
     }
 
-    @Override
-    public DataProjectionFiltrationInfo createCopy() {
-        return new DataProjectionFiltrationInfo(this);
-    }
-
-    @Override
-    public ILogicalExpression getFilterExpression() {
-        return filterExpression;
+    public ARecordType getProjectedType() {
+        return projectedType;
     }
 
     public Map<String, FunctionCallInformation> getFunctionCallInfoMap() {
         return functionCallInfoMap;
     }
 
-    public Map<ILogicalExpression, ARecordType> getNormalizedPaths() {
-        return normalizedPaths;
-    }
-
-    public Map<ILogicalExpression, ARecordType> getActualPaths() {
-        return actualPaths;
-    }
-
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -101,23 +78,41 @@
         if (o == null || getClass() != o.getClass()) {
             return false;
         }
-        DataProjectionFiltrationInfo otherInfo = (DataProjectionFiltrationInfo) o;
-        return root.deepEqual(otherInfo.root) && Objects.equals(functionCallInfoMap, otherInfo.functionCallInfoMap)
-                && Objects.equals(filterExpression, otherInfo.filterExpression)
-                && Objects.equals(normalizedPaths, otherInfo.normalizedPaths);
+        ExternalDatasetProjectionInfo otherInfo = (ExternalDatasetProjectionInfo) o;
+        return projectedType.deepEqual(otherInfo.projectedType)
+                && Objects.equals(functionCallInfoMap, otherInfo.functionCallInfoMap);
     }
 
     @Override
-    public String toString() {
-        if (root == ALL_FIELDS_TYPE || root == EMPTY_TYPE) {
-            //Return the type name if all fields or empty types
-            return root.getTypeName();
+    public void print(AlgebricksStringBuilderWriter writer) {
+        if (projectedType == ALL_FIELDS_TYPE) {
+            return;
         }
-        //Return a oneliner JSON like representation for the requested fields
-        StringBuilder builder = new StringBuilder();
+
+        writer.append(" project (");
+        if (projectedType == EMPTY_TYPE) {
+            writer.append(projectedType.getTypeName());
+        } else {
+            writer.append(getOnelinerSchema(projectedType, new StringBuilder()));
+        }
+        writer.append(')');
+    }
+
+    @Override
+    public void print(JsonGenerator generator) throws IOException {
+        if (projectedType == ALL_FIELDS_TYPE) {
+            return;
+        }
+        generator.writeStringField("project", getOnelinerSchema(projectedType, new StringBuilder()));
+    }
+
+    protected String getOnelinerSchema(ARecordType type, StringBuilder builder) {
+        //Return oneliner JSON like representation for the requested fields
         SimpleStringBuilderForIATypeVisitor visitor = new SimpleStringBuilderForIATypeVisitor();
-        root.accept(visitor, builder);
-        return builder.toString();
+        type.accept(visitor, builder);
+        String onelinerSchema = builder.toString();
+        builder.setLength(0);
+        return onelinerSchema;
     }
 
     /**
@@ -177,8 +172,4 @@
         }
         return functionCallInfoMap;
     }
-
-    private static ARecordType createType(String typeName) {
-        return new ARecordType(typeName, new String[] {}, new IAType[] {}, true);
-    }
 }
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/base/DefaultProjectionFiltrationInfo.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/base/DefaultProjectionFiltrationInfo.java
new file mode 100644
index 0000000..ff13007
--- /dev/null
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/base/DefaultProjectionFiltrationInfo.java
@@ -0,0 +1,48 @@
+/*
+ * 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.hyracks.algebricks.core.algebra.base;
+
+import java.io.IOException;
+
+import org.apache.hyracks.algebricks.core.algebra.metadata.IProjectionFiltrationInfo;
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.AlgebricksStringBuilderWriter;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+
+public class DefaultProjectionFiltrationInfo implements IProjectionFiltrationInfo {
+    public static final IProjectionFiltrationInfo INSTANCE = new DefaultProjectionFiltrationInfo();
+
+    private DefaultProjectionFiltrationInfo() {
+    }
+
+    @Override
+    public IProjectionFiltrationInfo createCopy() {
+        return INSTANCE;
+    }
+
+    @Override
+    public void print(AlgebricksStringBuilderWriter writer) {
+        // NoOp
+    }
+
+    @Override
+    public void print(JsonGenerator generator) throws IOException {
+        // NoOp
+    }
+}
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IMetadataProvider.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IMetadataProvider.java
index e34ff21..4f3d8e4 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IMetadataProvider.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IMetadataProvider.java
@@ -54,8 +54,7 @@
             List<LogicalVariable> minFilterVars, List<LogicalVariable> maxFilterVars,
             ITupleFilterFactory tupleFilterFactory, long outputLimit, IOperatorSchema opSchema,
             IVariableTypeEnvironment typeEnv, JobGenContext context, JobSpecification jobSpec, Object implConfig,
-            IProjectionFiltrationInfo<?> projectionInfo, IProjectionFiltrationInfo<?> metaProjectionInfo)
-            throws AlgebricksException;
+            IProjectionFiltrationInfo projectionFiltrationInfo) throws AlgebricksException;
 
     Pair<IPushRuntimeFactory, AlgebricksPartitionConstraint> getWriteFileRuntime(IDataSink sink, int[] printColumns,
             IPrinterFactory[] printerFactories, IAWriterFactory writerFactory, RecordDescriptor inputDesc)
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IProjectionFiltrationInfo.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IProjectionFiltrationInfo.java
index 9973d08..149731f 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IProjectionFiltrationInfo.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/metadata/IProjectionFiltrationInfo.java
@@ -18,22 +18,33 @@
  */
 package org.apache.hyracks.algebricks.core.algebra.metadata;
 
-import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import java.io.IOException;
+
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.AlgebricksStringBuilderWriter;
+
+import com.fasterxml.jackson.core.JsonGenerator;
 
 /**
  * Generic interface to include the projection information for
  * {@link org.apache.hyracks.algebricks.core.algebra.operators.logical.DataSourceScanOperator}
  */
-public interface IProjectionFiltrationInfo<T> {
-    /**
-     * @return projected values' information
-     */
-    T getProjectionInfo();
-
-    ILogicalExpression getFilterExpression();
-
+public interface IProjectionFiltrationInfo {
     /**
      * @return a copy of the {@link IProjectionFiltrationInfo}
      */
-    IProjectionFiltrationInfo<T> createCopy();
+    IProjectionFiltrationInfo createCopy();
+
+    /**
+     * Write the information in text query plan
+     *
+     * @param writer plan writer
+     */
+    void print(AlgebricksStringBuilderWriter writer);
+
+    /**
+     * Write the information in json query plan
+     *
+     * @param generator json plan generator
+     */
+    void print(JsonGenerator generator) throws IOException;
 }
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/DataSourceScanOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/DataSourceScanOperator.java
index e3ce82d..c0f71fe 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/DataSourceScanOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/DataSourceScanOperator.java
@@ -24,6 +24,7 @@
 
 import org.apache.commons.lang3.mutable.Mutable;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.DefaultProjectionFiltrationInfo;
 import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
@@ -50,22 +51,20 @@
     // the maximum of number of results output by this operator
     private long outputLimit = -1;
 
-    private IProjectionFiltrationInfo<?> datasetProjectionInfo;
-    private IProjectionFiltrationInfo<?> metaProjectionInfo;
+    private IProjectionFiltrationInfo projectionFiltrationInfo;
 
     public DataSourceScanOperator(List<LogicalVariable> variables, IDataSource<?> dataSource) {
-        this(variables, dataSource, null, -1, null, null);
+        this(variables, dataSource, null, -1, DefaultProjectionFiltrationInfo.INSTANCE);
     }
 
     public DataSourceScanOperator(List<LogicalVariable> variables, IDataSource<?> dataSource,
             Mutable<ILogicalExpression> selectCondition, long outputLimit,
-            IProjectionFiltrationInfo<?> datasetProjectionInfo, IProjectionFiltrationInfo<?> metaProjectionInfo) {
+            IProjectionFiltrationInfo projectionFiltrationInfo) {
         super(variables, dataSource);
         projectVars = new ArrayList<>();
         this.selectCondition = selectCondition;
         this.outputLimit = outputLimit;
-        this.datasetProjectionInfo = datasetProjectionInfo;
-        this.metaProjectionInfo = metaProjectionInfo;
+        setProjectionFiltrationInfo(projectionFiltrationInfo);
     }
 
     @Override
@@ -176,19 +175,12 @@
         this.outputLimit = outputLimit;
     }
 
-    public void setDatasetProjectionInfo(IProjectionFiltrationInfo<?> datasetProjectionInfo) {
-        this.datasetProjectionInfo = datasetProjectionInfo;
+    public void setProjectionFiltrationInfo(IProjectionFiltrationInfo projectionFiltrationInfo) {
+        this.projectionFiltrationInfo =
+                projectionFiltrationInfo == null ? DefaultProjectionFiltrationInfo.INSTANCE : projectionFiltrationInfo;
     }
 
-    public IProjectionFiltrationInfo<?> getDatasetProjectionInfo() {
-        return datasetProjectionInfo;
-    }
-
-    public void setMetaProjectionInfo(IProjectionFiltrationInfo<?> metaProjectionInfo) {
-        this.metaProjectionInfo = metaProjectionInfo;
-    }
-
-    public IProjectionFiltrationInfo<?> getMetaProjectionInfo() {
-        return metaProjectionInfo;
+    public IProjectionFiltrationInfo getProjectionFiltrationInfo() {
+        return projectionFiltrationInfo;
     }
 }
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/UnnestMapOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/UnnestMapOperator.java
index 6d11931..2f97c88 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/UnnestMapOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/UnnestMapOperator.java
@@ -22,6 +22,7 @@
 
 import org.apache.commons.lang3.mutable.Mutable;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.DefaultProjectionFiltrationInfo;
 import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
@@ -40,21 +41,20 @@
     // the maximum of number of results output by this operator
     private long outputLimit = -1;
 
-    private IProjectionFiltrationInfo<?> datasetProjectionInfo;
-    private IProjectionFiltrationInfo<?> metaProjectionInfo;
+    private IProjectionFiltrationInfo projectionFiltrationInfo;
 
     public UnnestMapOperator(List<LogicalVariable> variables, Mutable<ILogicalExpression> expression,
             List<Object> variableTypes, boolean propagateInput) {
-        this(variables, expression, variableTypes, propagateInput, null, -1, null, null);
+        this(variables, expression, variableTypes, propagateInput, null, -1, DefaultProjectionFiltrationInfo.INSTANCE);
     }
 
     public UnnestMapOperator(List<LogicalVariable> variables, Mutable<ILogicalExpression> expression,
             List<Object> variableTypes, boolean propagateInput, Mutable<ILogicalExpression> selectCondition,
-            long outputLimit, IProjectionFiltrationInfo<?> datasetProjectionInfo,
-            IProjectionFiltrationInfo<?> metaProjectionInfo) {
+            long outputLimit, IProjectionFiltrationInfo projectionFiltrationInfo) {
         super(variables, expression, variableTypes, propagateInput);
         this.selectCondition = selectCondition;
         this.outputLimit = outputLimit;
+        setProjectionFiltrationInfo(projectionFiltrationInfo);
     }
 
     @Override
@@ -106,20 +106,13 @@
         this.outputLimit = outputLimit;
     }
 
-    public void setDatasetProjectionInfo(IProjectionFiltrationInfo<?> projectionInfo) {
-        this.datasetProjectionInfo = projectionInfo;
+    public void setProjectionFiltrationInfo(IProjectionFiltrationInfo projectionFiltrationInfo) {
+        this.projectionFiltrationInfo =
+                projectionFiltrationInfo == null ? DefaultProjectionFiltrationInfo.INSTANCE : projectionFiltrationInfo;
     }
 
-    public IProjectionFiltrationInfo<?> getDatasetProjectionInfo() {
-        return datasetProjectionInfo;
-    }
-
-    public void setMetaProjectionInfo(IProjectionFiltrationInfo<?> metaProjectionInfo) {
-        this.metaProjectionInfo = metaProjectionInfo;
-    }
-
-    public IProjectionFiltrationInfo<?> getMetaProjectionInfo() {
-        return metaProjectionInfo;
+    public IProjectionFiltrationInfo getProjectionFiltrationInfo() {
+        return projectionFiltrationInfo;
     }
 
 }
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/IsomorphismOperatorVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/IsomorphismOperatorVisitor.java
index 592728e..137dd2e 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/IsomorphismOperatorVisitor.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/IsomorphismOperatorVisitor.java
@@ -452,8 +452,7 @@
             return Boolean.FALSE;
         }
         isomorphic = op.getExpressionRef().getValue().equals(unnestOpArg.getExpressionRef().getValue())
-                && Objects.equals(op.getDatasetProjectionInfo(), unnestOpArg.getDatasetProjectionInfo())
-                && Objects.equals(op.getMetaProjectionInfo(), unnestOpArg.getMetaProjectionInfo());
+                && Objects.equals(op.getProjectionFiltrationInfo(), unnestOpArg.getProjectionFiltrationInfo());
         return isomorphic;
     }
 
@@ -486,8 +485,7 @@
         DataSourceScanOperator argScan = (DataSourceScanOperator) arg;
         boolean isomorphic = op.getDataSource().getId().equals(argScan.getDataSource().getId())
                 && op.getOutputLimit() == argScan.getOutputLimit()
-                && Objects.equals(op.getDatasetProjectionInfo(), argScan.getDatasetProjectionInfo())
-                && Objects.equals(op.getMetaProjectionInfo(), argScan.getMetaProjectionInfo());
+                && Objects.equals(op.getProjectionFiltrationInfo(), argScan.getProjectionFiltrationInfo());
 
         if (!isomorphic) {
             return Boolean.FALSE;
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java
index b49fbcc..6377d44 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java
@@ -318,12 +318,9 @@
             throws AlgebricksException {
         Mutable<ILogicalExpression> newSelectCondition = op.getSelectCondition() != null
                 ? exprDeepCopyVisitor.deepCopyExpressionReference(op.getSelectCondition()) : null;
-        IProjectionFiltrationInfo<?> datasetProjectionInfo =
-                op.getDatasetProjectionInfo() != null ? op.getDatasetProjectionInfo().createCopy() : null;
-        IProjectionFiltrationInfo<?> metaProjectionInfo =
-                op.getMetaProjectionInfo() != null ? op.getMetaProjectionInfo().createCopy() : null;
+        IProjectionFiltrationInfo projectionFiltrationInfo = op.getProjectionFiltrationInfo().createCopy();
         DataSourceScanOperator opCopy = new DataSourceScanOperator(deepCopyVariableList(op.getVariables()),
-                op.getDataSource(), newSelectCondition, op.getOutputLimit(), datasetProjectionInfo, metaProjectionInfo);
+                op.getDataSource(), newSelectCondition, op.getOutputLimit(), projectionFiltrationInfo);
         deepCopyInputsAnnotationsAndExecutionMode(op, arg, opCopy);
         return opCopy;
     }
@@ -539,14 +536,11 @@
             throws AlgebricksException {
         Mutable<ILogicalExpression> newSelectCondition = op.getSelectCondition() != null
                 ? exprDeepCopyVisitor.deepCopyExpressionReference(op.getSelectCondition()) : null;
-        IProjectionFiltrationInfo<?> datasetProjectionInfo =
-                op.getDatasetProjectionInfo() != null ? op.getDatasetProjectionInfo().createCopy() : null;
-        IProjectionFiltrationInfo<?> metaProjectionInfo =
-                op.getMetaProjectionInfo() != null ? op.getMetaProjectionInfo().createCopy() : null;
+        IProjectionFiltrationInfo projectionFiltrationInfo =
+                op.getProjectionFiltrationInfo() != null ? op.getProjectionFiltrationInfo().createCopy() : null;
         UnnestMapOperator opCopy = new UnnestMapOperator(deepCopyVariableList(op.getVariables()),
                 exprDeepCopyVisitor.deepCopyExpressionReference(op.getExpressionRef()), op.getVariableTypes(),
-                op.propagatesInput(), newSelectCondition, op.getOutputLimit(), datasetProjectionInfo,
-                metaProjectionInfo);
+                op.propagatesInput(), newSelectCondition, op.getOutputLimit(), projectionFiltrationInfo);
         deepCopyInputsAnnotationsAndExecutionMode(op, arg, opCopy);
         return opCopy;
     }
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java
index e29b843..960e399 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java
@@ -249,13 +249,10 @@
         newInputList.addAll(op.getVariables());
         Mutable<ILogicalExpression> newSelectCondition =
                 op.getSelectCondition() != null ? deepCopyExpressionRef(op.getSelectCondition()) : null;
-        IProjectionFiltrationInfo<?> datasetProjectionInfo =
-                op.getDatasetProjectionInfo() != null ? op.getDatasetProjectionInfo().createCopy() : null;
-        IProjectionFiltrationInfo<?> metaProjectionInfo =
-                op.getMetaProjectionInfo() != null ? op.getMetaProjectionInfo().createCopy() : null;
+        IProjectionFiltrationInfo projectionFiltrationInfo = op.getProjectionFiltrationInfo().createCopy();
         return new UnnestMapOperator(newInputList, deepCopyExpressionRef(op.getExpressionRef()),
                 new ArrayList<>(op.getVariableTypes()), op.propagatesInput(), newSelectCondition, op.getOutputLimit(),
-                datasetProjectionInfo, metaProjectionInfo);
+                projectionFiltrationInfo);
     }
 
     @Override
@@ -273,13 +270,9 @@
         newInputList.addAll(op.getVariables());
         Mutable<ILogicalExpression> newSelectCondition =
                 op.getSelectCondition() != null ? deepCopyExpressionRef(op.getSelectCondition()) : null;
-        IProjectionFiltrationInfo<?> datasetProjectionInfo =
-                op.getDatasetProjectionInfo() != null ? op.getDatasetProjectionInfo().createCopy() : null;
-        IProjectionFiltrationInfo<?> metaProjectionInfo =
-                op.getMetaProjectionInfo() != null ? op.getMetaProjectionInfo().createCopy() : null;
-
+        IProjectionFiltrationInfo projectionFiltrationInfo = op.getProjectionFiltrationInfo().createCopy();
         return new DataSourceScanOperator(newInputList, op.getDataSource(), newSelectCondition, op.getOutputLimit(),
-                datasetProjectionInfo, metaProjectionInfo);
+                projectionFiltrationInfo);
     }
 
     @Override
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DataSourceScanPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DataSourceScanPOperator.java
index 8a4b3f0..03cb83b 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DataSourceScanPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DataSourceScanPOperator.java
@@ -117,10 +117,10 @@
                     scan.getSelectCondition().getValue(), context);
         }
 
-        Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> p = mp.getScannerRuntime(dataSource, vars, projectVars,
-                scan.isProjectPushed(), scan.getMinFilterVars(), scan.getMaxFilterVars(), tupleFilterFactory,
-                scan.getOutputLimit(), opSchema, typeEnv, context, builder.getJobSpec(), implConfig,
-                scan.getDatasetProjectionInfo(), scan.getMetaProjectionInfo());
+        Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> p =
+                mp.getScannerRuntime(dataSource, vars, projectVars, scan.isProjectPushed(), scan.getMinFilterVars(),
+                        scan.getMaxFilterVars(), tupleFilterFactory, scan.getOutputLimit(), opSchema, typeEnv, context,
+                        builder.getJobSpec(), implConfig, scan.getProjectionFiltrationInfo());
         IOperatorDescriptor opDesc = p.first;
         opDesc.setSourceLocation(scan.getSourceLocation());
         builder.contributeHyracksOperator(scan, opDesc);
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitor.java
index d6df7c6..e7fc5f9 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitor.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitor.java
@@ -33,7 +33,6 @@
 import org.apache.hyracks.algebricks.core.algebra.base.IPhysicalOperator;
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
 import org.apache.hyracks.algebricks.core.algebra.expressions.IAlgebricksConstantValue;
-import org.apache.hyracks.algebricks.core.algebra.metadata.IProjectionFiltrationInfo;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractOperatorWithNestedPlans;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractUnnestMapOperator;
@@ -350,9 +349,7 @@
         AlgebricksStringBuilderWriter plan = printAbstractUnnestMapOperator(op, indent, "unnest-map", null);
         appendSelectConditionInformation(plan, op.getSelectCondition(), indent);
         appendLimitInformation(plan, op.getOutputLimit());
-        appendProjectInformation(plan, "project", op.getDatasetProjectionInfo());
-        appendProjectInformation(plan, "project-meta", op.getMetaProjectionInfo());
-        appendFilterExpression(plan, op.getDatasetProjectionInfo());
+        op.getProjectionFiltrationInfo().print(plan);
         return null;
     }
 
@@ -375,14 +372,13 @@
 
     @Override
     public Void visitDataScanOperator(DataSourceScanOperator op, Integer indent) throws AlgebricksException {
-        AlgebricksStringBuilderWriter plan = addIndent(indent).append(
-                "data-scan " + op.getProjectVariables() + "<-" + op.getVariables() + " <- " + op.getDataSource());
+        AlgebricksStringBuilderWriter plan = addIndent(indent).append("data-scan ")
+                .append(String.valueOf(op.getProjectVariables())).append("<-").append(String.valueOf(op.getVariables()))
+                .append(" <- ").append(String.valueOf(op.getDataSource()));
         appendFilterInformation(plan, op.getMinFilterVars(), op.getMaxFilterVars());
         appendSelectConditionInformation(plan, op.getSelectCondition(), indent);
         appendLimitInformation(plan, op.getOutputLimit());
-        appendProjectInformation(plan, "project", op.getDatasetProjectionInfo());
-        appendProjectInformation(plan, "project-meta", op.getMetaProjectionInfo());
-        appendFilterExpression(plan, op.getDatasetProjectionInfo());
+        op.getProjectionFiltrationInfo().print(plan);
         return null;
     }
 
@@ -413,30 +409,6 @@
         }
     }
 
-    private void appendProjectInformation(AlgebricksStringBuilderWriter plan, String projectionSource,
-            IProjectionFiltrationInfo<?> projectionInfo) {
-        final String projectedFields = projectionInfo == null ? "" : projectionInfo.toString();
-        if (!projectedFields.isEmpty()) {
-            plan.append(" ");
-            plan.append(projectionSource);
-            plan.append(" (");
-            plan.append(projectedFields);
-            plan.append(")");
-        }
-    }
-
-    private void appendFilterExpression(AlgebricksStringBuilderWriter plan,
-            IProjectionFiltrationInfo<?> projectionInfo) {
-        final String filterExpr = projectionInfo == null || projectionInfo.getFilterExpression() == null ? ""
-                : projectionInfo.getFilterExpression().toString();
-        if (!filterExpr.isEmpty()) {
-            plan.append(" filter on ");
-            plan.append("(");
-            plan.append(filterExpr);
-            plan.append(")");
-        }
-    }
-
     @Override
     public Void visitLimitOperator(LimitOperator op, Integer indent) throws AlgebricksException {
         addIndent(indent).append("limit");
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitorJson.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitorJson.java
index b58ca75..4360c67 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitorJson.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitorJson.java
@@ -38,7 +38,6 @@
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
 import org.apache.hyracks.algebricks.core.algebra.base.PhysicalOperatorTag;
 import org.apache.hyracks.algebricks.core.algebra.expressions.IAlgebricksConstantValue;
-import org.apache.hyracks.algebricks.core.algebra.metadata.IProjectionFiltrationInfo;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractBinaryJoinOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractOperatorWithNestedPlans;
@@ -531,9 +530,7 @@
         try {
             writeUnnestMapOperator(op, indent, "unnest-map", null);
             writeSelectLimitInformation(op.getSelectCondition(), op.getOutputLimit(), indent);
-            writeProjectInformation("project", op.getDatasetProjectionInfo());
-            writeProjectInformation("project-meta", op.getMetaProjectionInfo());
-            writeFilterInformation(op.getDatasetProjectionInfo());
+            op.getProjectionFiltrationInfo().print(jsonGenerator);
             return null;
         } catch (IOException e) {
             throw AlgebricksException.create(ErrorCode.ERROR_PRINTING_PLAN, e, String.valueOf(e));
@@ -563,9 +560,7 @@
             }
             writeFilterInformation(op.getMinFilterVars(), op.getMaxFilterVars());
             writeSelectLimitInformation(op.getSelectCondition(), op.getOutputLimit(), indent);
-            writeProjectInformation("project", op.getDatasetProjectionInfo());
-            writeProjectInformation("project-meta", op.getMetaProjectionInfo());
-            writeFilterInformation(op.getDatasetProjectionInfo());
+            op.getProjectionFiltrationInfo().print(jsonGenerator);
             return null;
         } catch (IOException e) {
             throw AlgebricksException.create(ErrorCode.ERROR_PRINTING_PLAN, e, String.valueOf(e));
@@ -894,22 +889,6 @@
         }
     }
 
-    private void writeProjectInformation(String projectionSource, IProjectionFiltrationInfo<?> projectionInfo)
-            throws IOException {
-        final String projectedFields = projectionInfo == null ? "" : projectionInfo.toString();
-        if (!projectedFields.isEmpty()) {
-            jsonGenerator.writeStringField(projectionSource, projectedFields);
-        }
-    }
-
-    private void writeFilterInformation(IProjectionFiltrationInfo<?> projectionInfo) throws IOException {
-        final String filterExpr = projectionInfo == null || projectionInfo.getFilterExpression() == null ? ""
-                : projectionInfo.getFilterExpression().toString();
-        if (!filterExpr.isEmpty()) {
-            jsonGenerator.writeStringField("filter-on", filterExpr);
-        }
-    }
-
     private void writeVariablesAndExpressions(List<LogicalVariable> variables,
             List<Mutable<ILogicalExpression>> expressions, Void indent) throws IOException, AlgebricksException {
         if (!variables.isEmpty()) {