diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/AsterixOptimizationContext.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/AsterixOptimizationContext.java
index ea3404d..20673eb 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/AsterixOptimizationContext.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/AsterixOptimizationContext.java
@@ -30,7 +30,7 @@
 import org.apache.hyracks.algebricks.core.algebra.expressions.IExpressionTypeComputer;
 import org.apache.hyracks.algebricks.core.algebra.expressions.IMergeAggregationExpressionFactory;
 import org.apache.hyracks.algebricks.core.algebra.expressions.IMissableTypeComputer;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.LogicalOperatorPrettyPrintVisitor;
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.IPlanPrettyPrinter;
 import org.apache.hyracks.algebricks.core.rewriter.base.AlgebricksOptimizationContext;
 import org.apache.hyracks.algebricks.core.rewriter.base.PhysicalOptimizationConfig;
 import org.apache.hyracks.api.exceptions.IWarningCollector;
@@ -46,11 +46,11 @@
             IMergeAggregationExpressionFactory mergeAggregationExpressionFactory,
             IExpressionTypeComputer expressionTypeComputer, IMissableTypeComputer nullableTypeComputer,
             IConflictingTypeResolver conflictingTypeResovler, PhysicalOptimizationConfig physicalOptimizationConfig,
-            AlgebricksPartitionConstraint clusterLocations, LogicalOperatorPrettyPrintVisitor prettyPrintVisitor,
+            AlgebricksPartitionConstraint clusterLocations, IPlanPrettyPrinter prettyPrinter,
             IWarningCollector warningCollector) {
         super(varCounter, expressionEvalSizeComputer, mergeAggregationExpressionFactory, expressionTypeComputer,
                 nullableTypeComputer, conflictingTypeResovler, physicalOptimizationConfig, clusterLocations,
-                prettyPrintVisitor, warningCollector);
+                prettyPrinter, warningCollector);
     }
 
     public void addDataSource(DataSource dataSource) {
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/APIFramework.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/APIFramework.java
index 5069b9f..d573b52 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/APIFramework.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/APIFramework.java
@@ -95,10 +95,8 @@
 import org.apache.hyracks.algebricks.core.algebra.expressions.IExpressionTypeComputer;
 import org.apache.hyracks.algebricks.core.algebra.expressions.IMergeAggregationExpressionFactory;
 import org.apache.hyracks.algebricks.core.algebra.expressions.IMissableTypeComputer;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.AbstractLogicalOperatorPrettyPrintVisitor;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.AlgebricksAppendable;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.LogicalOperatorPrettyPrintVisitor;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.LogicalOperatorPrettyPrintVisitorJson;
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.AlgebricksStringBuilderWriter;
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.IPlanPrettyPrinter;
 import org.apache.hyracks.algebricks.core.algebra.prettyprint.PlanPrettyPrinter;
 import org.apache.hyracks.algebricks.core.rewriter.base.IOptimizationContextFactory;
 import org.apache.hyracks.algebricks.core.rewriter.base.PhysicalOptimizationConfig;
@@ -171,10 +169,10 @@
                 IExpressionTypeComputer expressionTypeComputer, IMissableTypeComputer missableTypeComputer,
                 IConflictingTypeResolver conflictingTypeResolver, PhysicalOptimizationConfig physicalOptimizationConfig,
                 AlgebricksPartitionConstraint clusterLocations, IWarningCollector warningCollector) {
-            LogicalOperatorPrettyPrintVisitor prettyPrintVisitor = new LogicalOperatorPrettyPrintVisitor();
+            IPlanPrettyPrinter prettyPrinter = PlanPrettyPrinter.createStringPlanPrettyPrinter();
             return new AsterixOptimizationContext(varCounter, expressionEvalSizeComputer,
                     mergeAggregationExpressionFactory, expressionTypeComputer, missableTypeComputer,
-                    conflictingTypeResolver, physicalOptimizationConfig, clusterLocations, prettyPrintVisitor,
+                    conflictingTypeResolver, physicalOptimizationConfig, clusterLocations, prettyPrinter,
                     warningCollector);
         }
     }
@@ -258,8 +256,9 @@
             if (conf.is(SessionConfig.OOB_OPTIMIZED_LOGICAL_PLAN) || isExplainOnly) {
                 if (conf.is(SessionConfig.FORMAT_ONLY_PHYSICAL_OPS)) {
                     // For Optimizer tests.
-                    AlgebricksAppendable buffer = new AlgebricksAppendable(output.out());
-                    PlanPrettyPrinter.printPhysicalOps(plan, buffer, 0);
+                    AlgebricksStringBuilderWriter buf = new AlgebricksStringBuilderWriter(PlanPrettyPrinter.INIT_SIZE);
+                    PlanPrettyPrinter.printPhysicalOps(plan, buf, 0);
+                    output.out().write(buf.toString());
                 } else {
                     if (isQuery || isLoad) {
                         generateOptimizedLogicalPlan(plan, output.config().getPlanFormat());
@@ -341,10 +340,9 @@
         }
     }
 
-    private AbstractLogicalOperatorPrettyPrintVisitor getPrettyPrintVisitor(SessionConfig.PlanFormat planFormat,
-            PrintWriter out) {
-        return planFormat.equals(SessionConfig.PlanFormat.JSON) ? new LogicalOperatorPrettyPrintVisitorJson(out)
-                : new LogicalOperatorPrettyPrintVisitor(out);
+    private IPlanPrettyPrinter getPrettyPrintVisitor(SessionConfig.PlanFormat planFormat) {
+        return planFormat.equals(SessionConfig.PlanFormat.JSON) ? PlanPrettyPrinter.createJsonPlanPrettyPrinter()
+                : PlanPrettyPrinter.createStringPlanPrettyPrinter();
     }
 
     public void executeJobArray(IHyracksClientConnection hcc, JobSpecification[] specs, PrintWriter out)
@@ -485,20 +483,12 @@
     }
 
     private void generateLogicalPlan(ILogicalPlan plan, SessionConfig.PlanFormat format) throws AlgebricksException {
-        final StringWriter stringWriter = new StringWriter();
-        try (PrintWriter writer = new PrintWriter(stringWriter)) {
-            PlanPrettyPrinter.printPlan(plan, getPrettyPrintVisitor(format, writer), 0);
-            executionPlans.setLogicalPlan(stringWriter.toString());
-        }
+        executionPlans.setLogicalPlan(getPrettyPrintVisitor(format).printPlan(plan).toString());
     }
 
     private void generateOptimizedLogicalPlan(ILogicalPlan plan, SessionConfig.PlanFormat format)
             throws AlgebricksException {
-        final StringWriter stringWriter = new StringWriter();
-        try (PrintWriter writer = new PrintWriter(stringWriter)) {
-            PlanPrettyPrinter.printPlan(plan, getPrettyPrintVisitor(format, writer), 0);
-            executionPlans.setOptimizedLogicalPlan(stringWriter.toString());
-        }
+        executionPlans.setOptimizedLogicalPlan(getPrettyPrintVisitor(format).printPlan(plan).toString());
     }
 
     private void generateJob(JobSpecification spec) {
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java
index 1f8e44c..1cdf6f7 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java
@@ -37,6 +37,7 @@
 import org.apache.asterix.lang.common.base.Statement;
 import org.apache.asterix.metadata.MetadataManager;
 import org.apache.asterix.om.base.IAObject;
+import org.apache.asterix.translator.ExecutionPlans;
 import org.apache.asterix.translator.IRequestParameters;
 import org.apache.asterix.translator.IStatementExecutor;
 import org.apache.asterix.translator.IStatementExecutorFactory;
@@ -63,6 +64,7 @@
     private final IStorageComponentProvider storageComponentProvider;
     private ICcApplicationContext appCtx;
     private Map<String, IAObject> statementParams;
+    private ExecutionPlans executionPlans;
 
     public AsterixJavaClient(ICcApplicationContext appCtx, IHyracksClientConnection hcc, Reader queryText,
             PrintWriter writer, ILangCompilationProvider compilationProvider,
@@ -107,6 +109,7 @@
             PlanFormat pformat) throws Exception {
         queryJobSpec = null;
         dmlJobs = null;
+        executionPlans = null;
 
         if (queryText == null) {
             return;
@@ -136,6 +139,7 @@
                 new ResultProperties(IStatementExecutor.ResultDelivery.IMMEDIATE), new IStatementExecutor.Stats(), null,
                 null, null, statementParams, true);
         translator.compileAndExecute(hcc, requestParameters);
+        executionPlans = translator.getExecutionPlans();
         writer.flush();
     }
 
@@ -148,4 +152,7 @@
         }
     }
 
+    public ExecutionPlans getExecutionPlans() {
+        return executionPlans;
+    }
 }
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/jsonplan/JsonLogicalPlanTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/jsonplan/JsonLogicalPlanTest.java
index 38b38c3..8663918 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/jsonplan/JsonLogicalPlanTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/jsonplan/JsonLogicalPlanTest.java
@@ -47,6 +47,7 @@
 import org.apache.asterix.test.base.AsterixTestHelper;
 import org.apache.asterix.test.common.TestHelper;
 import org.apache.asterix.test.runtime.HDFSCluster;
+import org.apache.asterix.translator.ExecutionPlans;
 import org.apache.asterix.translator.IStatementExecutorFactory;
 import org.apache.asterix.translator.SessionConfig.PlanFormat;
 import org.apache.commons.io.FileUtils;
@@ -62,18 +63,24 @@
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
 
-import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 @RunWith(Parameterized.class)
 public class JsonLogicalPlanTest {
 
     private static final Logger LOGGER = LogManager.getLogger();
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+    static {
+        OBJECT_MAPPER.enable(DeserializationFeature.FAIL_ON_READING_DUP_TREE_KEY);
+        OBJECT_MAPPER.enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
+    }
 
     protected static final String SEPARATOR = File.separator;
     private static final String EXTENSION_AQL = "aql";
     private static final String EXTENSION_SQLPP = "sqlpp";
-    private static final String EXTENSION_RESULT = "plan";
+    private static final String EXTENSION_RESULT = "plan.json";
     private static final String FILENAME_IGNORE = "ignore.txt";
     private static final String FILENAME_ONLY = "only.txt";
     private static final String PATH_BASE =
@@ -189,13 +196,16 @@
                 provider = extensionLangCompilationProvider;
             }
             IHyracksClientConnection hcc = integrationUtil.getHyracksClientConnection();
-
+            String planStr;
             try (PrintWriter plan = new PrintWriter(actualFile)) {
                 AsterixJavaClient asterix = new AsterixJavaClient(
                         (ICcApplicationContext) integrationUtil.cc.getApplicationContext(), hcc,
                         new StringReader(query), plan, provider, statementExecutorFactory, storageComponentProvider);
                 asterix.setStatementParameters(queryParams);
                 asterix.compile(true, false, !optimized, optimized, false, false, false, PlanFormat.JSON);
+                ExecutionPlans executionPlans = asterix.getExecutionPlans();
+                planStr = optimized ? executionPlans.getOptimizedLogicalPlan() : executionPlans.getLogicalPlan();
+                plan.write(planStr);
             } catch (AsterixException e) {
                 throw new Exception("Compile ERROR for " + queryFile + ": " + e.getMessage(), e);
             }
@@ -217,8 +227,9 @@
             }
 
             try {
-                final JsonParser parser = new ObjectMapper().getJsonFactory().createJsonParser(objectActual);
-                while (parser.nextToken() != null) {
+                JsonNode jsonNode = OBJECT_MAPPER.readTree(planStr);
+                if (jsonNode == null || !jsonNode.isObject()) {
+                    throw new Exception("ERROR: No JSON plan or plan is malformed!");
                 }
             } finally {
                 readerActual.close();
diff --git a/hyracks-fullstack/algebricks/algebricks-compiler/src/main/java/org/apache/hyracks/algebricks/compiler/api/HeuristicCompilerFactoryBuilder.java b/hyracks-fullstack/algebricks/algebricks-compiler/src/main/java/org/apache/hyracks/algebricks/compiler/api/HeuristicCompilerFactoryBuilder.java
index d51c363..ab68701 100644
--- a/hyracks-fullstack/algebricks/algebricks-compiler/src/main/java/org/apache/hyracks/algebricks/compiler/api/HeuristicCompilerFactoryBuilder.java
+++ b/hyracks-fullstack/algebricks/algebricks-compiler/src/main/java/org/apache/hyracks/algebricks/compiler/api/HeuristicCompilerFactoryBuilder.java
@@ -28,7 +28,8 @@
 import org.apache.hyracks.algebricks.core.algebra.expressions.IMergeAggregationExpressionFactory;
 import org.apache.hyracks.algebricks.core.algebra.expressions.IMissableTypeComputer;
 import org.apache.hyracks.algebricks.core.algebra.metadata.IMetadataProvider;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.LogicalOperatorPrettyPrintVisitor;
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.IPlanPrettyPrinter;
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.PlanPrettyPrinter;
 import org.apache.hyracks.algebricks.core.config.AlgebricksConfig;
 import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenContext;
 import org.apache.hyracks.algebricks.core.jobgen.impl.PlanCompiler;
@@ -56,10 +57,10 @@
                 IExpressionTypeComputer expressionTypeComputer, IMissableTypeComputer missableTypeComputer,
                 IConflictingTypeResolver conflictingTypeResolver, PhysicalOptimizationConfig physicalOptimizationConfig,
                 AlgebricksPartitionConstraint clusterLocations, IWarningCollector warningCollector) {
-            LogicalOperatorPrettyPrintVisitor prettyPrintVisitor = new LogicalOperatorPrettyPrintVisitor();
+            IPlanPrettyPrinter prettyPrinter = PlanPrettyPrinter.createStringPlanPrettyPrinter();
             return new AlgebricksOptimizationContext(varCounter, expressionEvalSizeComputer,
                     mergeAggregationExpressionFactory, expressionTypeComputer, missableTypeComputer,
-                    conflictingTypeResolver, physicalOptimizationConfig, clusterLocations, prettyPrintVisitor,
+                    conflictingTypeResolver, physicalOptimizationConfig, clusterLocations, prettyPrinter,
                     warningCollector);
         }
     }
diff --git a/hyracks-fullstack/algebricks/algebricks-core/pom.xml b/hyracks-fullstack/algebricks/algebricks-core/pom.xml
index 42e30dd..35acfb8 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/pom.xml
+++ b/hyracks-fullstack/algebricks/algebricks-core/pom.xml
@@ -87,6 +87,10 @@
       <artifactId>commons-lang3</artifactId>
     </dependency>
     <dependency>
+      <groupId>commons-io</groupId>
+      <artifactId>commons-io</artifactId>
+    </dependency>
+    <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-api</artifactId>
     </dependency>
@@ -94,5 +98,9 @@
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-collections4</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-core</artifactId>
+    </dependency>
   </dependencies>
 </project>
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/base/IOptimizationContext.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/base/IOptimizationContext.java
index c31cb88..726a5df 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/base/IOptimizationContext.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/base/IOptimizationContext.java
@@ -25,7 +25,7 @@
 import org.apache.hyracks.algebricks.core.algebra.expressions.IMergeAggregationExpressionFactory;
 import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableEvalSizeEnvironment;
 import org.apache.hyracks.algebricks.core.algebra.metadata.IMetadataProvider;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.AbstractLogicalOperatorPrettyPrintVisitor;
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.IPlanPrettyPrinter;
 import org.apache.hyracks.algebricks.core.algebra.properties.FunctionalDependency;
 import org.apache.hyracks.algebricks.core.algebra.properties.ILogicalPropertiesVector;
 import org.apache.hyracks.algebricks.core.algebra.properties.INodeDomain;
@@ -82,7 +82,7 @@
 
     public void updatePrimaryKeys(Map<LogicalVariable, LogicalVariable> mappedVars);
 
-    public AbstractLogicalOperatorPrettyPrintVisitor getPrettyPrintVisitor();
+    public IPlanPrettyPrinter getPrettyPrinter();
 
     public INodeDomain getComputationNodeDomain();
 
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/IntersectOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/IntersectOperator.java
index 36d2db5..f77d9db 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/IntersectOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/IntersectOperator.java
@@ -158,6 +158,14 @@
         return inputExtraVars.get(inputIndex);
     }
 
+    public List<List<LogicalVariable>> getAllInputsCompareVariables() {
+        return inputCompareVars;
+    }
+
+    public List<List<LogicalVariable>> getAllInputsExtraVariables() {
+        return inputExtraVars;
+    }
+
     public List<LogicalVariable> getOutputCompareVariables() {
         return outputCompareVars;
     }
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/plan/ALogicalPlanImpl.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/plan/ALogicalPlanImpl.java
index f8d0a50..b5b01a0 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/plan/ALogicalPlanImpl.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/plan/ALogicalPlanImpl.java
@@ -25,14 +25,13 @@
 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.ILogicalPlan;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.LogicalOperatorPrettyPrintVisitor;
 import org.apache.hyracks.algebricks.core.algebra.prettyprint.PlanPrettyPrinter;
 
 public class ALogicalPlanImpl implements ILogicalPlan {
     private List<Mutable<ILogicalOperator>> roots;
 
     public ALogicalPlanImpl() {
-        this.roots = new ArrayList<Mutable<ILogicalOperator>>();
+        this.roots = new ArrayList<>();
     }
 
     public ALogicalPlanImpl(List<Mutable<ILogicalOperator>> roots) {
@@ -40,7 +39,7 @@
     }
 
     public ALogicalPlanImpl(Mutable<ILogicalOperator> root) {
-        roots = new ArrayList<Mutable<ILogicalOperator>>(1);
+        roots = new ArrayList<>(1);
         roots.add(root);
     }
 
@@ -54,9 +53,7 @@
     }
 
     public static String prettyPrintPlan(ILogicalPlan plan) throws AlgebricksException {
-        LogicalOperatorPrettyPrintVisitor pvisitor = new LogicalOperatorPrettyPrintVisitor();
-        PlanPrettyPrinter.printPlan(plan, pvisitor, 0);
-        return pvisitor.get().toString();
+        return PlanPrettyPrinter.createStringPlanPrettyPrinter().printPlan(plan).toString();
     }
 
     @Override
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/AbstractLogicalOperatorPrettyPrintVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/AbstractLogicalOperatorPrettyPrintVisitor.java
index 5319e6a2..9164e6d 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/AbstractLogicalOperatorPrettyPrintVisitor.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/AbstractLogicalOperatorPrettyPrintVisitor.java
@@ -28,36 +28,25 @@
 import org.apache.hyracks.algebricks.core.algebra.visitors.ILogicalExpressionVisitor;
 import org.apache.hyracks.algebricks.core.algebra.visitors.ILogicalOperatorVisitor;
 
-public abstract class AbstractLogicalOperatorPrettyPrintVisitor implements ILogicalOperatorVisitor<Void, Integer> {
-    ILogicalExpressionVisitor<String, Integer> exprVisitor;
-    AlgebricksAppendable buffer;
+public abstract class AbstractLogicalOperatorPrettyPrintVisitor<T> implements ILogicalOperatorVisitor<Void, T> {
 
-    public AbstractLogicalOperatorPrettyPrintVisitor() {
-        this(new AlgebricksAppendable());
-    }
+    protected final ILogicalExpressionVisitor<String, T> exprVisitor;
+    protected final AlgebricksStringBuilderWriter buffer;
 
-    public AbstractLogicalOperatorPrettyPrintVisitor(Appendable app) {
-        this(new AlgebricksAppendable(app), new LogicalExpressionPrettyPrintVisitor());
-    }
-
-    public AbstractLogicalOperatorPrettyPrintVisitor(AlgebricksAppendable buffer) {
-        this(buffer, new LogicalExpressionPrettyPrintVisitor());
-    }
-
-    public AbstractLogicalOperatorPrettyPrintVisitor(AlgebricksAppendable buffer,
-            ILogicalExpressionVisitor<String, Integer> exprVisitor) {
-        this.buffer = buffer;
+    public AbstractLogicalOperatorPrettyPrintVisitor(ILogicalExpressionVisitor<String, T> exprVisitor) {
+        this.buffer = new AlgebricksStringBuilderWriter(PlanPrettyPrinter.INIT_SIZE);
         this.exprVisitor = exprVisitor;
     }
 
-    public AlgebricksAppendable reset(AlgebricksAppendable buffer) {
-        AlgebricksAppendable old = this.buffer;
-        this.buffer = buffer;
-        return old;
+    public static void printPhysicalOps(ILogicalPlan plan, AlgebricksStringBuilderWriter out, int indent)
+            throws AlgebricksException {
+        for (Mutable<ILogicalOperator> root : plan.getRoots()) {
+            printPhysicalOperator((AbstractLogicalOperator) root.getValue(), indent, out);
+        }
     }
 
-    public AlgebricksAppendable get() {
-        return buffer;
+    protected void resetState() {
+        buffer.getBuilder().setLength(0);
     }
 
     @Override
@@ -65,41 +54,33 @@
         return buffer.toString();
     }
 
-    CharSequence str(Object o) {
+    String str(Object o) {
         return String.valueOf(o);
     }
 
-    protected static void appendln(AlgebricksAppendable buf, String s) throws AlgebricksException {
+    protected static void appendln(AlgebricksStringBuilderWriter buf, String s) {
         buf.append(s);
         buf.append("\n");
     }
 
-    protected static void append(AlgebricksAppendable buf, String s) throws AlgebricksException {
+    protected static void append(AlgebricksStringBuilderWriter buf, String s) {
         buf.append(s);
     }
 
-    protected static void pad(AlgebricksAppendable buf, int indent) throws AlgebricksException {
+    protected static void pad(AlgebricksStringBuilderWriter buf, int indent) {
         for (int i = 0; i < indent; ++i) {
             buf.append(' ');
         }
     }
 
-    protected AlgebricksAppendable addIndent(int level) throws AlgebricksException {
+    protected AlgebricksStringBuilderWriter addIndent(int level) {
         for (int i = 0; i < level; ++i) {
             buffer.append(' ');
         }
         return buffer;
     }
 
-    public void printPlan(ILogicalPlan plan, int indent) throws AlgebricksException {
-        for (Mutable<ILogicalOperator> root : plan.getRoots()) {
-            printOperator((AbstractLogicalOperator) root.getValue(), indent);
-        }
-    }
-
-    public abstract void printOperator(AbstractLogicalOperator op, int indent) throws AlgebricksException;
-
-    public static void printPhysicalOperator(AbstractLogicalOperator op, int indent, AlgebricksAppendable out)
+    private static void printPhysicalOperator(AbstractLogicalOperator op, int indent, AlgebricksStringBuilderWriter out)
             throws AlgebricksException {
         IPhysicalOperator pOp = op.getPhysicalOperator();
         pad(out, indent);
@@ -118,11 +99,4 @@
             printPhysicalOperator((AbstractLogicalOperator) i.getValue(), indent + 2, out);
         }
     }
-
-    public static void printPhysicalOps(ILogicalPlan plan, AlgebricksAppendable out, int indent)
-            throws AlgebricksException {
-        for (Mutable<ILogicalOperator> root : plan.getRoots()) {
-            printPhysicalOperator((AbstractLogicalOperator) root.getValue(), indent, out);
-        }
-    }
 }
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/AlgebricksStringBuilderWriter.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/AlgebricksStringBuilderWriter.java
new file mode 100644
index 0000000..2444ce8
--- /dev/null
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/AlgebricksStringBuilderWriter.java
@@ -0,0 +1,58 @@
+/*
+ * 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.prettyprint;
+
+import java.io.IOException;
+
+import org.apache.commons.io.output.StringBuilderWriter;
+import org.apache.hyracks.util.annotations.NotThreadSafe;
+
+/**
+ * String writer based on string builder to provide un-synchronized writer. It overrides append methods to allow
+ * chaining to the same string-builder-based writer to avoid throwing {@link IOException}
+ */
+@NotThreadSafe
+public class AlgebricksStringBuilderWriter extends StringBuilderWriter {
+
+    public AlgebricksStringBuilderWriter() {
+        super();
+    }
+
+    public AlgebricksStringBuilderWriter(final int capacity) {
+        super(capacity);
+    }
+
+    @Override
+    public AlgebricksStringBuilderWriter append(final char value) {
+        super.append(value);
+        return this;
+    }
+
+    @Override
+    public AlgebricksStringBuilderWriter append(final CharSequence value) {
+        super.append(value);
+        return this;
+    }
+
+    @Override
+    public AlgebricksStringBuilderWriter append(final CharSequence value, final int start, final int end) {
+        super.append(value, start, end);
+        return this;
+    }
+}
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/IPlanPrettyPrinter.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/IPlanPrettyPrinter.java
new file mode 100644
index 0000000..922e0ee
--- /dev/null
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/IPlanPrettyPrinter.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.algebricks.core.algebra.prettyprint;
+
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalPlan;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+
+/**
+ * Note: Some implementations may be stateful and not thread-safe.
+ */
+public interface IPlanPrettyPrinter {
+
+    /** Prints the plan rooted at the operator argument. */
+    IPlanPrettyPrinter printOperator(AbstractLogicalOperator operator) throws AlgebricksException;
+
+    /** Prints the whole logical plan. */
+    IPlanPrettyPrinter printPlan(ILogicalPlan plan) throws AlgebricksException;
+
+    /** Resets the state of the pretty printer. */
+    IPlanPrettyPrinter reset() throws AlgebricksException;
+
+    /** @return the string of the printed plan. */
+    String toString();
+}
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalExpressionPrettyPrintVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalExpressionPrettyPrintVisitor.java
index 72f891a..6794c64 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalExpressionPrettyPrintVisitor.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalExpressionPrettyPrintVisitor.java
@@ -27,39 +27,38 @@
 import org.apache.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
 import org.apache.hyracks.algebricks.core.algebra.visitors.ILogicalExpressionVisitor;
 
-public class LogicalExpressionPrettyPrintVisitor implements ILogicalExpressionVisitor<String, Integer> {
+public class LogicalExpressionPrettyPrintVisitor<T> implements ILogicalExpressionVisitor<String, T> {
 
     @Override
-    public String visitConstantExpression(ConstantExpression expr, Integer indent) throws AlgebricksException {
+    public String visitConstantExpression(ConstantExpression expr, T arg) throws AlgebricksException {
         return expr.toString();
     }
 
     @Override
-    public String visitVariableReferenceExpression(VariableReferenceExpression expr, Integer indent)
+    public String visitVariableReferenceExpression(VariableReferenceExpression expr, T arg) throws AlgebricksException {
+        return expr.toString();
+    }
+
+    @Override
+    public String visitAggregateFunctionCallExpression(AggregateFunctionCallExpression expr, T arg)
             throws AlgebricksException {
         return expr.toString();
     }
 
     @Override
-    public String visitAggregateFunctionCallExpression(AggregateFunctionCallExpression expr, Integer indent)
+    public String visitScalarFunctionCallExpression(ScalarFunctionCallExpression expr, T arg)
             throws AlgebricksException {
         return expr.toString();
     }
 
     @Override
-    public String visitScalarFunctionCallExpression(ScalarFunctionCallExpression expr, Integer indent)
+    public String visitStatefulFunctionCallExpression(StatefulFunctionCallExpression expr, T arg)
             throws AlgebricksException {
         return expr.toString();
     }
 
     @Override
-    public String visitStatefulFunctionCallExpression(StatefulFunctionCallExpression expr, Integer indent)
-            throws AlgebricksException {
-        return expr.toString();
-    }
-
-    @Override
-    public String visitUnnestingFunctionCallExpression(UnnestingFunctionCallExpression expr, Integer indent)
+    public String visitUnnestingFunctionCallExpression(UnnestingFunctionCallExpression expr, T arg)
             throws AlgebricksException {
         return expr.toString();
     }
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 142d4cc..52c7e2a 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
@@ -70,43 +70,55 @@
 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.operators.logical.WriteResultOperator;
-import org.apache.hyracks.algebricks.core.algebra.visitors.ILogicalExpressionVisitor;
 
-public class LogicalOperatorPrettyPrintVisitor extends AbstractLogicalOperatorPrettyPrintVisitor {
+public class LogicalOperatorPrettyPrintVisitor extends AbstractLogicalOperatorPrettyPrintVisitor<Integer>
+        implements IPlanPrettyPrinter {
 
-    public LogicalOperatorPrettyPrintVisitor() {
-        super();
-    }
+    private static final int INIT_INDENT = 2;
+    private static final int SUBPLAN_INDENT = INIT_INDENT * 5;
 
-    public LogicalOperatorPrettyPrintVisitor(AlgebricksAppendable buffer,
-            ILogicalExpressionVisitor<String, Integer> exprVisitor) {
-        super(buffer, exprVisitor);
-    }
-
-    public LogicalOperatorPrettyPrintVisitor(AlgebricksAppendable buffer) {
-        super(buffer);
-    }
-
-    public LogicalOperatorPrettyPrintVisitor(Appendable app) {
-        super(app);
+    LogicalOperatorPrettyPrintVisitor() {
+        super(new LogicalExpressionPrettyPrintVisitor<>());
     }
 
     @Override
-    public void printOperator(AbstractLogicalOperator op, int indent) throws AlgebricksException {
-        final AlgebricksAppendable out = this.get();
+    public final IPlanPrettyPrinter reset() throws AlgebricksException {
+        resetState();
+        return this;
+    }
+
+    @Override
+    public final IPlanPrettyPrinter printPlan(ILogicalPlan plan) throws AlgebricksException {
+        printPlanImpl(plan, 0);
+        return this;
+    }
+
+    @Override
+    public final IPlanPrettyPrinter printOperator(AbstractLogicalOperator op) throws AlgebricksException {
+        printOperatorImpl(op, 0);
+        return this;
+    }
+
+    private void printPlanImpl(ILogicalPlan plan, int indent) throws AlgebricksException {
+        for (Mutable<ILogicalOperator> root : plan.getRoots()) {
+            printOperatorImpl((AbstractLogicalOperator) root.getValue(), indent);
+        }
+    }
+
+    private void printOperatorImpl(AbstractLogicalOperator op, int indent) throws AlgebricksException {
         op.accept(this, indent);
         IPhysicalOperator pOp = op.getPhysicalOperator();
 
         if (pOp != null) {
-            out.append("\n");
-            pad(out, indent);
-            appendln(out, "-- " + pOp.toString() + "  |" + op.getExecutionMode() + "|");
+            buffer.append("\n");
+            pad(buffer, indent);
+            appendln(buffer, "-- " + pOp.toString() + "  |" + op.getExecutionMode() + "|");
         } else {
-            appendln(out, " -- |" + op.getExecutionMode() + "|");
+            appendln(buffer, " -- |" + op.getExecutionMode() + "|");
         }
 
         for (Mutable<ILogicalOperator> i : op.getInputs()) {
-            printOperator((AbstractLogicalOperator) i.getValue(), indent + 2);
+            printOperatorImpl((AbstractLogicalOperator) i.getValue(), indent + INIT_INDENT);
         }
     }
 
@@ -293,7 +305,7 @@
 
     @Override
     public Void visitUnnestMapOperator(UnnestMapOperator op, Integer indent) throws AlgebricksException {
-        AlgebricksAppendable plan = printAbstractUnnestMapOperator(op, indent, "unnest-map");
+        AlgebricksStringBuilderWriter plan = printAbstractUnnestMapOperator(op, indent, "unnest-map");
         appendSelectConditionInformation(plan, op.getSelectCondition(), indent);
         appendLimitInformation(plan, op.getOutputLimit());
         return null;
@@ -306,9 +318,9 @@
         return null;
     }
 
-    private AlgebricksAppendable printAbstractUnnestMapOperator(AbstractUnnestMapOperator op, Integer indent,
+    private AlgebricksStringBuilderWriter printAbstractUnnestMapOperator(AbstractUnnestMapOperator op, Integer indent,
             String opSignature) throws AlgebricksException {
-        AlgebricksAppendable plan = addIndent(indent).append(opSignature + " " + op.getVariables() + " <- "
+        AlgebricksStringBuilderWriter plan = addIndent(indent).append(opSignature + " " + op.getVariables() + " <- "
                 + op.getExpressionRef().getValue().accept(exprVisitor, indent));
         appendFilterInformation(plan, op.getMinFilterVars(), op.getMaxFilterVars());
         return plan;
@@ -316,7 +328,7 @@
 
     @Override
     public Void visitDataScanOperator(DataSourceScanOperator op, Integer indent) throws AlgebricksException {
-        AlgebricksAppendable plan = addIndent(indent).append(
+        AlgebricksStringBuilderWriter plan = addIndent(indent).append(
                 "data-scan " + op.getProjectVariables() + "<-" + op.getVariables() + " <- " + op.getDataSource());
         appendFilterInformation(plan, op.getMinFilterVars(), op.getMaxFilterVars());
         appendSelectConditionInformation(plan, op.getSelectCondition(), indent);
@@ -324,23 +336,21 @@
         return null;
     }
 
-    private Void appendSelectConditionInformation(AlgebricksAppendable plan,
+    private void appendSelectConditionInformation(AlgebricksStringBuilderWriter plan,
             Mutable<ILogicalExpression> selectCondition, Integer indent) throws AlgebricksException {
         if (selectCondition != null) {
             plan.append(" condition (").append(selectCondition.getValue().accept(exprVisitor, indent)).append(")");
         }
-
-        return null;
     }
 
-    private Void appendLimitInformation(AlgebricksAppendable plan, long outputLimit) throws AlgebricksException {
+    private void appendLimitInformation(AlgebricksStringBuilderWriter plan, long outputLimit)
+            throws AlgebricksException {
         if (outputLimit >= 0) {
             plan.append(" limit ").append(String.valueOf(outputLimit));
         }
-        return null;
     }
 
-    private Void appendFilterInformation(AlgebricksAppendable plan, List<LogicalVariable> minFilterVars,
+    private void appendFilterInformation(AlgebricksStringBuilderWriter plan, List<LogicalVariable> minFilterVars,
             List<LogicalVariable> maxFilterVars) throws AlgebricksException {
         if (minFilterVars != null || maxFilterVars != null) {
             plan.append(" with filter on");
@@ -351,7 +361,6 @@
         if (maxFilterVars != null) {
             plan.append(" max:" + maxFilterVars);
         }
-        return null;
     }
 
     @Override
@@ -565,7 +574,7 @@
                 } else {
                     addIndent(indent).append("       {\n");
                 }
-                printPlan(p, indent + 10);
+                printPlanImpl(p, indent + SUBPLAN_INDENT);
                 addIndent(indent).append("       }");
             }
         }
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 14c3549..6422302 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
@@ -18,6 +18,7 @@
  */
 package org.apache.hyracks.algebricks.core.algebra.prettyprint;
 
+import java.io.IOException;
 import java.util.Deque;
 import java.util.HashMap;
 import java.util.LinkedList;
@@ -36,6 +37,7 @@
 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;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractUnnestNonMapOperator;
 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;
@@ -73,39 +75,59 @@
 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.operators.logical.WriteResultOperator;
+import org.apache.hyracks.api.exceptions.ErrorCode;
 
-public class LogicalOperatorPrettyPrintVisitorJson extends AbstractLogicalOperatorPrettyPrintVisitor {
-    Map<AbstractLogicalOperator, String> operatorIdentity = new HashMap<>();
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.util.DefaultIndenter;
+import com.fasterxml.jackson.core.util.DefaultPrettyPrinter;
 
-    public LogicalOperatorPrettyPrintVisitorJson() {
+public class LogicalOperatorPrettyPrintVisitorJson extends AbstractLogicalOperatorPrettyPrintVisitor<Void>
+        implements IPlanPrettyPrinter {
+
+    private static final JsonFactory JSON_FACTORY = new JsonFactory();
+    private static final DefaultIndenter OBJECT_INDENT = new DefaultIndenter("   ", DefaultIndenter.SYS_LF);
+    private static final String OPERATOR_FIELD = "operator";
+    private static final String VARIABLES_FIELD = "variables";
+    private static final String EXPRESSIONS_FIELD = "expressions";
+    private static final String EXPRESSION_FIELD = "expression";
+    private static final String CONDITION_FIELD = "condition";
+
+    private final Map<AbstractLogicalOperator, String> operatorIdentity = new HashMap<>();
+    private final IdCounter idCounter = new IdCounter();
+    private final JsonGenerator jsonGenerator;
+
+    LogicalOperatorPrettyPrintVisitorJson() {
+        super(new LogicalExpressionPrettyPrintVisitor<>());
+        DefaultPrettyPrinter prettyPrinter = new DefaultPrettyPrinter(DefaultIndenter.SYS_LF);
+        prettyPrinter.indentObjectsWith(OBJECT_INDENT);
+        try {
+            jsonGenerator = JSON_FACTORY.createGenerator(buffer).setPrettyPrinter(prettyPrinter);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
     }
 
-    public LogicalOperatorPrettyPrintVisitorJson(Appendable app) {
-        super(app);
-    }
-
-    IdCounter idCounter = new IdCounter();
-
-    public class IdCounter {
+    private class IdCounter {
         private int id;
         private final Deque<Integer> prefix;
 
-        public IdCounter() {
-            prefix = new LinkedList<Integer>();
+        private IdCounter() {
+            prefix = new LinkedList<>();
             prefix.add(1);
             this.id = 0;
         }
 
-        public void previousPrefix() {
+        private void previousPrefix() {
             this.id = prefix.removeLast();
         }
 
-        public void nextPrefix() {
+        private void nextPrefix() {
             prefix.add(this.id);
             this.id = 0;
         }
 
-        public String printOperatorId(AbstractLogicalOperator op) {
+        private String printOperatorId(AbstractLogicalOperator op) {
             String stringPrefix = "";
             Object[] values = this.prefix.toArray();
             for (Object val : values) {
@@ -121,531 +143,673 @@
     }
 
     @Override
-    public AlgebricksAppendable reset(AlgebricksAppendable buffer) {
+    public final IPlanPrettyPrinter reset() throws AlgebricksException {
+        flushContentToWriter();
+        resetState();
         operatorIdentity.clear();
-        return super.reset(buffer);
+        return this;
     }
 
     @Override
-    public void printOperator(AbstractLogicalOperator op, int indent) throws AlgebricksException {
-        int currentIndent = indent;
-        final AlgebricksAppendable out = get();
-        pad(out, currentIndent);
-        appendln(out, "{");
-        currentIndent++;
-        op.accept(this, currentIndent);
-        appendln(out, ",");
-        pad(out, currentIndent);
-        append(out, "\"operatorId\": \"" + idCounter.printOperatorId(op) + "\"");
-        IPhysicalOperator pOp = op.getPhysicalOperator();
-        if (pOp != null) {
-            appendln(out, ",");
-            pad(out, currentIndent);
-            String pOperator = "\"physical-operator\": \"" + pOp.toString() + "\"";
-            append(out, pOperator);
+    public final IPlanPrettyPrinter printPlan(ILogicalPlan plan) throws AlgebricksException {
+        printPlanImpl(plan);
+        flushContentToWriter();
+        return this;
+    }
+
+    @Override
+    public final IPlanPrettyPrinter printOperator(AbstractLogicalOperator op) throws AlgebricksException {
+        printOperatorImpl(op);
+        flushContentToWriter();
+        return this;
+    }
+
+    private void printPlanImpl(ILogicalPlan plan) throws AlgebricksException {
+        try {
+            boolean writeArrayOfRoots = plan.getRoots().size() > 1;
+            if (writeArrayOfRoots) {
+                jsonGenerator.writeStartArray();
+            }
+            for (Mutable<ILogicalOperator> root : plan.getRoots()) {
+                printOperatorImpl((AbstractLogicalOperator) root.getValue());
+            }
+            if (writeArrayOfRoots) {
+                jsonGenerator.writeEndArray();
+            }
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
         }
-        appendln(out, ",");
-        pad(out, currentIndent);
-        append(out, "\"execution-mode\": \"" + op.getExecutionMode() + '"');
-        if (!op.getInputs().isEmpty()) {
-            appendln(out, ",");
-            pad(out, currentIndent);
-            appendln(out, "\"inputs\": [");
-            boolean moreInputes = false;
-            for (Mutable<ILogicalOperator> k : op.getInputs()) {
-                if (moreInputes) {
-                    append(out, ",");
+    }
+
+    private void printOperatorImpl(AbstractLogicalOperator op) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStartObject();
+            op.accept(this, null);
+            jsonGenerator.writeStringField("operatorId", idCounter.printOperatorId(op));
+            IPhysicalOperator pOp = op.getPhysicalOperator();
+            if (pOp != null) {
+                jsonGenerator.writeStringField("physical-operator", pOp.toString());
+            }
+            jsonGenerator.writeStringField("execution-mode", op.getExecutionMode().toString());
+            if (!op.getInputs().isEmpty()) {
+                jsonGenerator.writeArrayFieldStart("inputs");
+                for (Mutable<ILogicalOperator> k : op.getInputs()) {
+                    printOperatorImpl((AbstractLogicalOperator) k.getValue());
                 }
-                printOperator((AbstractLogicalOperator) k.getValue(), currentIndent + 4);
-                moreInputes = true;
+                jsonGenerator.writeEndArray();
             }
-            pad(out, currentIndent + 2);
-            appendln(out, "]");
-        } else {
-            out.append("\n");
-        }
-        pad(out, currentIndent - 1);
-        appendln(out, "}");
-    }
-
-    public void variablePrintHelper(List<LogicalVariable> variables, Integer indent) throws AlgebricksException {
-        if (!variables.isEmpty()) {
-            addIndent(0).append(",\n");
-            addIndent(indent).append("\"variables\": [");
-            appendVars(variables);
-            buffer.append("]");
+            jsonGenerator.writeEndObject();
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
         }
     }
 
     @Override
-    public Void visitAggregateOperator(AggregateOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"aggregate\"");
-        if (!op.getExpressions().isEmpty()) {
-            addIndent(0).append(",\n");
-            pprintExprList(op.getExpressions(), indent);
+    public Void visitAggregateOperator(AggregateOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "aggregate");
+            writeVariablesAndExpressions(op.getVariables(), op.getExpressions(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
         }
-        variablePrintHelper(op.getVariables(), indent);
-        return null;
     }
 
     @Override
-    public Void visitRunningAggregateOperator(RunningAggregateOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"running-aggregate\"");
-        variablePrintHelper(op.getVariables(), indent);
-        if (!op.getExpressions().isEmpty()) {
-            addIndent(0).append(",\n");
-            pprintExprList(op.getExpressions(), indent);
+    public Void visitRunningAggregateOperator(RunningAggregateOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "running-aggregate");
+            writeVariablesAndExpressions(op.getVariables(), op.getExpressions(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
         }
-        return null;
     }
 
     @Override
-    public Void visitEmptyTupleSourceOperator(EmptyTupleSourceOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"empty-tuple-source\"");
-        return null;
-    }
-
-    @Override
-    public Void visitGroupByOperator(GroupByOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"group-by\"");
-
-        if (op.isGroupAll()) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"option\": \"all\"");
+    public Void visitEmptyTupleSourceOperator(EmptyTupleSourceOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "empty-tuple-source");
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
         }
-        if (!op.getGroupByList().isEmpty()) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"group-by-list\": ");
-            pprintVeList(op.getGroupByList(), indent);
-        }
-        if (!op.getDecorList().isEmpty()) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"decor-list\": ");
-            pprintVeList(op.getDecorList(), indent);
-        }
-        if (!op.getNestedPlans().isEmpty()) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"subplan\": ");
-            printNestedPlans(op, indent);
-        }
-        return null;
     }
 
     @Override
-    public Void visitDistinctOperator(DistinctOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"distinct\"");
-        if (!op.getExpressions().isEmpty()) {
-            addIndent(0).append(",\n");
-            pprintExprList(op.getExpressions(), indent);
-        }
-        return null;
-    }
-
-    @Override
-    public Void visitInnerJoinOperator(InnerJoinOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"join\",\n");
-        addIndent(indent).append("\"condition\": \"" + op.getCondition().getValue().accept(exprVisitor, indent) + "\"");
-        return null;
-    }
-
-    @Override
-    public Void visitLeftOuterJoinOperator(LeftOuterJoinOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"left-outer-join\",\n");
-        addIndent(indent).append("\"condition\": \"" + op.getCondition().getValue().accept(exprVisitor, indent) + "\"");
-        return null;
-    }
-
-    @Override
-    public Void visitNestedTupleSourceOperator(NestedTupleSourceOperator op, Integer indent)
-            throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"nested-tuple-source\"");
-        return null;
-    }
-
-    @Override
-    public Void visitOrderOperator(OrderOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"order\"");
-        buffer.append(",\n");
-        int topK = op.getTopK();
-        if (topK != -1) {
-            addIndent(indent).append("\"topK\": \"" + topK + "\",\n");
-        }
-        addIndent(indent).append("\"order-by-list\": ");
-        pprintOrderExprList(op.getOrderExpressions(), indent);
-        return null;
-    }
-
-    @Override
-    public Void visitAssignOperator(AssignOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"assign\"");
-        variablePrintHelper(op.getVariables(), indent);
-        if (!op.getExpressions().isEmpty()) {
-            addIndent(0).append(",\n");
-            pprintExprList(op.getExpressions(), indent);
-        }
-        return null;
-    }
-
-    @Override
-    public Void visitWriteOperator(WriteOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"write\"");
-        if (!op.getExpressions().isEmpty()) {
-            addIndent(0).append(",\n");
-            pprintExprList(op.getExpressions(), indent);
-        }
-        return null;
-    }
-
-    @Override
-    public Void visitDistributeResultOperator(DistributeResultOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"distribute-result\"");
-        if (!op.getExpressions().isEmpty()) {
-            addIndent(0).append(",\n");
-            pprintExprList(op.getExpressions(), indent);
-        }
-        return null;
-    }
-
-    @Override
-    public Void visitWriteResultOperator(WriteResultOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"load\",\n");
-        addIndent(indent).append(str(op.getDataSource())).append("\"from\":")
-                .append(op.getPayloadExpression().getValue().accept(exprVisitor, indent) + ",\n");
-        addIndent(indent).append("\"partitioned-by\": {");
-        pprintExprList(op.getKeyExpressions(), indent);
-        addIndent(indent).append("}");
-        return null;
-    }
-
-    @Override
-    public Void visitSelectOperator(SelectOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"select\",\n");
-        addIndent(indent).append("\"expressions\": \""
-                + op.getCondition().getValue().accept(exprVisitor, indent).replace('"', ' ') + "\"");
-        return null;
-    }
-
-    @Override
-    public Void visitProjectOperator(ProjectOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"project\"");
-        variablePrintHelper(op.getVariables(), indent);
-        return null;
-    }
-
-    @Override
-    public Void visitSubplanOperator(SubplanOperator op, Integer indent) throws AlgebricksException {
-        if (!op.getNestedPlans().isEmpty()) {
-            addIndent(indent).append("\"subplan\": ");
-            printNestedPlans(op, indent);
-        }
-        return null;
-    }
-
-    @Override
-    public Void visitUnionOperator(UnionAllOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"union\"");
-        for (Triple<LogicalVariable, LogicalVariable, LogicalVariable> v : op.getVariableMappings()) {
-            buffer.append(",\n");
-            addIndent(indent).append(
-                    "\"values\": [" + "\"" + v.first + "\"," + "\"" + v.second + "\"," + "\"" + v.third + "\"]");
-        }
-        return null;
-    }
-
-    @Override
-    public Void visitIntersectOperator(IntersectOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"intersect\",\n");
-
-        addIndent(indent).append("\"output-compare-variables\": [");
-        appendVars(op.getOutputCompareVariables());
-        buffer.append(']');
-        if (op.hasExtraVariables()) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"output-extra-variables\": [");
-            appendVars(op.getOutputExtraVariables());
-            buffer.append(']');
-        }
-        buffer.append(",\n");
-        addIndent(indent).append("\"input-compare-variables\": [");
-        for (int i = 0, n = op.getNumInput(); i < n; i++) {
-            if (i > 0) {
-                buffer.append(", ");
+    public Void visitGroupByOperator(GroupByOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "group-by");
+            if (op.isGroupAll()) {
+                jsonGenerator.writeStringField("option", "all");
             }
-            buffer.append('[');
-            appendVars(op.getInputCompareVariables(i));
-            buffer.append(']');
+            if (!op.getGroupByList().isEmpty()) {
+                writeArrayFieldOfVariableExpressionPairs("group-by-list", op.getGroupByList(), indent);
+            }
+            if (!op.getDecorList().isEmpty()) {
+                writeArrayFieldOfVariableExpressionPairs("decor-list", op.getDecorList(), indent);
+            }
+            if (!op.getNestedPlans().isEmpty()) {
+                writeNestedPlans(op, indent);
+            }
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
         }
-        buffer.append(']');
-        if (op.hasExtraVariables()) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"input-extra-variables\": [");
-            for (int i = 0, n = op.getNumInput(); i < n; i++) {
-                if (i > 0) {
-                    buffer.append(", ");
+    }
+
+    @Override
+    public Void visitDistinctOperator(DistinctOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "distinct");
+            List<Mutable<ILogicalExpression>> expressions = op.getExpressions();
+            if (!expressions.isEmpty()) {
+                writeArrayFieldOfExpressions(EXPRESSIONS_FIELD, expressions, indent);
+            }
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitInnerJoinOperator(InnerJoinOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "join");
+            writeStringFieldExpression(CONDITION_FIELD, op.getCondition(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitLeftOuterJoinOperator(LeftOuterJoinOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "left-outer-join");
+            writeStringFieldExpression(CONDITION_FIELD, op.getCondition(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitNestedTupleSourceOperator(NestedTupleSourceOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "nested-tuple-source");
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitOrderOperator(OrderOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "order");
+            int topK = op.getTopK();
+            if (topK != -1) {
+                jsonGenerator.writeStringField("topK", String.valueOf(topK));
+            }
+            writeArrayFieldOfOrderExprList("order-by-list", op.getOrderExpressions(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitAssignOperator(AssignOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "assign");
+            writeVariablesAndExpressions(op.getVariables(), op.getExpressions(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitWriteOperator(WriteOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "write");
+            List<Mutable<ILogicalExpression>> expressions = op.getExpressions();
+            if (!expressions.isEmpty()) {
+                writeArrayFieldOfExpressions(EXPRESSIONS_FIELD, expressions, indent);
+            }
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitDistributeResultOperator(DistributeResultOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "distribute-result");
+            List<Mutable<ILogicalExpression>> expressions = op.getExpressions();
+            if (!expressions.isEmpty()) {
+                writeArrayFieldOfExpressions(EXPRESSIONS_FIELD, expressions, indent);
+            }
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitWriteResultOperator(WriteResultOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "load");
+            jsonGenerator.writeStringField("data-source", String.valueOf(op.getDataSource()));
+            writeStringFieldExpression("from", op.getPayloadExpression(), indent);
+            writeObjectFieldWithExpressions("partitioned-by", op.getKeyExpressions(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitSelectOperator(SelectOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "select");
+            writeStringFieldExpression(EXPRESSIONS_FIELD, op.getCondition(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitProjectOperator(ProjectOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "project");
+            List<LogicalVariable> variables = op.getVariables();
+            if (!variables.isEmpty()) {
+                writeArrayFieldOfVariables(VARIABLES_FIELD, variables);
+            }
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitSubplanOperator(SubplanOperator op, Void indent) throws AlgebricksException {
+        if (!op.getNestedPlans().isEmpty()) {
+            writeNestedPlans(op, indent);
+        }
+        return null;
+    }
+
+    @Override
+    public Void visitUnionOperator(UnionAllOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "union");
+            jsonGenerator.writeArrayFieldStart("values");
+            for (Triple<LogicalVariable, LogicalVariable, LogicalVariable> v : op.getVariableMappings()) {
+                jsonGenerator.writeStartArray();
+                jsonGenerator.writeString(String.valueOf(v.first));
+                jsonGenerator.writeString(String.valueOf(v.second));
+                jsonGenerator.writeString(String.valueOf(v.third));
+                jsonGenerator.writeEndArray();
+            }
+            jsonGenerator.writeEndArray();
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitIntersectOperator(IntersectOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "intersect");
+            writeArrayFieldOfVariables("output-compare-variables", op.getOutputCompareVariables());
+            if (op.hasExtraVariables()) {
+                writeArrayFieldOfVariables("output-extra-variables", op.getOutputExtraVariables());
+            }
+            writeArrayFieldOfNestedVariablesList("input-compare-variables", op.getAllInputsCompareVariables());
+            if (op.hasExtraVariables()) {
+                writeArrayFieldOfNestedVariablesList("input-extra-variables", op.getAllInputsExtraVariables());
+            }
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitUnnestOperator(UnnestOperator op, Void indent) throws AlgebricksException {
+        writeUnnestNonMapOperator(op, "unnest", indent);
+        return null;
+    }
+
+    @Override
+    public Void visitLeftOuterUnnestOperator(LeftOuterUnnestOperator op, Void indent) throws AlgebricksException {
+        writeUnnestNonMapOperator(op, "outer-unnest", indent);
+        return null;
+    }
+
+    @Override
+    public Void visitUnnestMapOperator(UnnestMapOperator op, Void indent) throws AlgebricksException {
+        try {
+            writeUnnestMapOperator(op, indent, "unnest-map");
+            writeSelectLimitInformation(op.getSelectCondition(), op.getOutputLimit(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitLeftOuterUnnestMapOperator(LeftOuterUnnestMapOperator op, Void indent) throws AlgebricksException {
+        writeUnnestMapOperator(op, indent, "left-outer-unnest-map");
+        return null;
+    }
+
+    @Override
+    public Void visitDataScanOperator(DataSourceScanOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "data-scan");
+            List<LogicalVariable> projectVariables = op.getProjectVariables();
+            if (!projectVariables.isEmpty()) {
+                writeArrayFieldOfVariables("project-variables", projectVariables);
+            }
+            List<LogicalVariable> variables = op.getVariables();
+            if (!variables.isEmpty()) {
+                writeArrayFieldOfVariables(VARIABLES_FIELD, variables);
+            }
+            if (op.getDataSource() != null) {
+                jsonGenerator.writeStringField("data-source", String.valueOf(op.getDataSource()));
+            }
+            writeFilterInformation(op.getMinFilterVars(), op.getMaxFilterVars());
+            writeSelectLimitInformation(op.getSelectCondition(), op.getOutputLimit(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitLimitOperator(LimitOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "limit");
+            writeStringFieldExpression("value", op.getMaxObjects(), indent);
+            Mutable<ILogicalExpression> offsetRef = op.getOffset();
+            if (offsetRef != null && offsetRef.getValue() != null) {
+                writeStringFieldExpression("offset", offsetRef, indent);
+            }
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitExchangeOperator(ExchangeOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "exchange");
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitScriptOperator(ScriptOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "script");
+            List<LogicalVariable> inputVariables = op.getInputVariables();
+            if (!inputVariables.isEmpty()) {
+                writeArrayFieldOfVariables("in", inputVariables);
+            }
+            List<LogicalVariable> outputVariables = op.getOutputVariables();
+            if (!outputVariables.isEmpty()) {
+                writeArrayFieldOfVariables("out", outputVariables);
+            }
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitReplicateOperator(ReplicateOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "replicate");
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitSplitOperator(SplitOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "split");
+            writeStringFieldExpression(EXPRESSION_FIELD, op.getBranchingExpression(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitMaterializeOperator(MaterializeOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "materialize");
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitInsertDeleteUpsertOperator(InsertDeleteUpsertOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, getIndexOpString(op.getOperation()));
+            jsonGenerator.writeStringField("data-source", String.valueOf(op.getDataSource()));
+            writeStringFieldExpression("from-record", op.getPayloadExpression(), indent);
+            if (op.getAdditionalNonFilteringExpressions() != null) {
+                writeObjectFieldWithExpressions("meta", op.getAdditionalNonFilteringExpressions(), indent);
+            }
+            writeObjectFieldWithExpressions("partitioned-by", op.getPrimaryKeyExpressions(), indent);
+            if (op.getOperation() == Kind.UPSERT) {
+                jsonGenerator.writeObjectFieldStart("out");
+                jsonGenerator.writeStringField("record-before-upsert", String.valueOf(op.getBeforeOpRecordVar()));
+                if (op.getBeforeOpAdditionalNonFilteringVars() != null) {
+                    jsonGenerator.writeStringField("additional-before-upsert",
+                            String.valueOf(op.getBeforeOpAdditionalNonFilteringVars()));
                 }
-                buffer.append('[');
-                appendVars(op.getInputExtraVariables(i));
-                buffer.append(']');
+                jsonGenerator.writeEndObject();
             }
-            buffer.append(']');
+            if (op.isBulkload()) {
+                jsonGenerator.writeBooleanField("bulkload", true);
+            }
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
         }
-        return null;
     }
 
     @Override
-    public Void visitUnnestOperator(UnnestOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"unnest\"");
-        variablePrintHelper(op.getVariables(), indent);
-        if (op.getPositionalVariable() != null) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"position\": \"" + op.getPositionalVariable() + "\"");
-        }
-        buffer.append(",\n");
-        addIndent(indent).append("\"expressions\": \""
-                + op.getExpressionRef().getValue().accept(exprVisitor, indent).replace('"', ' ') + "\"");
-        return null;
-    }
-
-    @Override
-    public Void visitLeftOuterUnnestOperator(LeftOuterUnnestOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"outer-unnest\",\n");
-        addIndent(indent).append("\"variables\": [\"" + op.getVariable() + "\"]");
-        if (op.getPositionalVariable() != null) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"position\": " + op.getPositionalVariable());
-        }
-        buffer.append(",\n");
-        addIndent(indent).append("\"expressions\": \""
-                + op.getExpressionRef().getValue().accept(exprVisitor, indent).replace('"', ' ') + "\"");
-        return null;
-    }
-
-    @Override
-    public Void visitUnnestMapOperator(UnnestMapOperator op, Integer indent) throws AlgebricksException {
-        AlgebricksAppendable plan = printAbstractUnnestMapOperator(op, indent, "unnest-map");
-        appendSelectConditionInformation(plan, op.getSelectCondition(), indent);
-        appendLimitInformation(plan, op.getOutputLimit(), indent);
-        return null;
-    }
-
-    @Override
-    public Void visitLeftOuterUnnestMapOperator(LeftOuterUnnestMapOperator op, Integer indent)
+    public Void visitIndexInsertDeleteUpsertOperator(IndexInsertDeleteUpsertOperator op, Void indent)
             throws AlgebricksException {
-        printAbstractUnnestMapOperator(op, indent, "left-outer-unnest-map");
-        return null;
-    }
-
-    private AlgebricksAppendable printAbstractUnnestMapOperator(AbstractUnnestMapOperator op, Integer indent,
-            String opSignature) throws AlgebricksException {
-        AlgebricksAppendable plan = addIndent(indent).append("\"operator\": \"" + opSignature + "\"");
-        variablePrintHelper(op.getVariables(), indent);
-        buffer.append(",\n");
-        addIndent(indent).append("\"expressions\": \""
-                + op.getExpressionRef().getValue().accept(exprVisitor, indent).replace('"', ' ') + "\"");
-        appendFilterInformation(plan, op.getMinFilterVars(), op.getMaxFilterVars(), indent);
-        return plan;
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, getIndexOpString(op.getOperation()));
+            jsonGenerator.writeStringField("index", op.getIndexName());
+            jsonGenerator.writeStringField("on", String.valueOf(op.getDataSourceIndex().getDataSource()));
+            jsonGenerator.writeObjectFieldStart("from");
+            if (op.getOperation() == Kind.UPSERT) {
+                writeArrayFieldOfExpressions("replace", op.getPrevSecondaryKeyExprs(), indent);
+                writeArrayFieldOfExpressions("with", op.getSecondaryKeyExpressions(), indent);
+            } else {
+                writeArrayFieldOfExpressions(EXPRESSIONS_FIELD, op.getSecondaryKeyExpressions(), indent);
+            }
+            jsonGenerator.writeEndObject();
+            if (op.isBulkload()) {
+                jsonGenerator.writeBooleanField("bulkload", true);
+            }
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
     }
 
     @Override
-    public Void visitDataScanOperator(DataSourceScanOperator op, Integer indent) throws AlgebricksException {
-        AlgebricksAppendable plan = addIndent(indent).append("\"operator\": \"data-scan\"");
-        if (!op.getProjectVariables().isEmpty()) {
-            addIndent(0).append(",\n");
-            addIndent(indent).append("\"project-variables\": [");
-            appendVars(op.getProjectVariables());
-            buffer.append("]");
+    public Void visitTokenizeOperator(TokenizeOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "tokenize");
+            writeVariablesAndExpressions(op.getTokenizeVars(), op.getSecondaryKeyExpressions(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
         }
-        variablePrintHelper(op.getVariables(), indent);
-        if (op.getDataSource() != null) {
-            addIndent(0).append(",\n");
-            addIndent(indent).append("\"data-source\": \"" + op.getDataSource() + "\"");
-        }
-        appendFilterInformation(plan, op.getMinFilterVars(), op.getMaxFilterVars(), indent);
-        appendSelectConditionInformation(plan, op.getSelectCondition(), indent);
-        appendLimitInformation(plan, op.getOutputLimit(), indent);
-        return null;
     }
 
-    private Void appendFilterInformation(AlgebricksAppendable plan, List<LogicalVariable> minFilterVars,
-            List<LogicalVariable> maxFilterVars, Integer indent) throws AlgebricksException {
-        if (minFilterVars != null || maxFilterVars != null) {
-            plan.append(",\n");
-            addIndent(indent);
-            plan.append("\"with-filter-on\": {");
+    @Override
+    public Void visitForwardOperator(ForwardOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "forward");
+            writeStringFieldExpression(EXPRESSION_FIELD, op.getSideDataExpression(), indent);
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
         }
-        if (minFilterVars != null) {
-            buffer.append("\n");
-            addIndent(indent).append("\"min\": [");
-            appendVars(minFilterVars);
-            buffer.append("]");
-        }
-        if (minFilterVars != null && maxFilterVars != null) {
-            buffer.append(",");
-        }
-        if (maxFilterVars != null) {
-            buffer.append("\n");
-            addIndent(indent).append("\"max\": [");
-            appendVars(maxFilterVars);
-            buffer.append("]");
-        }
-        if (minFilterVars != null || maxFilterVars != null) {
-            plan.append("\n");
-            addIndent(indent).append("}");
-        }
-        return null;
     }
 
-    private Void appendSelectConditionInformation(AlgebricksAppendable plan, Mutable<ILogicalExpression> condition,
-            Integer indent) throws AlgebricksException {
-        if (condition != null) {
-            plan.append(",\n");
-            addIndent(indent).append(
-                    "\"condition\": \"" + condition.getValue().accept(exprVisitor, indent).replace('"', ' ') + "\"");
+    @Override
+    public Void visitSinkOperator(SinkOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "sink");
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
         }
-        return null;
     }
 
-    private Void appendLimitInformation(AlgebricksAppendable plan, long outputLimit, Integer indent)
+    @Override
+    public Void visitDelegateOperator(DelegateOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, op.toString());
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    @Override
+    public Void visitWindowOperator(WindowOperator op, Void indent) throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, "window-aggregate");
+            writeVariablesAndExpressions(op.getVariables(), op.getExpressions(), indent);
+            List<Mutable<ILogicalExpression>> partitionExpressions = op.getPartitionExpressions();
+            if (!partitionExpressions.isEmpty()) {
+                writeObjectFieldWithExpressions("partition-by", partitionExpressions, indent);
+            }
+            List<Pair<OrderOperator.IOrder, Mutable<ILogicalExpression>>> orderExpressions = op.getOrderExpressions();
+            if (!orderExpressions.isEmpty()) {
+                writeArrayFieldOfOrderExprList("order-by", orderExpressions, indent);
+            }
+            if (op.hasNestedPlans()) {
+                List<Pair<OrderOperator.IOrder, Mutable<ILogicalExpression>>> frameValueExpressions =
+                        op.getFrameValueExpressions();
+                if (!frameValueExpressions.isEmpty()) {
+                    writeArrayFieldOfOrderExprList("frame-on", frameValueExpressions, indent);
+                }
+                List<Mutable<ILogicalExpression>> frameStartExpressions = op.getFrameStartExpressions();
+                if (!frameStartExpressions.isEmpty()) {
+                    writeObjectFieldWithExpressions("frame-start", frameStartExpressions, indent);
+                }
+                List<Mutable<ILogicalExpression>> frameStartValidationExpressions =
+                        op.getFrameStartValidationExpressions();
+                if (!frameStartValidationExpressions.isEmpty()) {
+                    writeObjectFieldWithExpressions("frame-start-if", frameStartValidationExpressions, indent);
+                }
+                List<Mutable<ILogicalExpression>> frameEndExpressions = op.getFrameEndExpressions();
+                if (!frameEndExpressions.isEmpty()) {
+                    writeObjectFieldWithExpressions("frame-end", frameEndExpressions, indent);
+                }
+                List<Mutable<ILogicalExpression>> frameEndValidationExpressions = op.getFrameEndValidationExpressions();
+                if (!frameEndValidationExpressions.isEmpty()) {
+                    writeObjectFieldWithExpressions("frame-end-if", frameEndValidationExpressions, indent);
+                }
+                List<Mutable<ILogicalExpression>> frameExcludeExpressions = op.getFrameExcludeExpressions();
+                if (!frameExcludeExpressions.isEmpty()) {
+                    writeObjectFieldWithExpressions("frame-exclude", frameExcludeExpressions, indent);
+                    jsonGenerator.writeStringField("frame-exclude-negation-start",
+                            String.valueOf(op.getFrameExcludeNegationStartIdx()));
+                }
+                Mutable<ILogicalExpression> frameExcludeUnaryExpression = op.getFrameExcludeUnaryExpression();
+                if (frameExcludeUnaryExpression.getValue() != null) {
+                    writeStringFieldExpression("frame-exclude-unary", frameExcludeUnaryExpression, indent);
+                }
+                Mutable<ILogicalExpression> frameOffsetExpression = op.getFrameOffsetExpression();
+                if (frameOffsetExpression.getValue() != null) {
+                    writeStringFieldExpression("frame-offset", frameOffsetExpression, indent);
+                }
+                int frameMaxObjects = op.getFrameMaxObjects();
+                if (frameMaxObjects != -1) {
+                    jsonGenerator.writeStringField("frame-max-objects", String.valueOf(frameMaxObjects));
+                }
+                writeNestedPlans(op, indent);
+            }
+            return null;
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    private void writeNestedPlans(AbstractOperatorWithNestedPlans op, Void indent) throws AlgebricksException {
+        try {
+            idCounter.nextPrefix();
+            jsonGenerator.writeArrayFieldStart("subplan");
+            List<ILogicalPlan> nestedPlans = op.getNestedPlans();
+            for (int i = 0, size = nestedPlans.size(); i < size; i++) {
+                printPlanImpl(nestedPlans.get(i));
+            }
+            jsonGenerator.writeEndArray();
+            idCounter.previousPrefix();
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    private void writeUnnestNonMapOperator(AbstractUnnestNonMapOperator op, String opName, Void indent)
             throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, opName);
+            List<LogicalVariable> variables = op.getVariables();
+            if (!variables.isEmpty()) {
+                writeArrayFieldOfVariables(VARIABLES_FIELD, variables);
+            }
+            LogicalVariable positionalVariable = op.getPositionalVariable();
+            if (positionalVariable != null) {
+                jsonGenerator.writeStringField("position", String.valueOf(positionalVariable));
+            }
+            writeArrayFieldOfExpression(EXPRESSIONS_FIELD, op.getExpressionRef(), indent);
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    private void writeUnnestMapOperator(AbstractUnnestMapOperator op, Void indent, String opName)
+            throws AlgebricksException {
+        try {
+            jsonGenerator.writeStringField(OPERATOR_FIELD, opName);
+            List<LogicalVariable> variables = op.getVariables();
+            if (!variables.isEmpty()) {
+                writeArrayFieldOfVariables(VARIABLES_FIELD, variables);
+            }
+            writeArrayFieldOfExpression(EXPRESSIONS_FIELD, op.getExpressionRef(), indent);
+            writeFilterInformation(op.getMinFilterVars(), op.getMaxFilterVars());
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    private void writeFilterInformation(List<LogicalVariable> minFilterVars, List<LogicalVariable> maxFilterVars)
+            throws AlgebricksException {
+        try {
+            if (minFilterVars != null || maxFilterVars != null) {
+                jsonGenerator.writeObjectFieldStart("with-filter-on");
+                if (minFilterVars != null) {
+                    writeArrayFieldOfVariables("min", minFilterVars);
+                }
+                if (maxFilterVars != null) {
+                    writeArrayFieldOfVariables("max", maxFilterVars);
+                }
+                jsonGenerator.writeEndObject();
+            }
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
+        }
+    }
+
+    private void writeSelectLimitInformation(Mutable<ILogicalExpression> selectCondition, long outputLimit, Void i)
+            throws AlgebricksException, IOException {
+        if (selectCondition != null) {
+            writeStringFieldExpression(CONDITION_FIELD, selectCondition, i);
+        }
         if (outputLimit >= 0) {
-            plan.append(",\n");
-            addIndent(indent).append("\"limit\": \"" + outputLimit + "\"");
-        }
-        return null;
-    }
-
-    private void appendVars(List<LogicalVariable> minFilterVars) throws AlgebricksException {
-        boolean first = true;
-        for (LogicalVariable v : minFilterVars) {
-            if (!first) {
-                buffer.append(",");
-            }
-            buffer.append("\"" + str(v) + "\"");
-            first = false;
+            jsonGenerator.writeStringField("limit", String.valueOf(outputLimit));
         }
     }
 
-    @Override
-    public Void visitLimitOperator(LimitOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"limit\",\n");
-        addIndent(indent).append("\"value\": \"" + op.getMaxObjects().getValue().accept(exprVisitor, indent) + "\"");
-        ILogicalExpression offset = op.getOffset().getValue();
-        if (offset != null) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"offset\": \"" + offset.accept(exprVisitor, indent) + "\"");
+    private void writeVariablesAndExpressions(List<LogicalVariable> variables,
+            List<Mutable<ILogicalExpression>> expressions, Void indent) throws IOException, AlgebricksException {
+        if (!variables.isEmpty()) {
+            writeArrayFieldOfVariables(VARIABLES_FIELD, variables);
         }
-        return null;
+        if (!expressions.isEmpty()) {
+            writeArrayFieldOfExpressions(EXPRESSIONS_FIELD, expressions, indent);
+        }
     }
 
-    @Override
-    public Void visitExchangeOperator(ExchangeOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"exchange\"");
-        return null;
-    }
-
-    @Override
-    public Void visitScriptOperator(ScriptOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"script\"");
-        if (!op.getInputVariables().isEmpty()) {
-            addIndent(0).append(",\n");
-            addIndent(indent).append("\"in\": [");
-            appendVars(op.getInputVariables());
-            buffer.append("]");
-        }
-        if (!op.getOutputVariables().isEmpty()) {
-            addIndent(0).append(",\n");
-            addIndent(indent).append("\"out\": [");
-            appendVars(op.getOutputVariables());
-            buffer.append("]");
-        }
-        return null;
-    }
-
-    @Override
-    public Void visitReplicateOperator(ReplicateOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"replicate\"");
-        return null;
-    }
-
-    @Override
-    public Void visitSplitOperator(SplitOperator op, Integer indent) throws AlgebricksException {
-        Mutable<ILogicalExpression> branchingExpression = op.getBranchingExpression();
-        addIndent(indent).append("\"operator\": \"split\",\n");
-        addIndent(indent).append("\"expressions\": \""
-                + branchingExpression.getValue().accept(exprVisitor, indent).replace('"', ' ') + "\"");
-        return null;
-    }
-
-    @Override
-    public Void visitMaterializeOperator(MaterializeOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"materialize\"");
-        return null;
-    }
-
-    @Override
-    public Void visitInsertDeleteUpsertOperator(InsertDeleteUpsertOperator op, Integer indent)
-            throws AlgebricksException {
-        String header = "\"operator\": \"" + getIndexOpString(op.getOperation()) + "\",\n";
-        addIndent(indent).append(header);
-        addIndent(indent).append(str("\"data-source\": \"" + op.getDataSource() + "\",\n"));
-        addIndent(indent).append("\"from-record\": \"")
-                .append(op.getPayloadExpression().getValue().accept(exprVisitor, indent) + "\"");
-        if (op.getAdditionalNonFilteringExpressions() != null) {
-            buffer.append(",\n\"meta\": {");
-            pprintExprList(op.getAdditionalNonFilteringExpressions(), 0);
-            buffer.append("}");
-        }
-        buffer.append(",\n");
-        addIndent(indent).append("\"partitioned-by\": {");
-        pprintExprList(op.getPrimaryKeyExpressions(), 0);
-        buffer.append("}");
-        if (op.getOperation() == Kind.UPSERT) {
-            addIndent(indent).append(",\n\"out\": {\n");
-            addIndent(indent).append("\"record-before-upsert\": \"" + op.getBeforeOpRecordVar() + "\"");
-            if (op.getBeforeOpAdditionalNonFilteringVars() != null) {
-                buffer.append(",\n");
-                addIndent(indent)
-                        .append("\"additional-before-upsert\": \"" + op.getBeforeOpAdditionalNonFilteringVars() + "\"");
-            }
-            addIndent(indent).append("}");
-        }
-        if (op.isBulkload()) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"bulkload\": true");
-        }
-        return null;
-    }
-
-    @Override
-    public Void visitIndexInsertDeleteUpsertOperator(IndexInsertDeleteUpsertOperator op, Integer indent)
-            throws AlgebricksException {
-        String header = getIndexOpString(op.getOperation());
-        addIndent(indent).append("\"operator\": \"" + header + "\",\n");
-        addIndent(indent).append("\"index\": \"" + op.getIndexName() + "\",\n");
-        addIndent(indent).append("\"on\": \"").append(str(op.getDataSourceIndex().getDataSource()) + "\",\n");
-        addIndent(indent).append("\"from\": {");
-
-        if (op.getOperation() == Kind.UPSERT) {
-
-            addIndent(indent).append("[\"replace\": \"");
-            pprintExprList(op.getPrevSecondaryKeyExprs(), 0);
-            buffer.append("\",\n");
-            addIndent(indent).append("\"with\": \"");
-            pprintExprList(op.getSecondaryKeyExpressions(), 0);
-            buffer.append("\"}");
-        } else {
-            pprintExprList(op.getSecondaryKeyExpressions(), 0);
-        }
-        buffer.append("\n");
-        addIndent(indent).append("}");
-        if (op.isBulkload()) {
-            buffer.append(",\n");
-            buffer.append("\"bulkload\": true");
-        }
-        return null;
-    }
-
-    public String getIndexOpString(Kind opKind) {
+    private String getIndexOpString(Kind opKind) {
         switch (opKind) {
             case DELETE:
                 return "delete-from";
@@ -653,220 +817,115 @@
                 return "insert-into";
             case UPSERT:
                 return "upsert-into";
+            default:
+                throw new IllegalStateException();
+
         }
-        return null;
     }
 
-    @Override
-    public Void visitTokenizeOperator(TokenizeOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"tokenize\"");
-        variablePrintHelper(op.getTokenizeVars(), indent);
-        if (!op.getSecondaryKeyExpressions().isEmpty()) {
-            addIndent(0).append(",\n");
-            pprintExprList(op.getSecondaryKeyExpressions(), indent);
-        }
-        return null;
-    }
-
-    @Override
-    public Void visitForwardOperator(ForwardOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"forward\"");
-        addIndent(0).append(",\n");
-        addIndent(indent).append("\"expressions\": \""
-                + op.getSideDataExpression().getValue().accept(exprVisitor, indent).replace('"', ' ') + "\"");
-        return null;
-    }
-
-    @Override
-    public Void visitSinkOperator(SinkOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"sink\"");
-        return null;
-    }
-
-    @Override
-    public Void visitDelegateOperator(DelegateOperator op, Integer indent) throws AlgebricksException {
-        addIndent(indent).append("\"operator\": \"" + op.toString() + "\"");
-        return null;
-    }
-
-    @Override
-    public Void visitWindowOperator(WindowOperator op, Integer indent) throws AlgebricksException {
-        Integer fldIndent = indent + 2;
-        addIndent(indent).append("\"operator\": \"window-aggregate\"");
-        variablePrintHelper(op.getVariables(), indent);
-        List<Mutable<ILogicalExpression>> expressions = op.getExpressions();
-        if (!expressions.isEmpty()) {
-            buffer.append(",\n");
-            pprintExprList(expressions, indent);
-        }
-        List<Mutable<ILogicalExpression>> partitionExpressions = op.getPartitionExpressions();
-        if (!partitionExpressions.isEmpty()) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"partition-by\": {\n");
-            pprintExprList(partitionExpressions, fldIndent);
-            buffer.append("\n");
-            addIndent(indent).append("}");
-        }
-        List<Pair<OrderOperator.IOrder, Mutable<ILogicalExpression>>> orderExpressions = op.getOrderExpressions();
-        if (!orderExpressions.isEmpty()) {
-            buffer.append(",\n");
-            addIndent(indent).append("\"order-by\": ");
-            pprintOrderExprList(orderExpressions, fldIndent);
-        }
-        if (op.hasNestedPlans()) {
-            List<Pair<OrderOperator.IOrder, Mutable<ILogicalExpression>>> frameValueExpressions =
-                    op.getFrameValueExpressions();
-            if (!frameValueExpressions.isEmpty()) {
-                buffer.append(",\n");
-                addIndent(indent).append("\"frame-on\": ");
-                pprintOrderExprList(frameValueExpressions, fldIndent);
-            }
-            List<Mutable<ILogicalExpression>> frameStartExpressions = op.getFrameStartExpressions();
-            if (!frameStartExpressions.isEmpty()) {
-                buffer.append(",\n");
-                addIndent(indent).append("\"frame-start\": {\n");
-                pprintExprList(frameStartExpressions, fldIndent);
-                buffer.append("\n");
-                addIndent(indent).append("}");
-            }
-            List<Mutable<ILogicalExpression>> frameStartValidationExpressions = op.getFrameStartValidationExpressions();
-            if (!frameStartValidationExpressions.isEmpty()) {
-                buffer.append(",\n");
-                addIndent(indent).append("\"frame-start-if\": {\n");
-                pprintExprList(frameStartValidationExpressions, fldIndent);
-                buffer.append("\n");
-                addIndent(indent).append("}");
-            }
-            List<Mutable<ILogicalExpression>> frameEndExpressions = op.getFrameEndExpressions();
-            if (!frameEndExpressions.isEmpty()) {
-                buffer.append(",\n");
-                addIndent(indent).append("\"frame-end\": {\n");
-                pprintExprList(frameEndExpressions, fldIndent);
-                buffer.append("\n");
-                addIndent(indent).append("}");
-            }
-            List<Mutable<ILogicalExpression>> frameEndValidationExpressions = op.getFrameEndValidationExpressions();
-            if (!frameEndValidationExpressions.isEmpty()) {
-                buffer.append(",\n");
-                addIndent(indent).append("\"frame-end-if\": {\n");
-                pprintExprList(frameEndValidationExpressions, fldIndent);
-                buffer.append("\n");
-                addIndent(indent).append("}");
-            }
-            List<Mutable<ILogicalExpression>> frameExcludeExpressions = op.getFrameExcludeExpressions();
-            if (!frameExcludeExpressions.isEmpty()) {
-                buffer.append(",\n");
-                addIndent(indent).append("\"frame-exclude\": {\n");
-                pprintExprList(frameExcludeExpressions, fldIndent);
-                buffer.append("\n");
-                addIndent(indent).append("},\n");
-                addIndent(indent).append("\"frame-exclude-negation-start\": ")
-                        .append(String.valueOf(op.getFrameExcludeNegationStartIdx()));
-            }
-            Mutable<ILogicalExpression> frameExcludeUnaryExpression = op.getFrameExcludeUnaryExpression();
-            if (frameExcludeUnaryExpression.getValue() != null) {
-                buffer.append(",\n");
-                addIndent(indent).append("\"frame-exclude-unary\": ");
-                pprintExpr(frameExcludeUnaryExpression, fldIndent);
-            }
-            Mutable<ILogicalExpression> frameOffsetExpression = op.getFrameOffsetExpression();
-            if (frameOffsetExpression.getValue() != null) {
-                buffer.append(",\n");
-                addIndent(indent).append("\"frame-offset\": ");
-                pprintExpr(frameOffsetExpression, fldIndent);
-            }
-            int frameMaxObjects = op.getFrameMaxObjects();
-            if (frameMaxObjects != -1) {
-                buffer.append(",\n");
-                addIndent(indent).append("\"frame-max-objects\": ").append(String.valueOf(frameMaxObjects));
-            }
-            buffer.append(",\n");
-            addIndent(indent).append("\"subplan\": ");
-            printNestedPlans(op, fldIndent);
-        }
-        return null;
-    }
-
-    protected void printNestedPlans(AbstractOperatorWithNestedPlans op, Integer indent) throws AlgebricksException {
-        idCounter.nextPrefix();
-        buffer.append("[\n");
-        boolean first = true;
-        for (ILogicalPlan p : op.getNestedPlans()) {
-            if (!first) {
-                buffer.append(",");
-            }
-            printPlan(p, indent + 4);
-            first = false;
-        }
-        addIndent(indent).append("]");
-        idCounter.previousPrefix();
-    }
-
-    protected void pprintExprList(List<Mutable<ILogicalExpression>> expressions, Integer indent)
-            throws AlgebricksException {
-        addIndent(indent);
-        buffer.append("\"expressions\": \"");
-        boolean first = true;
-        for (Mutable<ILogicalExpression> exprRef : expressions) {
-            if (first) {
-                first = false;
-            } else {
-                buffer.append(", ");
-            }
-            pprintExpr(exprRef, indent);
-        }
-        buffer.append("\"");
-    }
-
-    protected void pprintExpr(Mutable<ILogicalExpression> exprRef, Integer indent) throws AlgebricksException {
-        buffer.append(exprRef.getValue().accept(exprVisitor, indent).replace('"', ' '));
-    }
-
-    protected void pprintVeList(List<Pair<LogicalVariable, Mutable<ILogicalExpression>>> vePairList, Integer indent)
-            throws AlgebricksException {
-        buffer.append("[");
-        boolean first = true;
-        for (Pair<LogicalVariable, Mutable<ILogicalExpression>> ve : vePairList) {
-            if (first) {
-                first = false;
-            } else {
-                buffer.append(",");
-            }
-            if (ve.first != null) {
-                buffer.append("{\"variable\": \"" + ve.first.toString().replace('"', ' ') + "\"," + "\"expression\": \""
-                        + ve.second.toString().replace('"', ' ') + "\"}");
-            } else {
-                buffer.append("{\"expression\": \"" + ve.second.getValue().accept(exprVisitor, indent).replace('"', ' ')
-                        + "\"}");
-            }
-        }
-        buffer.append("]");
-    }
-
-    private void pprintOrderExprList(List<Pair<OrderOperator.IOrder, Mutable<ILogicalExpression>>> orderExpressions,
-            Integer indent) throws AlgebricksException {
-        buffer.append("[");
-        boolean first = true;
-        for (Pair<OrderOperator.IOrder, Mutable<ILogicalExpression>> p : orderExpressions) {
-            if (first) {
-                first = false;
-            } else {
-                buffer.append(",");
-            }
-            buffer.append("{\"order\": \"" + getOrderString(p.first, indent) + "\"," + "\"expression\": \""
-                    + p.second.getValue().accept(exprVisitor, indent).replace('"', ' ') + "\"}");
-        }
-        buffer.append("]");
-    }
-
-    private String getOrderString(OrderOperator.IOrder order, Integer indent) throws AlgebricksException {
+    private String getOrderString(OrderOperator.IOrder order, Void indent) throws AlgebricksException {
         switch (order.getKind()) {
             case ASC:
                 return "ASC";
             case DESC:
                 return "DESC";
             default:
-                return order.getExpressionRef().getValue().accept(exprVisitor, indent).replace('"', ' ');
+                return order.getExpressionRef().getValue().accept(exprVisitor, indent);
+        }
+    }
+
+    /////////////// string fields ///////////////
+    /** Writes "fieldName": "expr" */
+    private void writeStringFieldExpression(String fieldName, Mutable<ILogicalExpression> expression, Void indent)
+            throws AlgebricksException, IOException {
+        jsonGenerator.writeStringField(fieldName, expression.getValue().accept(exprVisitor, indent));
+    }
+
+    /////////////// array fields ///////////////
+    /** Writes "fieldName": [ "var1", "var2", ... ] */
+    private void writeArrayFieldOfVariables(String fieldName, List<LogicalVariable> variables) throws IOException {
+        jsonGenerator.writeArrayFieldStart(fieldName);
+        for (int i = 0, size = variables.size(); i < size; i++) {
+            jsonGenerator.writeString(String.valueOf(variables.get(i)));
+        }
+        jsonGenerator.writeEndArray();
+    }
+
+    /** Writes "fieldName": [ ["var1", "var2", ...], ["var1", "var2", ...] ] */
+    private void writeArrayFieldOfNestedVariablesList(String fieldName, List<List<LogicalVariable>> nestedVarList)
+            throws IOException {
+        jsonGenerator.writeArrayFieldStart(fieldName);
+        for (int i = 0, size = nestedVarList.size(); i < size; i++) {
+            List<LogicalVariable> nextList = nestedVarList.get(i);
+            for (int k = 0, varSize = nextList.size(); k < varSize; k++) {
+                jsonGenerator.writeString(String.valueOf(nextList.get(k)));
+            }
+        }
+        jsonGenerator.writeEndArray();
+    }
+
+    /** Writes "fieldName" : [ "expr" ] */
+    private void writeArrayFieldOfExpression(String fieldName, Mutable<ILogicalExpression> expr, Void indent)
+            throws IOException, AlgebricksException {
+        jsonGenerator.writeArrayFieldStart(fieldName);
+        jsonGenerator.writeString(expr.getValue().accept(exprVisitor, indent));
+        jsonGenerator.writeEndArray();
+    }
+
+    /** Writes "fieldName" : [ "expr1", "expr2", ...] */
+    private void writeArrayFieldOfExpressions(String fieldName, List<Mutable<ILogicalExpression>> exprs, Void indent)
+            throws IOException, AlgebricksException {
+        jsonGenerator.writeArrayFieldStart(fieldName);
+        for (int i = 0, size = exprs.size(); i < size; i++) {
+            jsonGenerator.writeString(exprs.get(i).getValue().accept(exprVisitor, indent));
+        }
+        jsonGenerator.writeEndArray();
+    }
+
+    /** Writes "fieldName" : [ { "variable": "var1", "expression": "expr1" }, ... ] */
+    private void writeArrayFieldOfVariableExpressionPairs(String fieldName,
+            List<Pair<LogicalVariable, Mutable<ILogicalExpression>>> varExprPairs, Void indent)
+            throws AlgebricksException, IOException {
+        jsonGenerator.writeArrayFieldStart(fieldName);
+        for (Pair<LogicalVariable, Mutable<ILogicalExpression>> ve : varExprPairs) {
+            jsonGenerator.writeStartObject();
+            if (ve.first != null) {
+                jsonGenerator.writeStringField("variable", ve.first.toString());
+            }
+            writeStringFieldExpression(EXPRESSION_FIELD, ve.second, indent);
+            jsonGenerator.writeEndObject();
+        }
+        jsonGenerator.writeEndArray();
+    }
+
+    /** Writes "fieldName" : [ { "order": "", "expression": "" }, ... ] */
+    private void writeArrayFieldOfOrderExprList(String fieldName,
+            List<Pair<OrderOperator.IOrder, Mutable<ILogicalExpression>>> orderList, Void indent)
+            throws AlgebricksException, IOException {
+        jsonGenerator.writeArrayFieldStart(fieldName);
+        for (Pair<OrderOperator.IOrder, Mutable<ILogicalExpression>> p : orderList) {
+            jsonGenerator.writeStartObject();
+            jsonGenerator.writeStringField("order", getOrderString(p.first, indent));
+            writeStringFieldExpression(EXPRESSION_FIELD, p.second, indent);
+            jsonGenerator.writeEndObject();
+        }
+        jsonGenerator.writeEndArray();
+    }
+
+    /////////////// object fields ///////////////
+    /** Writes "fieldName" : { "expressions": [ "expr1", "expr2", ...] } */
+    private void writeObjectFieldWithExpressions(String fieldName, List<Mutable<ILogicalExpression>> exprs, Void indent)
+            throws IOException, AlgebricksException {
+        jsonGenerator.writeObjectFieldStart(fieldName);
+        writeArrayFieldOfExpressions(EXPRESSIONS_FIELD, exprs, indent);
+        jsonGenerator.writeEndObject();
+    }
+
+    private void flushContentToWriter() throws AlgebricksException {
+        try {
+            jsonGenerator.flush();
+        } catch (IOException e) {
+            throw new AlgebricksException(e, ErrorCode.ERROR_PRINTING_PLAN);
         }
     }
 }
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/PlanPrettyPrinter.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/PlanPrettyPrinter.java
index 67640c5..9b3a6d8 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/PlanPrettyPrinter.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/PlanPrettyPrinter.java
@@ -20,20 +20,20 @@
 
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.base.ILogicalPlan;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
 
 public class PlanPrettyPrinter {
-    public static void printOperator(AbstractLogicalOperator op, AbstractLogicalOperatorPrettyPrintVisitor pvisitor,
-            int indent) throws AlgebricksException {
-        pvisitor.printOperator(op, indent);
+
+    public static final int INIT_SIZE = 256;
+
+    public static IPlanPrettyPrinter createJsonPlanPrettyPrinter() {
+        return new LogicalOperatorPrettyPrintVisitorJson();
     }
 
-    public static void printPlan(ILogicalPlan plan, AbstractLogicalOperatorPrettyPrintVisitor pvisitor, int indent)
-            throws AlgebricksException {
-        pvisitor.printPlan(plan, indent);
+    public static IPlanPrettyPrinter createStringPlanPrettyPrinter() {
+        return new LogicalOperatorPrettyPrintVisitor();
     }
 
-    public static void printPhysicalOps(ILogicalPlan plan, AlgebricksAppendable out, int indent)
+    public static void printPhysicalOps(ILogicalPlan plan, AlgebricksStringBuilderWriter out, int indent)
             throws AlgebricksException {
         AbstractLogicalOperatorPrettyPrintVisitor.printPhysicalOps(plan, out, indent);
     }
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/AbstractRuleController.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/AbstractRuleController.java
index 6ed4c9e..96e15da 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/AbstractRuleController.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/AbstractRuleController.java
@@ -27,9 +27,7 @@
 import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
 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.prettyprint.AbstractLogicalOperatorPrettyPrintVisitor;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.AlgebricksAppendable;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.PlanPrettyPrinter;
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.IPlanPrettyPrinter;
 import org.apache.hyracks.algebricks.core.config.AlgebricksConfig;
 import org.apache.hyracks.util.LogRedactionUtil;
 
@@ -67,10 +65,8 @@
 
     private String getPlanString(Mutable<ILogicalOperator> opRef) throws AlgebricksException {
         if (AlgebricksConfig.ALGEBRICKS_LOGGER.isTraceEnabled() && context != null) {
-            AbstractLogicalOperatorPrettyPrintVisitor pvisitor = context.getPrettyPrintVisitor();
-            pvisitor.reset(new AlgebricksAppendable());
-            PlanPrettyPrinter.printOperator((AbstractLogicalOperator) opRef.getValue(), pvisitor, 0);
-            return pvisitor.get().toString();
+            IPlanPrettyPrinter prettyPrinter = context.getPrettyPrinter();
+            return prettyPrinter.reset().printOperator((AbstractLogicalOperator) opRef.getValue()).toString();
         }
         return null;
     }
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/AlgebricksOptimizationContext.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/AlgebricksOptimizationContext.java
index eb398e6..0fa0023 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/AlgebricksOptimizationContext.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/AlgebricksOptimizationContext.java
@@ -38,7 +38,7 @@
 import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableEvalSizeEnvironment;
 import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
 import org.apache.hyracks.algebricks.core.algebra.metadata.IMetadataProvider;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.AbstractLogicalOperatorPrettyPrintVisitor;
+import org.apache.hyracks.algebricks.core.algebra.prettyprint.IPlanPrettyPrinter;
 import org.apache.hyracks.algebricks.core.algebra.properties.DefaultNodeGroupDomain;
 import org.apache.hyracks.algebricks.core.algebra.properties.FunctionalDependency;
 import org.apache.hyracks.algebricks.core.algebra.properties.ILogicalPropertiesVector;
@@ -86,7 +86,7 @@
     private final IExpressionTypeComputer expressionTypeComputer;
     private final IMissableTypeComputer nullableTypeComputer;
     private final INodeDomain defaultNodeDomain;
-    private final AbstractLogicalOperatorPrettyPrintVisitor prettyPrintVisitor;
+    private final IPlanPrettyPrinter prettyPrinter;
     private final IConflictingTypeResolver conflictingTypeResovler;
     private final IWarningCollector warningCollector;
 
@@ -94,8 +94,8 @@
             IMergeAggregationExpressionFactory mergeAggregationExpressionFactory,
             IExpressionTypeComputer expressionTypeComputer, IMissableTypeComputer nullableTypeComputer,
             IConflictingTypeResolver conflictingTypeResovler, PhysicalOptimizationConfig physicalOptimizationConfig,
-            AlgebricksPartitionConstraint clusterLocations,
-            AbstractLogicalOperatorPrettyPrintVisitor prettyPrintVisitor, IWarningCollector warningCollector) {
+            AlgebricksPartitionConstraint clusterLocations, IPlanPrettyPrinter prettyPrinter,
+            IWarningCollector warningCollector) {
         this.varCounter = varCounter;
         this.expressionEvalSizeComputer = expressionEvalSizeComputer;
         this.mergeAggregationExpressionFactory = mergeAggregationExpressionFactory;
@@ -103,7 +103,7 @@
         this.nullableTypeComputer = nullableTypeComputer;
         this.physicalOptimizationConfig = physicalOptimizationConfig;
         this.defaultNodeDomain = new DefaultNodeGroupDomain(clusterLocations);
-        this.prettyPrintVisitor = prettyPrintVisitor;
+        this.prettyPrinter = prettyPrinter;
         this.conflictingTypeResovler = conflictingTypeResovler;
         this.warningCollector = warningCollector;
     }
@@ -324,8 +324,8 @@
     }
 
     @Override
-    public AbstractLogicalOperatorPrettyPrintVisitor getPrettyPrintVisitor() {
-        return prettyPrintVisitor;
+    public IPlanPrettyPrinter getPrettyPrinter() {
+        return prettyPrinter;
     }
 
     @Override
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java
index 426e9fb..25d53b0 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/rewriter/base/HeuristicOptimizer.java
@@ -29,9 +29,6 @@
 import org.apache.hyracks.algebricks.core.algebra.base.PhysicalOperatorTag;
 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.prettyprint.AbstractLogicalOperatorPrettyPrintVisitor;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.AlgebricksAppendable;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.PlanPrettyPrinter;
 import org.apache.hyracks.algebricks.core.config.AlgebricksConfig;
 import org.apache.hyracks.util.LogRedactionUtil;
 import org.apache.logging.log4j.Level;
@@ -69,11 +66,8 @@
 
     private void logPlanAt(String name, Level lvl) throws AlgebricksException {
         if (AlgebricksConfig.ALGEBRICKS_LOGGER.isEnabled(lvl)) {
-            final AbstractLogicalOperatorPrettyPrintVisitor pvisitor = context.getPrettyPrintVisitor();
-            pvisitor.reset(new AlgebricksAppendable());
-            PlanPrettyPrinter.printPlan(plan, pvisitor, 0);
-            AlgebricksConfig.ALGEBRICKS_LOGGER.log(lvl,
-                    name + ":\n" + LogRedactionUtil.userData(pvisitor.get().toString()));
+            String planStr = context.getPrettyPrinter().reset().printPlan(plan).toString();
+            AlgebricksConfig.ALGEBRICKS_LOGGER.log(lvl, name + ":\n" + LogRedactionUtil.userData(planStr));
         }
     }
 
diff --git a/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java b/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
index e174ab8..9236545 100644
--- a/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
+++ b/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
@@ -76,8 +76,6 @@
 import org.apache.hyracks.algebricks.core.algebra.operators.physical.SortForwardPOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.physical.SortMergeExchangePOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.physical.StableSortPOperator;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.LogicalOperatorPrettyPrintVisitor;
-import org.apache.hyracks.algebricks.core.algebra.prettyprint.PlanPrettyPrinter;
 import org.apache.hyracks.algebricks.core.algebra.properties.FunctionalDependency;
 import org.apache.hyracks.algebricks.core.algebra.properties.ILocalStructuralProperty;
 import org.apache.hyracks.algebricks.core.algebra.properties.ILocalStructuralProperty.PropertyType;
@@ -283,7 +281,7 @@
             if (loggerTraceEnabled) {
                 AlgebricksConfig.ALGEBRICKS_LOGGER
                         .trace(">>>> Removing redundant SORT operator " + op.getPhysicalOperator() + "\n");
-                printOp(op);
+                printOp(op, context);
             }
             changed = true;
             AbstractLogicalOperator nextOp = (AbstractLogicalOperator) op.getInputs().get(0).getValue();
@@ -528,7 +526,7 @@
         op.getInputs().set(i, topOp);
         OperatorPropertiesUtil.computeSchemaAndPropertiesRecIfNull((AbstractLogicalOperator) topOp.getValue(), context);
         OperatorManipulationUtil.setOperatorMode(op);
-        printOp((AbstractLogicalOperator) topOp.getValue());
+        printOp((AbstractLogicalOperator) topOp.getValue(), context);
     }
 
     private Mutable<ILogicalOperator> enforceOrderProperties(List<LocalOrderProperty> oList,
@@ -616,7 +614,7 @@
             if (AlgebricksConfig.ALGEBRICKS_LOGGER.isTraceEnabled()) {
                 AlgebricksConfig.ALGEBRICKS_LOGGER
                         .trace(">>>> Added partitioning enforcer " + exchg.getPhysicalOperator() + ".\n");
-                printOp((AbstractLogicalOperator) op);
+                printOp((AbstractLogicalOperator) op, context);
             }
         }
     }
@@ -875,11 +873,10 @@
         return !childLocalProperties.isEmpty();
     }
 
-    private void printOp(AbstractLogicalOperator op) throws AlgebricksException {
-        LogicalOperatorPrettyPrintVisitor pvisitor = new LogicalOperatorPrettyPrintVisitor();
-        PlanPrettyPrinter.printOperator(op, pvisitor, 0);
+    private void printOp(AbstractLogicalOperator op, IOptimizationContext ctx) throws AlgebricksException {
         if (AlgebricksConfig.ALGEBRICKS_LOGGER.isTraceEnabled()) {
-            AlgebricksConfig.ALGEBRICKS_LOGGER.trace(LogRedactionUtil.userData(pvisitor.get().toString()));
+            String plan = ctx.getPrettyPrinter().reset().printOperator(op).toString();
+            AlgebricksConfig.ALGEBRICKS_LOGGER.trace(LogRedactionUtil.userData(plan));
         }
     }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java
index 8aebccf..5716262 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java
@@ -155,6 +155,7 @@
     public static final int UNSUPPORTED_WINDOW_SPEC = 119;
     public static final int EOF = 120;
     public static final int NUMERIC_PROMOTION_ERROR = 121;
+    public static final int ERROR_PRINTING_PLAN = 122;
 
     // Compilation error codes.
     public static final int RULECOLLECTION_NOT_INSTANCE_OF_LIST = 10000;
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties b/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties
index 68a6418..fcbb6bb 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties
@@ -138,6 +138,7 @@
 119 = Unsupported window specification: PARTITION BY %1$s, ORDER BY %2$s
 120 = End of file
 121 = A numeric type promotion error has occurred: %1$s
+122 = Encountered an error while printing the plan
 
 10000 = The given rule collection %1$s is not an instance of the List class.
 10001 = Cannot compose partition constraint %1$s with %2$s