DRILL-1062: Implemented null ordering (NULLS FIRST/NULLS LAST).

Primary:
- Split "compare_to" function templates (for sorting) into
  "compare_to_nulls_high" and "compare_to_nulls_low" versions.
- Added tests to verify ORDER BY ordering.
- Added tests to verify merge join order correctness.
- Implemented java.sql.DatabaseMetaData.nullsAreSortedHigh(), etc.
Secondary:
- Eliminated DateInterfaceFunctions.java template (merged into other).
- Renamed comparison-related template data objects and file names.
- Eliminated unused template macros, function template classes.
- Overhauled Order.Ordering; added unit test.
- Regularized some generated-class names.
Miscellaneous:
- Added toString() to ExpressionPosition, Order.Ordering, JoinStatus.
- Fixed some typos.
- Fixed some comment syntax.

43 files changed, 2867 insertions, 1810 deletions

diff --git a/exec/java-exec/src/main/codegen/config.fmpp b/exec/java-exec/src/main/codegen/config.fmpp
index aff6240c0..5dc7360d4 100644
--- a/exec/java-exec/src/main/codegen/config.fmpp
+++ b/exec/java-exec/src/main/codegen/config.fmpp
@@ -15,29 +15,34 @@
 # limitations under the License.
 
 data: {
-    vv: tdd(../data/ValueVectorTypes.tdd),
-    compare: tdd(../data/CompareTypes.tdd),
-    cast: tdd(../data/Casts.tdd),
-    MathFunctionTypes: tdd(../data/MathFunctionTypes.tdd),
-    mathFunc:tdd(../data/MathFunc.tdd),
-    aggrtypes1: tdd(../data/AggrTypes1.tdd),
-    decimalaggrtypes1: tdd(../data/DecimalAggrTypes1.tdd),
-    decimalaggrtypes2: tdd(../data/DecimalAggrTypes2.tdd),
-    aggrtypes2: tdd(../data/AggrTypes2.tdd),
-    aggrtypes3: tdd(../data/AggrTypes3.tdd),
-    covarTypes: tdd(../data/CovarTypes.tdd),
-    corrTypes: tdd(../data/CorrelationTypes.tdd),
-    logicalTypes: tdd(../data/AggrBitwiseLogicalTypes.tdd),
-    date: tdd(../data/DateTypes.tdd),
-    extract: tdd(../data/ExtractTypes.tdd),
-    parser: tdd(../data/Parser.tdd),
-    decimal: tdd(../data/DecimalTypes.tdd),
-    dateIntervalFunc: tdd(../data/DateIntervalFunc.tdd),
-    intervalNumericTypes: tdd(../data/IntervalNumericTypes.tdd),
-    extract: tdd(../data/ExtractTypes.tdd),
-    sumzero: tdd(../data/SumZero.tdd),
-    numericTypes: tdd(../data/NumericTypes.tdd),
-    casthigh: tdd(../data/CastHigh.tdd)
+    vv:                       tdd(../data/ValueVectorTypes.tdd),
+
+    # Most types for comparison functions (for templates/ComparisonFunctions.java).
+    comparisonTypesMain:      tdd(../data/ComparisonTypesMain.tdd),
+
+    # Decimal types for comparison function (for
+    # templates/DecimalFunctions/ComparisonFunctions.java).
+    comparisonTypesDecimal:   tdd(../data/ComparisonTypesDecimal.tdd),
+
+    cast:                     tdd(../data/Casts.tdd),
+    MathFunctionTypes:        tdd(../data/MathFunctionTypes.tdd),
+    mathFunc:                 tdd(../data/MathFunc.tdd),
+    aggrtypes1:               tdd(../data/AggrTypes1.tdd),
+    decimalaggrtypes1:        tdd(../data/DecimalAggrTypes1.tdd),
+    decimalaggrtypes2:        tdd(../data/DecimalAggrTypes2.tdd),
+    aggrtypes2:               tdd(../data/AggrTypes2.tdd),
+    aggrtypes3:               tdd(../data/AggrTypes3.tdd),
+    covarTypes:               tdd(../data/CovarTypes.tdd),
+    corrTypes:                tdd(../data/CorrelationTypes.tdd),
+    logicalTypes:             tdd(../data/AggrBitwiseLogicalTypes.tdd),
+    extract:                  tdd(../data/ExtractTypes.tdd),
+    parser:                   tdd(../data/Parser.tdd),
+    dateIntervalFunc:         tdd(../data/DateIntervalFunc.tdd),
+    intervalNumericTypes:     tdd(../data/IntervalNumericTypes.tdd),
+    extract:                  tdd(../data/ExtractTypes.tdd),
+    sumzero:                  tdd(../data/SumZero.tdd),
+    numericTypes:             tdd(../data/NumericTypes.tdd),
+    casthigh:                 tdd(../data/CastHigh.tdd)
 }
 freemarkerLinks: {
     includes: includes/
diff --git a/exec/java-exec/src/main/codegen/data/CompareTypes.tdd b/exec/java-exec/src/main/codegen/data/CompareTypes.tdd
deleted file mode 100644
index f384d5299..000000000
--- a/exec/java-exec/src/main/codegen/data/CompareTypes.tdd
+++ /dev/null
@@ -1,21 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-# http:# www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-{
-  types: [
-    {comparables: ["Float4", "Float8", "Int", "BigInt", "NullableFloat4", "NullableFloat8", "NullableInt", "NullableBigInt"], mode: "fixed"}, 
-    {comparables: ["VarChar", "VarBinary", "NullableVarChar", "NullableVarBinary"], mode: "var"} 
-  ]
-} 
diff --git a/exec/java-exec/src/main/codegen/data/DecimalTypes.tdd b/exec/java-exec/src/main/codegen/data/ComparisonTypesDecimal.tdd
index 423fe895d..423fe895d 100644
--- a/exec/java-exec/src/main/codegen/data/DecimalTypes.tdd
+++ b/exec/java-exec/src/main/codegen/data/ComparisonTypesDecimal.tdd
diff --git a/exec/java-exec/src/main/codegen/data/ComparisonTypesMain.tdd b/exec/java-exec/src/main/codegen/data/ComparisonTypesMain.tdd
new file mode 100644
index 000000000..c4d924c31
--- /dev/null
+++ b/exec/java-exec/src/main/codegen/data/ComparisonTypesMain.tdd
@@ -0,0 +1,69 @@
+# 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.
+
+# Template data/configuration file for ComparisonFunctions.java for
+# comparison functions--currently, for primitive numeric types, for Bit/Boolean,
+# and for character and byte strings:
+
+# Each group (value in "typeGroups" array is for types that can be compared to each
+# other:
+# - "comparables" is types in group
+# - "mode" is comparison mode (case of code needed to compare):
+#   - "primitive" - comparable with Java primitive-type "<", "==", and ">" operators
+#   - "varString" - comparable with ByteFunctionHelpers.compare(...)
+
+{
+  typeGroups: [
+
+    # Group:  Drill "Bit"/SQL BOOLEAN (represented with Java primitive type)
+    {comparables: [
+      "Bit"
+      ],
+      mode: "primitive"},
+
+    # Group:  Numeric types cross-comparable with Java primitive-type operators:
+    {comparables: [
+      "Int",
+      "BigInt",
+      "Float4",
+      "Float8"
+      ],
+      mode: "primitive"},
+
+
+    # Date/time types other than IntervalDay or Interval (comparable with Java
+    # primitive-type operators):
+
+    {comparables: ["TimeStamp"], mode: "primitive"},
+    {comparables: ["Date"], mode: "primitive"},
+    {comparables: ["Time"], mode: "primitive"},
+    {comparables: ["TimeStampTZ"], mode: "primitive"},
+    {comparables: ["IntervalYear"], mode: "primitive"},
+
+    {comparables: ["IntervalDay"], mode: "intervalDay"},
+    {comparables: ["Interval"], mode: "intervalNameThis"},
+
+
+    # Group:  Variable-length string types whose values can be compared using
+    # ByteFunctionHelpers.compare(...)
+    {comparables: [
+      "VarChar",
+      "VarBinary"
+      ],
+      mode: "varString"}
+
+  ]
+}
diff --git a/exec/java-exec/src/main/codegen/data/DateTypes.tdd b/exec/java-exec/src/main/codegen/data/DateTypes.tdd
deleted file mode 100644
index 5802df0bb..000000000
--- a/exec/java-exec/src/main/codegen/data/DateTypes.tdd
+++ /dev/null
@@ -1,27 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-# http:# www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-{
-  dateTypes: [
-    {name: "Date"},
-    {name: "TimeStamp"},
-    {name: "TimeStampTZ"},
-    {name: "Time"},
-    {name: "Interval"},
-    {name: "IntervalYear"},
-    {name: "IntervalDay"}
-  ]
-}
diff --git a/exec/java-exec/src/main/codegen/templates/ComparisonFunctions.java b/exec/java-exec/src/main/codegen/templates/ComparisonFunctions.java
index 628277c95..011d4d940 100644
--- a/exec/java-exec/src/main/codegen/templates/ComparisonFunctions.java
+++ b/exec/java-exec/src/main/codegen/templates/ComparisonFunctions.java
@@ -15,60 +15,143 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
+
+/**
+  NOTE: ComparisonFunctions.java does not contain/generate all comparison
+  functions.  DecimalFunctions.java and DateIntervalFunctions.java contain
+  some.
+*/
+<#-- TODO:  Refactor comparison code from DecimalFunctions.java and
+     DateIntervalFunctions.java into here (to eliminate duplicate template code
+     and so that ComparisonFunctions actually has all comparison functions. -->
+
+
 <@pp.dropOutputFile />
 
-<#macro compareBlock mode left right output nullCompare>
+<#macro intervalCompareBlock leftType rightType leftMonths leftDays leftMillis rightMonths rightDays rightMillis output>
+
+        org.joda.time.MutableDateTime leftDate  =
+            new org.joda.time.MutableDateTime(1970, 1, 1, 0, 0, 0, 0, org.joda.time.DateTimeZone.UTC);
+        org.joda.time.MutableDateTime rightDate =
+            new org.joda.time.MutableDateTime(1970, 1, 1, 0, 0, 0, 0, org.joda.time.DateTimeZone.UTC);
+
+        // Left and right date have the same starting point (epoch), add the interval period and compare the two
+        leftDate.addMonths(${leftMonths});
+        leftDate.addDays(${leftDays});
+        leftDate.add(${leftMillis});
+
+        rightDate.addMonths(${rightMonths});
+        rightDate.addDays(${rightDays});
+        rightDate.add(${rightMillis});
 
-outside:{
-  
+        long leftMS  = leftDate.getMillis();
+        long rightMS = rightDate.getMillis();
+
+        ${output} = leftMS < rightMS ? -1 : (leftMS > rightMS ? 1 : 0);
+
+</#macro>
+
+
+<#macro compareNullsSubblock leftType rightType output breakTarget nullCompare nullComparesHigh>
   <#if nullCompare>
-      <#if left?starts_with("Nullable")>
-        <#if right?starts_with("Nullable")>
+    <#if nullComparesHigh>
+      <#assign leftNullResult = 1> <#-- if only L is null and nulls are high, then "L > R" (1) -->
+      <#assign rightNullResult = -1>
+    <#else>
+      <#assign leftNullResult = -1> <#-- if only L is null and nulls are low, then "L < R" (-1) -->
+      <#assign rightNullResult = 1>
+    </#if>
+
+    <#if leftType?starts_with("Nullable")>
+      <#if rightType?starts_with("Nullable")>
         <#-- Both are nullable. -->
-        if(left.isSet == 0){
-          if(right.isSet == 0){
+        if ( left.isSet == 0 ) {
+          if ( right.isSet == 0 ) {
+            <#-- Both are null--result is "L = R". -->
             ${output} = 0;
-            break outside;
-          }else{
-            ${output} = 1;
-            break outside;
+            break ${breakTarget};
+          } else {
+            <#-- Only left is null--result is "L < R" or "L > R" per null ordering. -->
+            ${output} = ${leftNullResult};
+            break ${breakTarget};
           }
-        }else if(right.isSet == 0){
-          ${output} = -1;
-          break outside;
+        } else if ( right.isSet == 0 ) {
+          <#-- Only right is null--result is "L > R" or "L < R" per null ordering. -->
+          ${output} = ${rightNullResult};
+          break ${breakTarget};
         }
-        <#else>
+      <#else>
         <#-- Left is nullable but right is not. -->
-        if(left.isSet == 0){
-          ${output} = 1;
-          break outside;
+        if ( left.isSet == 0 ) {
+          <#-- Only left is null--result is "L < R" or "L > R" per null ordering. -->
+          ${output} = ${leftNullResult};
+          break ${breakTarget};
         }
-        </#if>
-    <#elseif right?starts_with("Nullable")>
-      if(right.isSet == 0){
-        ${output} = -1;
-        break outside;
-      }
       </#if>
+    <#elseif rightType?starts_with("Nullable")>
+      <#-- Left is not nullable but right is. -->
+      if ( right.isSet == 0 ) {
+        <#-- Only right is null--result is "L > R" or "L < R" per null ordering. -->
+        ${output} = ${rightNullResult};
+        break ${breakTarget};
+      }
     </#if>
-    
-    <#if mode == "var">
-      ${output} = org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.compare(left.buffer, left.start, left.end, right.buffer, right.start, right.end);
-    <#elseif mode == "fixed">
-      ${output} = left.value < right.value ? -1 : ((left.value == right.value)? 0 : 1);
-    </#if>    
-  
+  </#if>
 
-  
-  
+</#macro>
 
-}
+
+<#-- macro compareBlock: Generates block handling comparison, including NULL. -->
+
+<#-- Parameters: >
+<#-- - mode: selects case of comparison code -->
+<#-- - leftType: name of left argument's type  (e.g., NullableFloat4)  -->
+<#-- - rightType: name of right argument's type  -->
+<#-- - output: output variable name -->
+<#-- - nullCompare: whether to generate null-comparison code -->
+<#-- - nullComparesHigh:  whether NULL compares as the highest value or the lowest
+       value -->
+
+<#macro compareBlock mode leftType rightType output nullCompare nullComparesHigh>
+     outside:
+      {
+        <@compareNullsSubblock leftType=leftType rightType=rightType
+          output="out.value" breakTarget="outside" nullCompare=true nullComparesHigh=nullComparesHigh />
+
+    <#if mode == "primitive">
+      ${output} = left.value < right.value ? -1 : (left.value == right.value ? 0 : 1);
+    <#elseif mode == "varString">
+      ${output} = org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.compare(
+          left.buffer, left.start, left.end, right.buffer, right.start, right.end );
+    <#elseif mode == "intervalNameThis">
+      <@intervalCompareBlock leftType=leftType rightType=rightType
+        leftMonths ="left.months"  leftDays ="left.days"  leftMillis ="left.milliseconds"
+        rightMonths="right.months" rightDays="right.days" rightMillis="right.milliseconds"
+        output="${output}"/>
+    <#elseif mode == "intervalDay">
+      <@intervalCompareBlock leftType=leftType rightType=rightType
+        leftMonths ="0" leftDays ="left.days"  leftMillis ="left.milliseconds"
+        rightMonths="0" rightDays="right.days" rightMillis="right.milliseconds"
+        output="${output}"/>
+       <#-- TODO:  Refactor other comparison code to here. -->
+    <#else>
+      ${mode_HAS_BAD_VALUE}
+    </#if>
+
+      } // outside
 </#macro>
 
-<#list compare.types as type>
-<#list type.comparables as left>
-<#list type.comparables as right>
-<@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/GCompare${left}${right}.java" />
+
+<#-- 1.  For each group of cross-comparable types: -->
+<#list comparisonTypesMain.typeGroups as typeGroup>
+
+<#-- 2.  For each pair of (cross-comparable) types in group: -->
+<#list typeGroup.comparables as leftTypeBase>
+<#list typeGroup.comparables as rightTypeBase>
+
+<#-- Generate one file for each pair of base types (includes Nullable cases). -->
+<@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/GCompare${leftTypeBase}Vs${rightTypeBase}.java" />
 
 <#include "/@includes/license.ftl" />
 
@@ -79,6 +162,7 @@ import org.apache.drill.exec.expr.annotations.FunctionTemplate;
 import org.apache.drill.exec.expr.annotations.FunctionTemplate.NullHandling;
 import org.apache.drill.exec.expr.annotations.Output;
 import org.apache.drill.exec.expr.annotations.Param;
+import org.apache.drill.exec.expr.fn.FunctionGenerationHelper;
 import org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers;
 import org.apache.drill.exec.expr.holders.*;
 import org.apache.drill.exec.record.RecordBatch;
@@ -86,157 +170,249 @@ import javax.inject.Inject;
 import io.netty.buffer.DrillBuf;
 
 @SuppressWarnings("unused")
-public class GCompare${left}${right}{
+public class GCompare${leftTypeBase}Vs${rightTypeBase} {
 
-  @FunctionTemplate(name = "compare_to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-  public static class Compare${left}${right} implements DrillSimpleFunc {
+<#-- 3.  For each combination of Nullable vs. not (of given non-nullable types):  -->
+<#list ["${leftTypeBase}", "Nullable${leftTypeBase}"] as leftType >
+<#list ["${rightTypeBase}", "Nullable${rightTypeBase}"] as rightType >
 
-      @Param ${left}Holder left;
-      @Param ${right}Holder right;
-      @Output IntHolder out;
 
-      public void setup(RecordBatch b) {}
+  <#-- Comparison function for sorting and grouping relational operators
+       (not for comparison expression operators (=, <, etc.)). -->
+  @FunctionTemplate(name = FunctionGenerationHelper.COMPARE_TO_NULLS_HIGH,
+                    scope = FunctionTemplate.FunctionScope.SIMPLE,
+                    nulls = NullHandling.INTERNAL)
+  public static class GCompare${leftType}Vs${rightType}NullHigh implements DrillSimpleFunc {
 
-      public void eval() {
-        <@compareBlock mode=type.mode left=left right=right output="out.value" nullCompare=true />
-      }
+    @Param ${leftType}Holder left;
+    @Param ${rightType}Holder right;
+    @Output IntHolder out;
+
+    public void setup(RecordBatch b) {}
+
+    public void eval() {
+      <@compareBlock mode=typeGroup.mode leftType=leftType rightType=rightType
+                     output="out.value" nullCompare=true nullComparesHigh=true />
+    }
+  }
+
+  <#-- Comparison function for sorting and grouping relational operators
+       (not for comparison expression operators (=, <, etc.)). -->
+  @FunctionTemplate(name = FunctionGenerationHelper.COMPARE_TO_NULLS_LOW,
+                    scope = FunctionTemplate.FunctionScope.SIMPLE,
+                    nulls = NullHandling.INTERNAL)
+  public static class GCompare${leftType}Vs${rightType}NullLow implements DrillSimpleFunc {
+
+    @Param ${leftType}Holder left;
+    @Param ${rightType}Holder right;
+    @Output IntHolder out;
+
+    public void setup(RecordBatch b) {}
+
+    public void eval() {
+      <@compareBlock mode=typeGroup.mode leftType=leftType rightType=rightType
+                     output="out.value" nullCompare=true nullComparesHigh=false />
+    }
   }
 
-  <#if ! left?starts_with("Nullable")  &&  ! right?starts_with("Nullable") >
+</#list>
+</#list> <#-- 3. Nullable combinations -->
+
 
-  @FunctionTemplate(names = {"less_than", "<"}, scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class LessThan${left}${right} implements DrillSimpleFunc {
 
-      @Param ${left}Holder left;
-      @Param ${right}Holder right;
+  <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+       not for sorting and grouping relational operators. -->
+  @FunctionTemplate(names = {"less_than", "<"},
+                    scope = FunctionTemplate.FunctionScope.SIMPLE,
+                    nulls = NullHandling.NULL_IF_NULL)
+  public static class LessThan${leftTypeBase}Vs${rightTypeBase} implements DrillSimpleFunc {
+
+      @Param ${leftTypeBase}Holder left;
+      @Param ${rightTypeBase}Holder right;
       @Output BitHolder out;
 
       public void setup(RecordBatch b) {}
 
       public void eval() {
-        
-        <#if type.mode == "var" >
-        int cmp;
-        <@compareBlock mode=type.mode left=left right=right output="cmp" nullCompare=false/>
-        out.value = cmp == -1 ? 1 : 0;
+
+        <#if typeGroup.mode == "primitive">
+          out.value = left.value < right.value ? 1 : 0;
+        <#elseif typeGroup.mode == "varString"
+            || typeGroup.mode == "intervalNameThis" || typeGroup.mode == "intervalDay" >
+          int cmp;
+          <@compareBlock mode=typeGroup.mode leftType=leftTypeBase rightType=rightTypeBase
+                         output="cmp" nullCompare=false nullComparesHigh=false />
+          out.value = cmp == -1 ? 1 : 0;
+        <#-- TODO:  Refactor other comparison code to here. -->
         <#else>
-        out.value = left.value < right.value ? 1 : 0;
+          ${mode_HAS_BAD_VALUE}
         </#if>
 
       }
   }
-  
-  @FunctionTemplate(names = {"less_than_or_equal_to", "<="}, scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class LessThanE${left}${right} implements DrillSimpleFunc {
 
-      @Param ${left}Holder left;
-      @Param ${right}Holder right;
+  <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+       not for sorting and grouping relational operators. -->
+  @FunctionTemplate(names = {"less_than_or_equal_to", "<="},
+                    scope = FunctionTemplate.FunctionScope.SIMPLE,
+                    nulls = NullHandling.NULL_IF_NULL)
+  public static class LessThanEq${leftTypeBase}Vs${rightTypeBase} implements DrillSimpleFunc {
+
+      @Param ${leftTypeBase}Holder left;
+      @Param ${rightTypeBase}Holder right;
       @Output BitHolder out;
 
       public void setup(RecordBatch b) {}
 
       public void eval() {
-        
-        <#if type.mode == "var" >
-        int cmp;
-        <@compareBlock mode=type.mode left=left right=right output="cmp" nullCompare=false/>
-        out.value = cmp < 1 ? 1 : 0;
+
+        <#if typeGroup.mode == "primitive">
+          out.value = left.value <= right.value ? 1 : 0;
+        <#elseif typeGroup.mode == "varString"
+            || typeGroup.mode == "intervalNameThis" || typeGroup.mode == "intervalDay" >
+          int cmp;
+          <@compareBlock mode=typeGroup.mode leftType=leftTypeBase rightType=rightTypeBase
+                         output="cmp" nullCompare=false nullComparesHigh=false />
+          out.value = cmp < 1 ? 1 : 0;
+        <#-- TODO:  Refactor other comparison code to here. -->
         <#else>
-        out.value = left.value <= right.value ? 1 : 0;
+          ${mode_HAS_BAD_VALUE}
         </#if>
 
     }
   }
-  
-  @FunctionTemplate(names = {"greater_than", ">"}, scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class GreaterThan${left}${right} implements DrillSimpleFunc {
 
-      @Param ${left}Holder left;
-      @Param ${right}Holder right;
+  <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+       not for sorting and grouping relational operators. -->
+  @FunctionTemplate(names = {"greater_than", ">"},
+                    scope = FunctionTemplate.FunctionScope.SIMPLE,
+                    nulls = NullHandling.NULL_IF_NULL)
+  public static class GreaterThan${leftTypeBase}Vs${rightTypeBase} implements DrillSimpleFunc {
+
+      @Param ${leftTypeBase}Holder left;
+      @Param ${rightTypeBase}Holder right;
       @Output BitHolder out;
 
       public void setup(RecordBatch b) {}
 
       public void eval() {
-        
-        <#if type.mode == "var" >
-        int cmp;
-        <@compareBlock mode=type.mode left=left right=right output="cmp" nullCompare=false/>
-        out.value = cmp == 1 ? 1 : 0;
+
+        <#if typeGroup.mode == "primitive">
+          out.value = left.value > right.value ? 1 : 0;
+        <#elseif typeGroup.mode == "varString"
+            || typeGroup.mode == "intervalNameThis" || typeGroup.mode == "intervalDay" >
+          int cmp;
+          <@compareBlock mode=typeGroup.mode leftType=leftTypeBase rightType=rightTypeBase
+                         output="cmp" nullCompare=false nullComparesHigh=false />
+          out.value = cmp == 1 ? 1 : 0;
+        <#-- TODO:  Refactor other comparison code to here. -->
         <#else>
-        out.value = left.value > right.value ? 1 : 0;
+          ${mode_HAS_BAD_VALUE}
         </#if>
 
     }
   }
-  
-  @FunctionTemplate(names = {"greater_than_or_equal_to", ">="}, scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class GreaterThanE${left}${right} implements DrillSimpleFunc {
 
-      @Param ${left}Holder left;
-      @Param ${right}Holder right;
+  <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+       not for sorting and grouping relational operators. -->
+  @FunctionTemplate(names = {"greater_than_or_equal_to", ">="},
+                    scope = FunctionTemplate.FunctionScope.SIMPLE,
+                    nulls = NullHandling.NULL_IF_NULL)
+  public static class GreaterThanEq${leftTypeBase}Vs${rightTypeBase} implements DrillSimpleFunc {
+
+      @Param ${leftTypeBase}Holder left;
+      @Param ${rightTypeBase}Holder right;
       @Output BitHolder out;
 
       public void setup(RecordBatch b) {}
 
       public void eval() {
-        
-        <#if type.mode == "var" >            
-        int cmp;
-        <@compareBlock mode=type.mode left=left right=right output="cmp" nullCompare=false/>
-        out.value = cmp > -1 ? 1 : 0;
+
+        <#if typeGroup.mode == "primitive">
+          out.value = left.value >= right.value ? 1 : 0;
+        <#elseif typeGroup.mode == "varString"
+            || typeGroup.mode == "intervalNameThis" || typeGroup.mode == "intervalDay" >
+          int cmp;
+          <@compareBlock mode=typeGroup.mode leftType=leftTypeBase rightType=rightTypeBase
+                         output="cmp" nullCompare=false nullComparesHigh=false />
+          out.value = cmp > -1 ? 1 : 0;
+        <#-- TODO:  Refactor other comparison code to here. -->
         <#else>
-        out.value = left.value >= right.value ? 1 : 0;
+          ${mode_HAS_BAD_VALUE}
         </#if>
 
       }
   }
-  
-  @FunctionTemplate(names = {"equal","==","="}, scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class Equals${left}${right} implements DrillSimpleFunc {
 
-      @Param ${left}Holder left;
-      @Param ${right}Holder right;
+  <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+       not for sorting and grouping relational operators. -->
+  @FunctionTemplate(names = {"equal", "==", "="},
+                    scope = FunctionTemplate.FunctionScope.SIMPLE,
+                    nulls = NullHandling.NULL_IF_NULL)
+  public static class Equals${leftTypeBase}Vs${rightTypeBase} implements DrillSimpleFunc {
+
+      @Param ${leftTypeBase}Holder left;
+      @Param ${rightTypeBase}Holder right;
       @Output BitHolder out;
 
       public void setup(RecordBatch b) {}
 
       public void eval() {
-          <#if type.mode == "var" >
-          out.value = org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.equal(left.buffer, left.start, left.end, right.buffer, right.start, right.end);
-          <#else>
+
+        <#if typeGroup.mode == "primitive">
           out.value = left.value == right.value ? 1 : 0;
-          </#if>
+        <#elseif typeGroup.mode == "varString" >
+          out.value = org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.equal(
+              left.buffer, left.start, left.end, right.buffer, right.start, right.end);
+        <#elseif typeGroup.mode == "intervalNameThis" || typeGroup.mode == "intervalDay" >
+          int cmp;
+          <@compareBlock mode=typeGroup.mode leftType=leftTypeBase rightType=rightTypeBase
+                         output="cmp" nullCompare=false nullComparesHigh=false />
+          out.value = cmp == 0 ? 1 : 0;
+        <#-- TODO:  Refactor other comparison code to here. -->
+        <#else>
+          ${mode_HAS_BAD_VALUE}
+        </#if>
 
       }
   }
-  
-  @FunctionTemplate(names = {"not_equal","<>","!="}, scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class NotEquals${left}${right} implements DrillSimpleFunc {
 
-      @Param ${left}Holder left;
-      @Param ${right}Holder right;
+  <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+       not for sorting and grouping relational operators. -->
+  @FunctionTemplate(names = {"not_equal", "<>", "!="},
+                    scope = FunctionTemplate.FunctionScope.SIMPLE,
+                    nulls = NullHandling.NULL_IF_NULL)
+  public static class NotEquals${leftTypeBase}Vs${rightTypeBase} implements DrillSimpleFunc {
+
+      @Param ${leftTypeBase}Holder left;
+      @Param ${rightTypeBase}Holder right;
       @Output BitHolder out;
 
       public void setup(RecordBatch b) {}
 
       public void eval() {
-        
-        <#if type.mode == "var" >            
-        int cmp;
-        <@compareBlock mode=type.mode left=left right=right output="cmp" nullCompare=false/>
-        out.value = cmp == 0 ? 0 : 1;
+
+        <#if typeGroup.mode == "primitive">
+          out.value = left.value != right.value ? 1 : 0;
+        <#elseif typeGroup.mode == "varString"
+            || typeGroup.mode == "intervalNameThis" || typeGroup.mode == "intervalDay" >
+          int cmp;
+          <@compareBlock mode=typeGroup.mode leftType=leftTypeBase rightType=rightTypeBase
+                         output="cmp" nullCompare=false nullComparesHigh=false />
+          out.value = cmp == 0 ? 0 : 1;
+        <#-- TODO:  Refactor other comparison code to here. -->
         <#else>
-        out.value = left.value != right.value ? 1 : 0;
+          ${mode_HAS_BAD_VALUE}
         </#if>
-                
+
       }
   }
 
-  </#if>
-
 }
+
+
+</#list> <#-- 2.  Pair of types-->
 </#list>
-</#list>
-</#list>
+
+</#list> <#-- 1. Group -->
 
diff --git a/exec/java-exec/src/main/codegen/templates/DateIntervalFunctions.java b/exec/java-exec/src/main/codegen/templates/DateIntervalFunctions.java
deleted file mode 100644
index 8fe13bbc4..000000000
--- a/exec/java-exec/src/main/codegen/templates/DateIntervalFunctions.java
+++ /dev/null
@@ -1,355 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-<@pp.dropOutputFile />
-
-<#list date.dateTypes as type>
-
-<#if type.name == "Date" || type.name == "TimeStamp" || type.name == "Time" || type.name == "TimeStampTZ" || type.name == "IntervalYear">       <#-- type.name is Date, TimeStamp, Time, TimeStampTZ or IntervalYear -->
-
-<@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/GCompare${type.name}Functions.java" />
-
-<#include "/@includes/license.ftl" />
-
-package org.apache.drill.exec.expr.fn.impl;
-
-import org.apache.drill.exec.expr.DrillSimpleFunc;
-import org.apache.drill.exec.expr.annotations.FunctionTemplate;
-import org.apache.drill.exec.expr.annotations.FunctionTemplate.FunctionCostCategory;
-import org.apache.drill.exec.expr.annotations.FunctionTemplate.NullHandling;
-import org.apache.drill.exec.expr.annotations.Output;
-import org.apache.drill.exec.expr.annotations.Param;
-import org.apache.drill.exec.expr.holders.*;
-import org.apache.drill.exec.record.RecordBatch;
-
-@SuppressWarnings("unused")
-public class GCompare${type.name}Functions {
-
-  @FunctionTemplate(name = "compare_to", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class Compare${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output IntHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-
-          out.value = (left.value > right.value) ? 1 : ((left.value < right.value) ? -1 : 0);
-      }
-  }
-
-  @FunctionTemplate(name = "less_than", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class LessThan${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-        out.value = left.value < right.value ? 1 : 0;
-      }
-  }
-
-  @FunctionTemplate(name = "less_than_or_equal_to", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class LessThanE${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-        out.value = left.value <= right.value ? 1 : 0;
-    }
-  }
-
-  @FunctionTemplate(name = "greater_than", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class GreaterThan${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-        out.value = left.value > right.value ? 1 : 0;
-    }
-  }
-
-  @FunctionTemplate(name = "greater_than_or_equal_to", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class GreaterThanE${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-        out.value = left.value >= right.value ? 1 : 0;
-      }
-  }
-
-  @FunctionTemplate(name = "equal", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class Equals${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-        out.value = left.value == right.value ? 1 : 0;
-      }
-  }
-
-  @FunctionTemplate(name = "not_equal", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class NotEquals${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-        out.value = left.value != right.value ? 1 : 0;
-      }
-  }
-
-}
-
-<#elseif type.name == "Interval" || type.name == "IntervalDay">
-
-<#macro intervalCompareBlock left right leftMonths leftDays leftMillis rightMonths rightDays rightMillis output>
-outside: {
-
-        org.joda.time.MutableDateTime leftDate  = new org.joda.time.MutableDateTime(1970, 1, 1, 0, 0, 0, 0, org.joda.time.DateTimeZone.UTC);
-        org.joda.time.MutableDateTime rightDate = new org.joda.time.MutableDateTime(1970, 1, 1, 0, 0, 0, 0, org.joda.time.DateTimeZone.UTC);
-
-        // Left and right date have the same starting point (epoch), add the interval period and compare the two
-        leftDate.addMonths(${leftMonths});
-        leftDate.addDays(${leftDays});
-        leftDate.add(${leftMillis});
-
-        rightDate.addMonths(${rightMonths});
-        rightDate.addDays(${rightDays});
-        rightDate.add(${rightMillis});
-
-        long leftMS  = leftDate.getMillis();
-        long rightMS = rightDate.getMillis();
-
-        ${output} = ((leftMS < rightMS) ? -1 : ((leftMS > rightMS) ? 1 : 0));
-
-    }
-</#macro>
-
-<#macro intervalConvertBlock left right leftMonths leftDays leftMillis rightMonths rightDays rightMillis>
-        org.joda.time.MutableDateTime leftDate  = new org.joda.time.MutableDateTime(1970, 1, 1, 0, 0, 0, 0, org.joda.time.DateTimeZone.UTC);
-        org.joda.time.MutableDateTime rightDate = new org.joda.time.MutableDateTime(1970, 1, 1, 0, 0, 0, 0, org.joda.time.DateTimeZone.UTC);
-
-        // Left and right date have the same starting point (epoch), add the interval period and compare the two
-        leftDate.addMonths(${leftMonths});
-        leftDate.addDays(${leftDays});
-        leftDate.add(${leftMillis});
-
-        rightDate.addMonths(${rightMonths});
-        rightDate.addDays(${rightDays});
-        rightDate.add(${rightMillis});
-
-        long leftMS  = leftDate.getMillis();
-        long rightMS = rightDate.getMillis();
-</#macro>
-
-<@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/GCompare${type.name}Functions.java" />
-
-<#include "/@includes/license.ftl" />
-
-package org.apache.drill.exec.expr.fn.impl;
-
-import org.apache.drill.exec.expr.DrillSimpleFunc;
-import org.apache.drill.exec.expr.annotations.FunctionTemplate;
-import org.apache.drill.exec.expr.annotations.FunctionTemplate.FunctionCostCategory;
-import org.apache.drill.exec.expr.annotations.FunctionTemplate.NullHandling;
-import org.apache.drill.exec.expr.annotations.Output;
-import org.apache.drill.exec.expr.annotations.Param;
-import org.apache.drill.exec.expr.holders.*;
-import org.apache.drill.exec.record.RecordBatch;
-
-@SuppressWarnings("unused")
-public class GCompare${type.name}Functions {
-
-  @FunctionTemplate(name = "compare_to", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL, 
-      costCategory = FunctionCostCategory.COMPLEX)
-  public static class GCCompare${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output IntHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-
-          <#if type.name == "Interval">
-          <@intervalCompareBlock left="left" right="right" leftMonths="left.months" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="right.months" rightDays="right.days" rightMillis="right.milliseconds" output="out.value"/>
-          <#else>
-          <@intervalCompareBlock left="left" right="right" leftMonths="0" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="0" rightDays="right.days" rightMillis="right.milliseconds" output="out.value"/>
-          </#if>
-
-      }
-  }
-
-  @FunctionTemplate(name = "less_than", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL, 
-      costCategory = FunctionCostCategory.COMPLEX)
-  public static class LessThan${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-
-          <#if type.name == "Interval">
-          <@intervalConvertBlock left="left" right="right" leftMonths="left.months" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="right.months" rightDays="right.days" rightMillis="right.milliseconds"/>
-          <#else>
-          <@intervalConvertBlock left="left" right="right" leftMonths="0" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="0" rightDays="right.days" rightMillis="right.milliseconds"/>
-          </#if>
-
-          out.value = leftMS < rightMS ? 1 : 0;
-      }
-  }
-
-  @FunctionTemplate(name = "less_than_or_equal_to", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL, 
-      costCategory = FunctionCostCategory.COMPLEX)
-  public static class LessThanE${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-
-          <#if type.name == "Interval">
-          <@intervalConvertBlock left="left" right="right" leftMonths="left.months" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="right.months" rightDays="right.days" rightMillis="right.milliseconds"/>
-          <#else>
-          <@intervalConvertBlock left="left" right="right" leftMonths="0" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="0" rightDays="right.days" rightMillis="right.milliseconds"/>
-          </#if>
-
-          out.value = leftMS <= rightMS ? 1 : 0;
-    }
-  }
-
-  @FunctionTemplate(name = "greater_than", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL,
-      costCategory = FunctionCostCategory.COMPLEX)
-  public static class GreaterThan${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-
-          <#if type.name == "Interval">
-          <@intervalConvertBlock left="left" right="right" leftMonths="left.months" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="right.months" rightDays="right.days" rightMillis="right.milliseconds"/>
-          <#else>
-          <@intervalConvertBlock left="left" right="right" leftMonths="0" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="0" rightDays="right.days" rightMillis="right.milliseconds"/>
-          </#if>
-
-          out.value = leftMS > rightMS ? 1 : 0;
-    }
-  }
-
-  @FunctionTemplate(name = "greater_than_or_equal_to", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL,
-      costCategory = FunctionCostCategory.COMPLEX)
-  public static class GreaterThanE${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-
-          <#if type.name == "Interval">
-          <@intervalConvertBlock left="left" right="right" leftMonths="left.months" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="right.months" rightDays="right.days" rightMillis="right.milliseconds"/>
-          <#else>
-          <@intervalConvertBlock left="left" right="right" leftMonths="0" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="0" rightDays="right.days" rightMillis="right.milliseconds"/>
-          </#if>
-
-          out.value = leftMS >= rightMS ? 1 : 0;
-      }
-  }
-
-  @FunctionTemplate(name = "equal", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL,
-      costCategory = FunctionCostCategory.COMPLEX)
-  public static class Equals${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-          <#if type.name == "Interval">
-          <@intervalConvertBlock left="left" right="right" leftMonths="left.months" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="right.months" rightDays="right.days" rightMillis="right.milliseconds"/>
-          <#else>
-          <@intervalConvertBlock left="left" right="right" leftMonths="0" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="0" rightDays="right.days" rightMillis="right.milliseconds"/>
-          </#if>
-
-          out.value = leftMS == rightMS ? 1 : 0;
-      }
-  }
-
-  @FunctionTemplate(name = "not_equal", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL,
-      costCategory = FunctionCostCategory.COMPLEX)
-  public static class NotEquals${type.name} implements DrillSimpleFunc {
-
-      @Param ${type.name}Holder left;
-      @Param ${type.name}Holder right;
-      @Output BitHolder out;
-
-      public void setup(RecordBatch b) {}
-
-      public void eval() {
-          <#if type.name == "Interval">
-          <@intervalConvertBlock left="left" right="right" leftMonths="left.months" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="right.months" rightDays="right.days" rightMillis="right.milliseconds"/>
-          <#else>
-          <@intervalConvertBlock left="left" right="right" leftMonths="0" leftDays="left.days" leftMillis="left.milliseconds" rightMonths="0" rightDays="right.days" rightMillis="right.milliseconds"/>
-          </#if>
-
-          out.value = leftMS != rightMS ? 1 : 0;
-      }
-  }
-}
-</#if>
-</#list>
\ No newline at end of file
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/CastVarCharDecimal.java b/exec/java-exec/src/main/codegen/templates/Decimal/CastVarCharDecimal.java
index 960368a50..84be392fe 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/CastVarCharDecimal.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/CastVarCharDecimal.java
@@ -160,6 +160,7 @@ public class CastEmptyString${type.from}ToNullable${type.to} implements DrillSim
             in.buffer.getBytes(in.start, buf, 0, in.end - in.start);
             throw new org.apache.drill.common.exceptions.DrillRuntimeException("Precision is insufficient for the provided input: " + new String(buf, com.google.common.base.Charsets.UTF_8) + " Precision: " + out.precision +
                                                                                " Total Digits: " + (out.scale + (integerEndIndex - integerStartIndex)));
+            // TODO:  Use JDK's java.nio.charset.StandardCharsets.UTF_8.
         }
 
         // Check if we need to round up
@@ -302,7 +303,7 @@ public class CastEmptyString${type.from}ToNullable${type.to} implements DrillSim
         int radix = 10;
         boolean leadingDigitFound = false;
         boolean round = false;
-    
+
         /* This is the first pass, we get the number of integer digits and based on the provided scale
          * we compute which index into the ByteBuf we start storing the integer part of the Decimal
          */
@@ -343,6 +344,9 @@ public class CastEmptyString${type.from}ToNullable${type.to} implements DrillSim
             }
         }
 
+        <#-- TODO:  Pull out much of this code into something parallel to
+             ByteFunctionHelpers but for DECIMAL type implementations. -->
+
         /* Based on the number of integer digits computed and the scale throw an
          * exception if the provided precision is not sufficient to store the value
          */
@@ -350,6 +354,13 @@ public class CastEmptyString${type.from}ToNullable${type.to} implements DrillSim
             byte[] buf = new byte[in.end - in.start];
             in.buffer.getBytes(in.start, buf, 0, in.end - in.start);
             throw new org.apache.drill.common.exceptions.DrillRuntimeException("Precision is insufficient for the provided input: " + new String(buf, com.google.common.base.Charsets.UTF_8) + " Precision: " + out.precision + " Total Digits: " + (out.scale + integerDigits));
+            <#-- TODO:  Revisit message.  (Message would be clearer and shorter
+                 as something like "Precision of X digits is insufficient for
+                 the provided input of "XXXXX.XXXXX" (X total digits)."  (An
+                 occurrence of "Precision is insufficient for the provided input:
+                 123456789.987654321 Precision: 5 Total Digits: 9" seemed to
+                 mean that 5 post-decimal digits and 9 total digits were allowed.)
+                 -->
         }
 
 
@@ -444,7 +455,7 @@ public class CastEmptyString${type.from}ToNullable${type.to} implements DrillSim
 
             int carry = 0;
             do {
-                // propogate the carry
+                // propagate the carry
                 int tempValue = out.getInteger(decimalBufferIndex, out.start, out.buffer) + carry;
                 if (tempValue >= org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE) {
                     carry = tempValue / org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE;
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/DecimalFunctions.java b/exec/java-exec/src/main/codegen/templates/Decimal/DecimalFunctions.java
index 0c4af01b7..b9029cd93 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/DecimalFunctions.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/DecimalFunctions.java
@@ -20,212 +20,76 @@ import org.apache.drill.exec.expr.annotations.Workspace;
 
 <@pp.dropOutputFile />
 
-<#macro compareBlock holderType left right absCompare output>
-
-        outside:{
-            ${output} = org.apache.drill.exec.util.DecimalUtility.compareSparseBytes(left.buffer, left.start, left.getSign(left.start, left.buffer),
-                            left.scale, left.precision, right.buffer,
-                            right.start, right.getSign(right.start, right.buffer), right.precision,
-                            right.scale, left.WIDTH, left.nDecimalDigits, ${absCompare});
-
-    }
-</#macro>
-
-<#macro subtractBlock holderType in1 in2 result>
-
-            int resultScaleRoundedUp = org.apache.drill.exec.util.DecimalUtility.roundUp(result.scale);
-            int resultIndex = result.nDecimalDigits- 1;
-
-            int leftScaleRoundedUp  = org.apache.drill.exec.util.DecimalUtility.roundUp(${in1}.scale);
-            int leftIntRoundedUp    = org.apache.drill.exec.util.DecimalUtility.roundUp(${in1}.precision - ${in1}.scale);
-            int rightScaleRoundedUp = org.apache.drill.exec.util.DecimalUtility.roundUp(${in2}.scale);
-
-            int leftIndex  = ${in1}.nDecimalDigits - 1;
-            int rightIndex = ${in2}.nDecimalDigits - 1;
-
-            /* If the left scale is bigger, simply copy over the digits into result */
-            while (leftScaleRoundedUp > rightScaleRoundedUp) {
-                result.setInteger(resultIndex, ${in1}.getInteger(leftIndex, ${in1}.start, ${in1}.buffer), result.start, result.buffer);
-                leftIndex--;
-                resultIndex--;
-                leftScaleRoundedUp--;
-            }
-
-            /* If the right scale is bigger, subtract with zero at each array location */
-            int carry = 0;
-            while(rightScaleRoundedUp > leftScaleRoundedUp) {
-
-                int difference = 0 - ${in2}.getInteger(rightIndex, ${in2}.start, ${in2}.buffer) - carry;
-                rightIndex--;
-
-                if (difference < 0) {
-                    carry = 1;
-                    result.setInteger(resultIndex, (difference + org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE), result.start, result.buffer);
-                } else {
-                    result.setInteger(resultIndex, difference, result.start, result.buffer);
-                    carry = 0;
-                }
-                resultIndex--;
-                rightScaleRoundedUp--;
-
-            }
-
-            /* Now both the scales are equal perform subtraction use one of the scales
-             * for terminal condition in the while loop
-             */
-            while (leftScaleRoundedUp > 0) {
-
-                int difference = ${in1}.getInteger(leftIndex, ${in1}.start, ${in1}.buffer) - ${in2}.getInteger(rightIndex, ${in2}.start, ${in2}.buffer) - carry;
-                leftIndex--;
-                rightIndex--;
-
-                if (difference < 0) {
-                    carry = 1;
-                    result.setInteger(resultIndex, (difference + org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE), result.start, result.buffer);
-                } else {
-                    result.setInteger(resultIndex, difference, result.start, result.buffer);
-                    carry = 0;
-                }
-                resultIndex--;
-                leftScaleRoundedUp--;
-            }
-
-            /* Since we are gurranteed to have the left input >= right input, iterate
-             * over the remaining left input's integers
-             */
-            while(leftIntRoundedUp > 0) {
-
-                int difference = ${in1}.getInteger(leftIndex, ${in1}.start, ${in1}.buffer);
-                leftIndex--;
-
-                if (rightIndex >= 0) {
-                    difference -= ${in2}.getInteger(rightIndex, ${in2}.start, ${in2}.buffer);
-                    rightIndex--;
-                }
-
-                difference -= carry;
-
-                if (difference < 0) {
-                    carry = 1;
-                    result.setInteger(resultIndex, (difference + org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE), result.start, result.buffer);
-                } else {
-                    carry = 0;
-                    result.setInteger(resultIndex, difference, result.start, result.buffer);
-                }
-                resultIndex--;
-                leftIntRoundedUp--;
-            }
+<#-- TODO:  Refactor comparison code from here into ComparisonFunctions (to
+     eliminate duplicate template code and so that ComparisonFunctions actually
+     as all comparison functions. -->
+
+<#macro compareNullsSubblock leftType rightType output breakTarget nullCompare nullComparesHigh>
+  <#if nullCompare>
+    <#if nullComparesHigh>
+      <#assign leftNullResult = 1> <#-- if only L is null and nulls are high, then "L > R" (1) -->
+      <#assign rightNullResult = -1>
+    <#else>
+      <#assign leftNullResult = -1> <#-- if only L is null and nulls are low, then "L < R" (-1) -->
+      <#assign rightNullResult = 1>
+    </#if>
+
+    <#if leftType?starts_with("Nullable")>
+      <#if rightType?starts_with("Nullable")>
+        <#-- Both are nullable. -->
+        if ( left.isSet == 0 ) {
+          if ( right.isSet == 0 ) {
+            <#-- Both are null--result is "L = R". -->
+            ${output} = 0;
+            break ${breakTarget};
+          } else {
+            <#-- Only left is null--result is "L < R" or "L > R" per null ordering. -->
+            ${output} = ${leftNullResult};
+            break ${breakTarget};
+          }
+        } else if ( right.isSet == 0 ) {
+          <#-- Only right is null--result is "L > R" or "L < R" per null ordering. -->
+          ${output} = ${rightNullResult};
+          break ${breakTarget};
+        }
+      <#else>
+        <#-- Left is nullable but right is not. -->
+        if ( left.isSet == 0 ) {
+          <#-- Only left is null--result is "L < R" or "L > R" per null ordering. -->
+          ${output} = ${leftNullResult};
+          break ${breakTarget};
+        }
+      </#if>
+    <#elseif rightType?starts_with("Nullable")>
+      <#-- Left is not nullable but right is. -->
+      if ( right.isSet == 0 ) {
+        <#-- Only right is null--result is "L > R" or "L < R" per null ordering. -->
+        ${output} = ${rightNullResult};
+        break ${breakTarget};
+      }
+    </#if>
+  </#if>
 
 </#macro>
 
-<#macro addBlock holderType in1 in2 result>
-
-        int resultScaleRoundedUp = org.apache.drill.exec.util.DecimalUtility.roundUp(result.scale);
-
-        int leftScaleRoundedUp  = org.apache.drill.exec.util.DecimalUtility.roundUp(${in1}.scale);
-        int rightScaleRoundedUp = org.apache.drill.exec.util.DecimalUtility.roundUp(${in2}.scale);
-
-        /* starting index for each decimal */
-        int leftIndex  = ${in1}.nDecimalDigits - 1;
-        int rightIndex = ${in2}.nDecimalDigits - 1;
-        int resultIndex = result.nDecimalDigits - 1;
-
-        /* If one of the scale is larger then simply copy it over
-         * to the result digits
-         */
-        while (leftScaleRoundedUp > rightScaleRoundedUp) {
-
-            result.setInteger(resultIndex, ${in1}.getInteger(leftIndex, ${in1}.start, ${in1}.buffer), result.start, result.buffer);
-            leftIndex--;
-            resultIndex--;
-            leftScaleRoundedUp--;
-            resultScaleRoundedUp--;
-        }
-
-        while (rightScaleRoundedUp > leftScaleRoundedUp) {
-            result.setInteger((resultIndex), ${in2}.getInteger(rightIndex, ${in2}.start, ${in2}.buffer), result.start, result.buffer);
-            rightIndex--;
-            resultIndex--;
-            rightScaleRoundedUp--;
-            resultScaleRoundedUp--;
-        }
-
-        int sum = 0;
 
-        /* now the two scales are at the same level, we can add them */
-        while (resultScaleRoundedUp > 0) {
-
-            sum += ${in1}.getInteger(leftIndex, ${in1}.start, ${in1}.buffer) + ${in2}.getInteger(rightIndex, ${in2}.start, ${in2}.buffer);
-            leftIndex--;
-            rightIndex--;
-
-            if (sum >= org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE) {
-                result.setInteger(resultIndex, (sum - org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE), result.start, result.buffer);
-                sum = 1;
-            } else {
-                result.setInteger(resultIndex, sum, result.start, result.buffer);
-                sum = 0;
-            }
-            resultIndex--;
-            resultScaleRoundedUp--;
-        }
-
-        /* add the integer part */
-        while (leftIndex >= 0 && rightIndex >= 0) {
-
-            sum += ${in1}.getInteger(leftIndex, ${in1}.start, ${in1}.buffer) + ${in2}.getInteger(rightIndex, ${in2}.start, ${in2}.buffer);
-            leftIndex--;
-            rightIndex--;
-
-            if (sum >= org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE) {
-                result.setInteger(resultIndex, (sum - org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE), result.start, result.buffer);
-                sum = 1;
-            } else {
-                result.setInteger(resultIndex, sum, result.start, result.buffer);
-                sum = 0;
-            }
-            resultIndex--;
-        }
-
-        while (resultIndex >= 0 && leftIndex >= 0) {
-            sum += ${in1}.getInteger(leftIndex, ${in1}.start, ${in1}.buffer);
-            leftIndex--;
-
-            if (sum >= org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE) {
-                result.setInteger(resultIndex, (sum - org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE), result.start, result.buffer);
-                sum = 1;
-            } else {
-                result.setInteger(resultIndex, sum, result.start, result.buffer);
-                sum = 0;
-            }
-        }
-
-        while (resultIndex >= 0 && rightIndex >= 0) {
-            sum += ${in2}.getInteger(rightIndex, ${in2}.start, ${in2}.buffer);
-            rightIndex--;
-
-            if (sum >= org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE) {
-                result.setInteger(resultIndex, (sum - org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE), result.start, result.buffer);
-                sum = 1;
-            } else {
-                result.setInteger(resultIndex, sum, result.start, result.buffer);
-                sum = 0;
-            }
-        }
+<#macro compareBlock leftType rightType absCompare output nullCompare nullComparesHigh>
+         outside:
+          {
+            <@compareNullsSubblock leftType=leftType rightType=rightType output=output breakTarget="outside" nullCompare=nullCompare nullComparesHigh=nullComparesHigh />
 
-        /* store the last carry */
-        if (sum > 0)
-        result.setInteger(resultIndex, sum, result.start, result.buffer);
+            ${output} = org.apache.drill.exec.util.DecimalUtility.compareSparseBytes(left.buffer, left.start, left.getSign(left.start, left.buffer),
+                            left.scale, left.precision, right.buffer,
+                            right.start, right.getSign(right.start, right.buffer), right.precision,
+                            right.scale, left.WIDTH, left.nDecimalDigits, ${absCompare});
 
+          } // outside
 </#macro>
 
-
-<#macro adjustScale holderType javaType left right>
+<#macro adjustScale javaType leftType rightType>
 
             // Adjust the scale of the two inputs to be the same
 
-            int adjustment = 0;
-
             if (left.scale < right.scale) {
                 left.value = (${javaType}) (org.apache.drill.exec.util.DecimalUtility.adjustScaleMultiply(left.value, (int) (right.scale - left.scale)));
                 left.scale = right.scale;
@@ -235,9 +99,11 @@ import org.apache.drill.exec.expr.annotations.Workspace;
             }
 </#macro>
 
-<#list decimal.decimalTypes as type>
+<#-- For each DECIMAL... type (in DecimalTypes.tdd) ... -->
+<#list comparisonTypesDecimal.decimalTypes as type>
 
 <#if type.name.endsWith("Sparse")>
+
 <@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/${type.name}Functions.java" />
 
 <#include "/@includes/license.ftl" />
@@ -252,6 +118,7 @@ import org.apache.drill.exec.expr.annotations.FunctionTemplate.NullHandling;
 import org.apache.drill.exec.expr.annotations.Output;
 import org.apache.drill.exec.expr.annotations.Param;
 import org.apache.drill.exec.expr.annotations.Workspace;
+import org.apache.drill.exec.expr.fn.FunctionGenerationHelper;
 import org.apache.drill.exec.expr.holders.*;
 import org.apache.drill.exec.record.RecordBatch;
 
@@ -295,7 +162,7 @@ public class ${type.name}Functions {
             java.math.BigDecimal rightInput = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromSparse(right.buffer, right.start, right.nDecimalDigits, right.scale);
             java.math.BigDecimal addResult = leftInput.subtract(rightInput);
 
-            // set the scale
+            // Set the scale
             addResult.setScale(result.scale, java.math.BigDecimal.ROUND_HALF_UP);
             org.apache.drill.exec.util.DecimalUtility.getSparseFromBigDecimal(addResult, result.buffer, result.start, result.scale, result.precision, result.nDecimalDigits);
         }
@@ -333,7 +200,7 @@ public class ${type.name}Functions {
             java.math.BigDecimal rightInput = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromSparse(right.buffer, right.start, right.nDecimalDigits, right.scale);
             java.math.BigDecimal addResult = leftInput.add(rightInput);
 
-            // set the scale
+            // Set the scale
             addResult.setScale(result.scale, java.math.BigDecimal.ROUND_HALF_UP);
             org.apache.drill.exec.util.DecimalUtility.getSparseFromBigDecimal(addResult, result.buffer, result.start, result.scale, result.precision, result.nDecimalDigits);
         }
@@ -387,7 +254,7 @@ public class ${type.name}Functions {
 
             int leftIntegerSize = leftStopIndex - leftIndex;
 
-            /* Remove the leaing zeroes from the integer part of the input */
+            /* Remove the leading zeroes from the integer part of the input */
             int rightIndex = 0;
             int rightStopIndex = right.nDecimalDigits - org.apache.drill.exec.util.DecimalUtility.roundUp(right.scale);
 
@@ -429,7 +296,7 @@ public class ${type.name}Functions {
 
                     currentIndex--;
                 }
-                /* propogate the carry */
+                /* Propagate the carry */
                 if (carry > 0)
                     tempResult[currentIndex] += carry;
 
@@ -442,17 +309,17 @@ public class ${type.name}Functions {
             resultScaleSize = org.apache.drill.exec.util.DecimalUtility.roundUp(result.scale);
 
             if (result.scale < (left.scale + right.scale)) {
-              /* The scale of the output data type is lesser than the scale
+              /* The scale of the output data type is less than the scale
                * we obtained as a result of multiplication, we need to round
                * a chunk of the fractional part
                */
               int lastScaleIndex = currentIndex + resultIntegerSize + resultScaleSize - 1;
 
-              // compute the power of 10 necessary to find if we need to round up
+              // Compute the power of 10 necessary to find if we need to round up
               int roundFactor = (int) (org.apache.drill.exec.util.DecimalUtility.getPowerOfTen(
                                         org.apache.drill.exec.util.DecimalUtility.MAX_DIGITS - ((result.scale + 1) % org.apache.drill.exec.util.DecimalUtility.MAX_DIGITS)));
 
-              // index of rounding digit
+              // Index of rounding digit
               int roundIndex = currentIndex + resultIntegerSize + org.apache.drill.exec.util.DecimalUtility.roundUp(result.scale + 1) - 1;
 
               // Check the first chopped digit to see if we need to round up
@@ -471,7 +338,7 @@ public class ${type.name}Functions {
                 carry *= scaleFactor;
               }
 
-              // propogate the carry
+              // Propagate the carry
               while (carry > 0 && lastScaleIndex >= 0) {
                 int tempSum = tempResult[lastScaleIndex] + carry;
                 if (tempSum >= org.apache.drill.exec.util.DecimalUtility.DIGITS_BASE) {
@@ -484,7 +351,7 @@ public class ${type.name}Functions {
                 lastScaleIndex--;
               }
 
-              // check if carry has increased integer digit
+              // Check if carry has increased integer digit
               if ((lastScaleIndex + 1) < currentIndex) {
                 resultIntegerSize++;
                 currentIndex = lastScaleIndex + 1;
@@ -599,7 +466,7 @@ public class ${type.name}Functions {
           out.buffer = in.buffer;
           out.start = in.start;
 
-          // set the output buffer with the positive sign
+          // Set the output buffer with the positive sign
           out.buffer.setInt(out.start, (out.buffer.getInt(out.start) & 0x7fffffff));
         }
     }
@@ -691,7 +558,7 @@ public class ${type.name}Functions {
               }
             }
           }
-          // set the sign
+          // Set the sign
           out.setSign(sign, out.start, out.buffer);
         }
     }
@@ -756,7 +623,7 @@ public class ${type.name}Functions {
               }
             }
           }
-          // set the sign
+          // Set the sign
           out.setSign(sign, out.start, out.buffer);
         }
     }
@@ -790,8 +657,8 @@ public class ${type.name}Functions {
           while (destIndex >= 0) {
             out.setInteger(destIndex--, 0, out.start, out.buffer);
           }
-            // set the sign
-            out.setSign(sign, out.start, out.buffer);
+          // Set the sign
+          out.setSign(sign, out.start, out.buffer);
         }
     }
 
@@ -867,8 +734,8 @@ public class ${type.name}Functions {
               }
             }
           }
-            // set the sign
-            result.setSign(sign, result.start, result.buffer);
+          // Set the sign
+          result.setSign(sign, result.start, result.buffer);
         }
     }
 
@@ -929,8 +796,8 @@ public class ${type.name}Functions {
               }
             }
           }
-            // set the sign
-            out.setSign(sign, out.start, out.buffer);
+          // Set the sign
+          out.setSign(sign, out.start, out.buffer);
         }
     }
 
@@ -952,26 +819,62 @@ public class ${type.name}Functions {
           boolean sign = left.getSign(left.start, left.buffer);
 
           org.apache.drill.exec.util.DecimalUtility.roundDecimal(result.buffer, result.start, result.nDecimalDigits, result.scale, left.scale);
-          // set the sign
+          // Set the sign
           result.setSign(sign, result.start, result.buffer);
         }
     }
 
-    @FunctionTemplate(name = "compare_to", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
-    public static class ${type.name}CompareTo implements DrillSimpleFunc {
+ <#-- Handle 2 x 2 combinations of nullable and non-nullable arguments. -->
+ <#list ["Nullable${type.name}", "${type.name}"] as leftType >
+ <#list ["Nullable${type.name}", "${type.name}"] as rightType >
 
-        @Param ${type.name}Holder left;
-        @Param ${type.name}Holder right;
-        @Output IntHolder out;
-        public void setup(RecordBatch incoming) {}
+  <#-- Comparison function for sorting and grouping relational operators
+       (not for comparison expression operators (=, <, etc.)). -->
+  @FunctionTemplate(name = FunctionGenerationHelper.COMPARE_TO_NULLS_HIGH,
+                    scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                    nulls = NullHandling.INTERNAL)
+  public static class GCompare${leftType}Vs${rightType}NullHigh implements DrillSimpleFunc {
 
-        public void eval() {
-             <@compareBlock holderType=type.name left="left" right="right" absCompare="false" output="out.value"/>
-        }
+    @Param ${leftType}Holder left;
+    @Param ${rightType}Holder right;
+    @Output IntHolder out;
+
+    public void setup(RecordBatch incoming) {}
+
+    public void eval() {
+      <@compareBlock leftType=leftType rightType=rightType absCompare="false" output="out.value" nullCompare=true nullComparesHigh=true />
     }
+  }
 
 
-    @FunctionTemplate(name = "less than", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+
+  <#-- Comparison function for sorting and grouping relational operators
+        (not for comparison expression operators (=, <, etc.)). -->
+  @FunctionTemplate(name = FunctionGenerationHelper.COMPARE_TO_NULLS_LOW,
+                    scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                    nulls = NullHandling.INTERNAL)
+  public static class GCompare${leftType}Vs${rightType}NullLow implements DrillSimpleFunc {
+
+    @Param ${leftType}Holder left;
+    @Param ${rightType}Holder right;
+    @Output IntHolder out;
+
+    public void setup(RecordBatch incoming) {}
+
+    public void eval() {
+      <@compareBlock leftType=leftType rightType=rightType absCompare="false" output="out.value" nullCompare=true nullComparesHigh=false />
+    }
+  }
+
+ </#list>
+ </#list>
+
+
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "less than",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}LessThan implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -981,12 +884,19 @@ public class ${type.name}Functions {
 
         public void eval() {
             int cmp;
-            <@compareBlock holderType=type.name left="left" right="right" absCompare="false" output="cmp"/>
+            <@compareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
             out.value = cmp == -1 ? 1 : 0;
         }
     }
 
-    @FunctionTemplate(name = "less than or equal to", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+
+    // TODO:  RESOLVE:  Here there are spaces in function template names, but
+    // elsewhere there are underlines.  Are things being looked up correctly?
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "less than or equal to",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}LessThanEq implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -996,12 +906,17 @@ public class ${type.name}Functions {
 
         public void eval() {
             int cmp;
-            <@compareBlock holderType=type.name left="left" right="right" absCompare="false" output="cmp"/>
+            <@compareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
             out.value = cmp < 1 ? 1 : 0;
         }
     }
 
-    @FunctionTemplate(name = "greater than", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "greater than",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}GreaterThan implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1011,12 +926,17 @@ public class ${type.name}Functions {
 
         public void eval() {
             int cmp;
-            <@compareBlock holderType=type.name left="left" right="right" absCompare="false" output="cmp"/>
+            <@compareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
             out.value = cmp == 1 ? 1 : 0;
         }
     }
 
-    @FunctionTemplate(name = "greater than or equal to", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "greater than or equal to",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}GreaterThanEq implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1026,12 +946,17 @@ public class ${type.name}Functions {
 
         public void eval() {
             int cmp;
-            <@compareBlock holderType=type.name left="left" right="right" absCompare="false" output="cmp"/>
+            <@compareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
             out.value = cmp > -1 ? 1 : 0;
         }
     }
 
-    @FunctionTemplate(name = "Equal", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "Equal",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}Equal implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1041,12 +966,17 @@ public class ${type.name}Functions {
 
         public void eval() {
             int cmp;
-            <@compareBlock holderType=type.name left="left" right="right" absCompare="false" output="cmp"/>
+            <@compareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
             out.value = cmp == 0 ? 1 : 0;
         }
     }
 
-    @FunctionTemplate(name = "not equal", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "not equal",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}NotEqual implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1056,13 +986,14 @@ public class ${type.name}Functions {
 
         public void eval() {
             int cmp;
-            <@compareBlock holderType=type.name left="left" right="right" absCompare="false" output="cmp"/>
+            <@compareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
             out.value = cmp != 0 ? 1 : 0;
         }
     }
 }
 
 <#elseif type.name.endsWith("Dense")>
+
 <@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/${type.name}Functions.java" />
 
 <#include "/@includes/license.ftl" />
@@ -1076,6 +1007,7 @@ import org.apache.drill.exec.expr.annotations.FunctionTemplate;
 import org.apache.drill.exec.expr.annotations.FunctionTemplate.NullHandling;
 import org.apache.drill.exec.expr.annotations.Output;
 import org.apache.drill.exec.expr.annotations.Param;
+import org.apache.drill.exec.expr.fn.FunctionGenerationHelper;
 import org.apache.drill.exec.expr.holders.*;
 import org.apache.drill.exec.record.RecordBatch;
 
@@ -1087,20 +1019,65 @@ import java.nio.ByteBuffer;
 public class ${type.name}Functions {
 
 
-    @FunctionTemplate(name = "compare_to", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
-    public static class ${type.name}CompareTo implements DrillSimpleFunc {
+ <#-- Handle 2 x 2 combinations of nullable and non-nullable arguments. -->
+ <#list ["Nullable${type.name}", "${type.name}"] as leftType >
+ <#list ["Nullable${type.name}", "${type.name}"] as rightType >
 
-        @Param ${type.name}Holder left;
-        @Param ${type.name}Holder right;
-        @Output IntHolder out;
-        public void setup(RecordBatch incoming) {}
+  <#-- Comparison function for sorting and grouping relational operators
+       (not for comparison expression operators (=, <, etc.)). -->
+  @FunctionTemplate(name = FunctionGenerationHelper.COMPARE_TO_NULLS_HIGH,
+                    scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                    nulls = NullHandling.INTERNAL)
+  public static class GCompare${leftType}Vs${rightType}NullHigh implements DrillSimpleFunc {
 
-        public void eval() {
-            out.value = org.apache.drill.exec.util.DecimalUtility.compareDenseBytes(left.buffer, left.start, left.getSign(left.start, left.buffer), right.buffer, right.start, right.getSign(right.start, right.buffer), left.WIDTH);
-        }
+    @Param ${leftType}Holder left;
+    @Param ${rightType}Holder right;
+    @Output IntHolder out;
+
+    public void setup(RecordBatch incoming) {}
+
+    public void eval() {
+     outside:
+      {
+        <@compareNullsSubblock leftType=leftType rightType=rightType output="out.value" breakTarget="outside" nullCompare=true nullComparesHigh=true />
+
+        out.value = org.apache.drill.exec.util.DecimalUtility.compareDenseBytes(left.buffer, left.start, left.getSign(left.start, left.buffer), right.buffer, right.start, right.getSign(right.start, right.buffer), left.WIDTH);
+      } // outside
+    }
+  }
+
+  <#-- Comparison function for sorting and grouping relational operators
+       (not for comparison expression operators (=, <, etc.)). -->
+  @FunctionTemplate(name = FunctionGenerationHelper.COMPARE_TO_NULLS_LOW,
+                    scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                    nulls = NullHandling.INTERNAL)
+  public static class GCompare${leftType}Vs${rightType}NullLow implements DrillSimpleFunc {
+
+    @Param ${leftType}Holder left;
+    @Param ${rightType}Holder right;
+    @Output IntHolder out;
+
+    public void setup(RecordBatch incoming) {}
+
+    public void eval() {
+     outside:
+      {
+        <@compareNullsSubblock leftType=leftType rightType=rightType output="out.value" breakTarget="outside" nullCompare=true nullComparesHigh=false />
+
+        out.value = org.apache.drill.exec.util.DecimalUtility.compareDenseBytes(left.buffer, left.start, left.getSign(left.start, left.buffer), right.buffer, right.start, right.getSign(right.start, right.buffer), left.WIDTH);
+      } // outside
     }
+  }
+
+ </#list>
+ </#list>
+
 
-    @FunctionTemplate(name = "less than", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "less than",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}LessThan implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1114,7 +1091,11 @@ public class ${type.name}Functions {
         }
     }
 
-    @FunctionTemplate(name = "less than or equal to", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "less than or equal to",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}LessThanEq implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1128,7 +1109,11 @@ public class ${type.name}Functions {
         }
     }
 
-    @FunctionTemplate(name = "greater than", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "greater than",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}GreaterThan implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1142,7 +1127,11 @@ public class ${type.name}Functions {
         }
     }
 
-    @FunctionTemplate(name = "greater than or equal to", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "greater than or equal to",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}GreaterThanEq implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1156,7 +1145,11 @@ public class ${type.name}Functions {
         }
     }
 
-    @FunctionTemplate(name = "Equal", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "Equal",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}Equal implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1171,7 +1164,11 @@ public class ${type.name}Functions {
     }
 
 
-    @FunctionTemplate(name = "not equal", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "not equal",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}NotEqual implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1186,6 +1183,7 @@ public class ${type.name}Functions {
         }
     }
 }
+
 <#elseif type.name.endsWith("Decimal9") || type.name.endsWith("Decimal18")>
 
 <@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/${type.name}Functions.java" />
@@ -1202,6 +1200,7 @@ import org.apache.drill.exec.expr.annotations.FunctionTemplate.NullHandling;
 import org.apache.drill.exec.expr.annotations.Output;
 import org.apache.drill.exec.expr.annotations.Param;
 import org.apache.drill.exec.expr.annotations.Workspace;
+import org.apache.drill.exec.expr.fn.FunctionGenerationHelper;
 import org.apache.drill.exec.expr.holders.*;
 import org.apache.drill.exec.record.RecordBatch;
 
@@ -1233,7 +1232,7 @@ public class ${type.name}Functions {
                 outputScale = resultScalePrec.getOutputScale();
                 outputPrecision = resultScalePrec.getOutputPrecision();
             }
-            <@adjustScale holderType=type.name javaType=type.storage left="left" right="right"/>
+            <@adjustScale javaType=type.storage leftType="leftType" rightType="rightType"/>
 
             result.value = left.value + right.value;
             result.precision = outputPrecision;
@@ -1261,7 +1260,7 @@ public class ${type.name}Functions {
                 outputScale = resultScalePrec.getOutputScale();
                 outputPrecision = resultScalePrec.getOutputPrecision();
             }
-            <@adjustScale holderType=type.name javaType=type.storage left="left" right="right"/>
+            <@adjustScale javaType=type.storage leftType="leftType" rightType="rightType"/>
 
             result.value = left.value - right.value;
             result.precision = outputPrecision;
@@ -1511,7 +1510,9 @@ public class ${type.name}Functions {
         }
     }
 
-    @FunctionTemplate(name = "round", scope = FunctionTemplate.FunctionScope.DECIMAL_SET_SCALE, nulls = NullHandling.NULL_IF_NULL)
+    @FunctionTemplate(name = "round",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_SET_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}RoundScaleFunction implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1562,21 +1563,65 @@ public class ${type.name}Functions {
         }
     }
 
-    @FunctionTemplate(name = "compare_to", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
-    public static class ${type.name}CompareTo implements DrillSimpleFunc {
+ <#-- Handle 2 x 2 combinations of nullable and non-nullable arguments. -->
+ <#list ["Nullable${type.name}", "${type.name}"] as leftType >
+ <#list ["Nullable${type.name}", "${type.name}"] as rightType >
 
-        @Param ${type.name}Holder left;
-        @Param ${type.name}Holder right;
-        @Output IntHolder out;
-        public void setup(RecordBatch incoming) {}
+  <#-- Comparison function for sorting and grouping relational operators
+       (not for comparison expression operators (=, <, etc.)). -->
+  @FunctionTemplate(name = FunctionGenerationHelper.COMPARE_TO_NULLS_HIGH,
+                    scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                    nulls = NullHandling.INTERNAL)
+  public static class GCompare${leftType}Vs${rightType}NullHigh implements DrillSimpleFunc {
 
-        public void eval() {
-            <@adjustScale holderType=type.name javaType=type.storage left="left" right="right"/>
-            out.value = (left.value < right.value) ? -1 : (left.value > right.value) ? 1 : 0;
-        }
+    @Param ${leftType}Holder left;
+    @Param ${rightType}Holder right;
+    @Output IntHolder out;
+
+    public void setup(RecordBatch incoming) {}
+
+    public void eval() {
+     outside:
+      {
+        <@compareNullsSubblock leftType=leftType rightType=rightType output="out.value" breakTarget="outside" nullCompare=true nullComparesHigh=true />
+
+        <@adjustScale javaType=type.storage leftType="leftType" rightType="rightType"/>
+        out.value = (left.value < right.value) ? -1 : (left.value > right.value) ? 1 : 0;
+      } // outside
     }
+  }
+
+  <#-- Comparison function for sorting and grouping relational operators
+       (not for comparison expression operators (=, <, etc.)). -->
+  @FunctionTemplate(name = FunctionGenerationHelper.COMPARE_TO_NULLS_LOW,
+                    scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                    nulls = NullHandling.INTERNAL)
+  public static class GCompare${leftType}Vs${rightType}NullLow implements DrillSimpleFunc {
+
+    @Param ${leftType}Holder left;
+    @Param ${rightType}Holder right;
+    @Output IntHolder out;
+
+    public void setup(RecordBatch incoming) {}
+
+    public void eval() {
+     outside:
+      {
+        <@compareNullsSubblock leftType=leftType rightType=rightType output="out.value" breakTarget="outside" nullCompare=true nullComparesHigh=false />
+        <@adjustScale javaType=type.storage leftType="leftType" rightType="rightType"/>
+        out.value = (left.value < right.value) ? -1 : (left.value > right.value) ? 1 : 0;
+      } // outside
+    }
+  }
+
+ </#list>
+ </#list>
 
-    @FunctionTemplate(name = "less than", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "less than",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}LessThan implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1585,13 +1630,16 @@ public class ${type.name}Functions {
         public void setup(RecordBatch incoming) {}
 
         public void eval() {
-            int cmp;
-            <@adjustScale holderType=type.name javaType=type.storage left="left" right="right"/>
+            <@adjustScale javaType=type.storage leftType="leftType" rightType="rightType"/>
             out.value = (left.value < right.value) ? 1 : 0;
         }
     }
 
-    @FunctionTemplate(name = "less than or equal to", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+     <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+          not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "less than or equal to",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}LessThanEq implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1600,13 +1648,16 @@ public class ${type.name}Functions {
         public void setup(RecordBatch incoming) {}
 
         public void eval() {
-            int cmp;
-            <@adjustScale holderType=type.name javaType=type.storage left="left" right="right"/>
+            <@adjustScale javaType=type.storage leftType="leftType" rightType="rightType"/>
             out.value = (left.value <= right.value) ? 1 : 0;
         }
     }
 
-    @FunctionTemplate(name = "greater than", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+     <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+          not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "greater than",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}GreaterThan implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1615,13 +1666,16 @@ public class ${type.name}Functions {
         public void setup(RecordBatch incoming) {}
 
         public void eval() {
-            int cmp;
-            <@adjustScale holderType=type.name javaType=type.storage left="left" right="right"/>
+            <@adjustScale javaType=type.storage leftType="leftType" rightType="rightType"/>
             out.value = (left.value > right.value) ? 1 : 0;
         }
     }
 
-    @FunctionTemplate(name = "greater than or equal to", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+     <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+          not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "greater than or equal to",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}GreaterThanEq implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1630,13 +1684,16 @@ public class ${type.name}Functions {
         public void setup(RecordBatch incoming) {}
 
         public void eval() {
-            int cmp;
-            <@adjustScale holderType=type.name javaType=type.storage left="left" right="right"/>
+            <@adjustScale javaType=type.storage leftType="leftType" rightType="rightType"/>
             out.value = (left.value >= right.value) ? 1 : 0;
         }
     }
 
-    @FunctionTemplate(name = "Equal", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+     <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+          not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "Equal",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}Equal implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1645,14 +1702,17 @@ public class ${type.name}Functions {
         public void setup(RecordBatch incoming) {}
 
         public void eval() {
-            int cmp;
-            <@adjustScale holderType=type.name javaType=type.storage left="left" right="right"/>
+            <@adjustScale javaType=type.storage leftType="leftType" rightType="rightType"/>
             out.value = (left.value == right.value) ? 1 : 0;
         }
     }
 
 
-    @FunctionTemplate(name = "not equal", scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE, nulls = NullHandling.NULL_IF_NULL)
+     <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+          not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "not equal",
+                      scope = FunctionTemplate.FunctionScope.DECIMAL_MAX_SCALE,
+                      nulls = NullHandling.NULL_IF_NULL)
     public static class ${type.name}NotEqual implements DrillSimpleFunc {
 
         @Param ${type.name}Holder left;
@@ -1661,13 +1721,12 @@ public class ${type.name}Functions {
         public void setup(RecordBatch incoming) {}
 
         public void eval() {
-
-            int cmp;
-            <@adjustScale holderType=type.name javaType=type.storage left="left" right="right"/>
+            <@adjustScale javaType=type.storage leftType="leftType" rightType="rightType"/>
             out.value = (left.value != right.value) ? 1 : 0;
         }
     }
 }
 
 </#if>
-</#list>
\ No newline at end of file
+
+</#list>
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java
index 1ed3cb349..926e703c5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java
@@ -53,6 +53,7 @@ public class PrintingResultsListener implements UserResultsListener {
   @Override
   public void submissionFailed(RpcException ex) {
     exception = ex;
+    System.out.println("Exception (no rows returned): " + ex );
     latch.countDown();
   }
 
@@ -87,7 +88,7 @@ public class PrintingResultsListener implements UserResultsListener {
     if (isLastChunk) {
       allocator.close();
       latch.countDown();
-      System.out.println("Total rows returned : " + count.get());
+      System.out.println("Total rows returned: " + count.get());
     }
 
   }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java
index e1ad85492..3565bf470 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java
@@ -203,7 +203,7 @@ public class ExpressionTreeMaterializer {
       List<LogicalExpression> args = Lists.newArrayList();
       for (int i = 0; i < call.args.size(); ++i) {
         LogicalExpression newExpr = call.args.get(i).accept(this, registry);
-        assert newExpr != null : String.format("Materialization of %s return a null expression.", call.args.get(i));
+        assert newExpr != null : String.format("Materialization of %s returned a null expression.", call.args.get(i));
         args.add(newExpr);
       }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/annotations/FunctionTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/annotations/FunctionTemplate.java
index 1f732a398..15c7e196d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/annotations/FunctionTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/annotations/FunctionTemplate.java
@@ -28,15 +28,27 @@ import java.lang.annotation.Target;
 public @interface FunctionTemplate {
 
   /**
-   * Use this annotation if there is only one name for function
-   * Note: If you use this annotation don't use {@link #names()}
+   * Name of function (when only one.)
+   * Use this annotation element if there is only one name for the function.
+   * Note: If you use this annotation don't use {@link #names()}.
+   * <p>
+   *   TODO:  Refer to wherever list of possible or at least known names is,
+   *   to resolve the current issue of spaces vs. underlines in names (e.g., we
+   *   have both "less_than" and "less than".
+   * </p>
    * @return
    */
   String name() default "";
 
   /**
-   * Use this annotation if there are multiple names for function
-   * Note: If you use this annotation don't use {@link #name()}
+   * Names of function (when multiple).
+   * Use this annotation element if there are multiple names for the function.
+   * Note: If you use this annotation don't use {@link #name()}.
+   * <p>
+   *   TODO:  Refer to wherever list of possible or at least known names is,
+   *   to resolve the current issue of spaces vs. underlines in names (e.g., we
+   *   have both "less_than" and "less than".
+   * </p>
    * @return
    */
   String[] names() default {};
@@ -49,10 +61,21 @@ public @interface FunctionTemplate {
   FunctionCostCategory costCategory() default FunctionCostCategory.SIMPLE;
 
   public static enum NullHandling {
-    INTERNAL, NULL_IF_NULL;
+    /**
+     * Method handles nulls.
+     */
+    INTERNAL,
+
+    /**
+     * Null output if any null input:
+     * Indicates that a method's associated logical operation returns NULL if
+     * either input is NULL, and therefore that the method must not be called
+     * with null inputs.  (The calling framework must handle NULLs.)
+     */
+    NULL_IF_NULL;
   }
 
-  public static enum FunctionScope{
+  public static enum FunctionScope {
     SIMPLE,
     POINT_AGGREGATE,
     DECIMAL_AGGREGATE,
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionGenerationHelper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionGenerationHelper.java
index d007d7c54..19cd1d851 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionGenerationHelper.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionGenerationHelper.java
@@ -29,28 +29,58 @@ import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.common.types.Types;
 import org.apache.drill.exec.expr.ClassGenerator.HoldingContainer;
 import org.apache.drill.exec.expr.HoldingContainerExpression;
+import org.eigenbase.rel.RelFieldCollation.NullDirection;
 
 public class FunctionGenerationHelper {
-  public static final String COMPARE_TO = "compare_to";
+  public static final String COMPARE_TO_NULLS_HIGH = "compare_to_nulls_high";
+  public static final String COMPARE_TO_NULLS_LOW = "compare_to_nulls_low";
 
   /**
-   * Given materialized arguments find the "compare_to" FunctionHolderExpression
-   * @param left
-   * @param right
-   * @param registry
-   * @return FunctionHolderExpression containing the function implementation
+   * Finds ordering comparator ("compare_to...") FunctionHolderExpression with
+   * a specified ordering for NULL (and considering NULLS <i>equal</i>).
+   * @param  null_high  whether NULL should compare as the lowest value (if
+   *                    {@code false}) or the highest value (if {@code true})
+   * @param  left  ...
+   * @param  right  ...
+   * @param  registry  ...
+   * @return
+   *     FunctionHolderExpression containing the found function implementation
    */
-  public static FunctionHolderExpression getComparator(HoldingContainer left,
-    HoldingContainer right,
-    FunctionImplementationRegistry registry) {
-    if (! isComparableType(left.getMajorType()) || ! isComparableType(right.getMajorType()) ){
-      throw new UnsupportedOperationException(formatCanNotCompareMsg(left.getMajorType(), right.getMajorType()));
+  public static FunctionHolderExpression getOrderingComparator(
+      boolean null_high,
+      HoldingContainer left,
+      HoldingContainer right,
+      FunctionImplementationRegistry registry) {
+    final String comparator_name =
+        null_high ? COMPARE_TO_NULLS_HIGH : COMPARE_TO_NULLS_LOW;
+
+    if (   ! isComparableType(left.getMajorType() )
+        || ! isComparableType(right.getMajorType() ) ) {
+      throw new UnsupportedOperationException(
+          formatCanNotCompareMsg(left.getMajorType(), right.getMajorType()));
     }
+    return getFunctionExpression(comparator_name, Types.required(MinorType.INT),
+                                 registry, left, right);
+  }
 
-    return getFunctionExpression(COMPARE_TO, Types.required(MinorType.INT), registry, left, right);
+  /**
+   * Finds ordering comparator ("compare_to...") FunctionHolderExpression with
+   * a "NULL high" ordering (and considering NULLS <i>equal</i>).
+   * @param  left  ...
+   * @param  right  ...
+   * @param  registry  ...
+   * @return FunctionHolderExpression containing the function implementation
+   * @see #getComparator
+   */
+  public static FunctionHolderExpression getOrderingComparatorNullsHigh(
+      HoldingContainer left,
+      HoldingContainer right,
+      FunctionImplementationRegistry registry) {
+    return getOrderingComparator(true, left, right, registry);
   }
 
-  public static FunctionHolderExpression getFunctionExpression(String name, MajorType returnType, FunctionImplementationRegistry registry, HoldingContainer... args) {
+  public static FunctionHolderExpression getFunctionExpression(
+      String name, MajorType returnType, FunctionImplementationRegistry registry, HoldingContainer... args) {
     List<MajorType> argTypes = new ArrayList<MajorType>(args.length);
     List<LogicalExpression> argExpressions = new ArrayList<LogicalExpression>(args.length);
     for(HoldingContainer c : args) {
@@ -69,10 +99,10 @@ public class FunctionGenerationHelper {
     sb.append("( ");
     for(int i =0; i < args.length; i++) {
       MajorType mt = args[i].getMajorType();
-      appendType(mt, sb);
       if (i != 0) {
         sb.append(", ");
       }
+      appendType(mt, sb);
     }
     sb.append(" ) returns ");
     appendType(returnType, sb);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionImplementationRegistry.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionImplementationRegistry.java
index 25dcbbcbe..55e4121d3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionImplementationRegistry.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionImplementationRegistry.java
@@ -99,11 +99,15 @@ public class FunctionImplementationRegistry {
     return functionResolver.getBestMatch(drillFuncRegistry.getMethods(functionReplacement(functionCall)), functionCall);
   }
 
-  // Check if this Function Replacement is Needed; if yes, return a new name. otherwise, return the original name
+  // Check if this Function Replacement is needed; if yes, return a new name. otherwise, return the original name
   private String functionReplacement(FunctionCall functionCall) {
     String funcName = functionCall.getName();
-    if(optionManager != null && optionManager.getOption(ExecConstants.CAST_TO_NULLABLE_NUMERIC).bool_val && CastFunctions.isReplacementNeeded(functionCall.args.get(0).getMajorType().getMinorType(), funcName)) {
-      org.apache.drill.common.types.TypeProtos.DataMode dataMode = functionCall.args.get(0).getMajorType().getMode();
+    if (optionManager != null
+        && optionManager.getOption(ExecConstants.CAST_TO_NULLABLE_NUMERIC).bool_val
+        && CastFunctions.isReplacementNeeded(functionCall.args.get(0).getMajorType().getMinorType(),
+                                             funcName)) {
+      org.apache.drill.common.types.TypeProtos.DataMode dataMode =
+          functionCall.args.get(0).getMajorType().getMode();
       funcName = CastFunctions.getReplacingCastFunction(funcName, dataMode);
     }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/BitFunctions.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/BitFunctions.java
index 3fe489fd9..ae428726a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/BitFunctions.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/BitFunctions.java
@@ -27,9 +27,17 @@ import org.apache.drill.exec.expr.holders.BitHolder;
 import org.apache.drill.exec.expr.holders.IntHolder;
 import org.apache.drill.exec.record.RecordBatch;
 
+/**
+ * Function templates for Bit/BOOLEAN functions other than comparison
+ * functions.  (Bit/BOOLEAN comparison functions are generated in
+ * ComparisonFunctions.java template.)
+ *
+ */
 public class BitFunctions {
 
-  @FunctionTemplate(names = {"booleanOr", "or", "||"}, scope = FunctionScope.SC_BOOLEAN_OPERATOR, nulls = NullHandling.NULL_IF_NULL)
+  @FunctionTemplate(names = {"booleanOr", "or", "||"},
+                    scope = FunctionScope.SC_BOOLEAN_OPERATOR,
+                    nulls = NullHandling.NULL_IF_NULL)
   public static class BitOr implements DrillSimpleFunc {
 
     @Param BitHolder left;
@@ -43,7 +51,9 @@ public class BitFunctions {
     }
   }
 
-  @FunctionTemplate(names = {"booleanAnd", "and", "&&"}, scope = FunctionScope.SC_BOOLEAN_OPERATOR, nulls = NullHandling.NULL_IF_NULL)
+  @FunctionTemplate(names = {"booleanAnd", "and", "&&"},
+                    scope = FunctionScope.SC_BOOLEAN_OPERATOR,
+                    nulls = NullHandling.NULL_IF_NULL)
   public static class BitAnd implements DrillSimpleFunc {
 
     @Param BitHolder left;
@@ -58,7 +68,9 @@ public class BitFunctions {
   }
 
 
-  @FunctionTemplate(names = {"xor", "^"}, scope = FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
+  @FunctionTemplate(names = {"xor", "^"},
+                    scope = FunctionScope.SIMPLE,
+                    nulls = NullHandling.NULL_IF_NULL)
   public static class IntXor implements DrillSimpleFunc {
 
     @Param IntHolder left;
@@ -72,32 +84,4 @@ public class BitFunctions {
     }
   }
 
-  @FunctionTemplate(names = {"equal","==","="}, scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class EqualsBitBit implements DrillSimpleFunc {
-
-    @Param BitHolder left;
-    @Param BitHolder right;
-    @Output BitHolder out;
-
-    public void setup(RecordBatch b) {}
-
-    public void eval() {
-      out.value = left.value == right.value ? 1 : 0;
-
-    }
-  }
-
-  @FunctionTemplate(name = "compare_to", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
-  public static class CompareBitBit implements DrillSimpleFunc {
-
-    @Param BitHolder left;
-    @Param BitHolder right;
-    @Output IntHolder out;
-
-    public void setup(RecordBatch b) {}
-
-    public void eval() {
-      out.value = left.value < right.value ? -1 : ((left.value == right.value)? 0 : 1);
-    }
-  }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ComparisonFunctions.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ComparisonFunctions.java
deleted file mode 100644
index bf42ce6c0..000000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ComparisonFunctions.java
+++ /dev/null
@@ -1,572 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.drill.exec.expr.fn.impl;
-
-
-public class ComparisonFunctions {
-    static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MathFunctions.class);
-
-//    private ComparisonFunctions() {}
-//
-//    @FunctionTemplate(name = "equal", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class IntEqual implements DrillSimpleFunc {
-//
-//        @Param IntHolder left;
-//        @Param IntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value == right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "equal", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class BigIntEqual implements DrillSimpleFunc {
-//
-//        @Param BigIntHolder left;
-//        @Param BigIntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value == right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "equal", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float4Equal implements DrillSimpleFunc {
-//
-//        @Param Float4Holder left;
-//        @Param Float4Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value == right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "equal", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float8Equal implements DrillSimpleFunc {
-//
-//        @Param Float8Holder left;
-//        @Param Float8Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value == right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "not equal", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class IntNotEqual implements DrillSimpleFunc {
-//
-//        @Param IntHolder left;
-//        @Param IntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value != right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "not equal", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class BigIntNotEqual implements DrillSimpleFunc {
-//
-//        @Param BigIntHolder left;
-//        @Param BigIntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value != right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "not equal", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float4NotEqual implements DrillSimpleFunc {
-//
-//        @Param Float4Holder left;
-//        @Param Float4Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value != right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "not equal", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float8NotEqual implements DrillSimpleFunc {
-//
-//        @Param Float8Holder left;
-//        @Param Float8Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value != right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class IntGreaterThan implements DrillSimpleFunc {
-//
-//        @Param IntHolder left;
-//        @Param IntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value > right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class BigIntGreaterThan implements DrillSimpleFunc {
-//
-//        @Param BigIntHolder left;
-//        @Param BigIntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value > right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float4GreaterThan implements DrillSimpleFunc {
-//
-//        @Param Float4Holder left;
-//        @Param Float4Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value > right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float8GreaterThan implements DrillSimpleFunc {
-//
-//        @Param Float8Holder left;
-//        @Param Float8Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value > right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class IntGreaterThanEqual implements DrillSimpleFunc {
-//
-//        @Param IntHolder left;
-//        @Param IntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value >= right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class BigIntGreaterThanEqual implements DrillSimpleFunc {
-//
-//        @Param BigIntHolder left;
-//        @Param BigIntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value >= right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float4GreaterThanEqual implements DrillSimpleFunc {
-//
-//        @Param Float4Holder left;
-//        @Param Float4Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value >= right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float8GreaterThanEqual implements DrillSimpleFunc {
-//
-//        @Param Float8Holder left;
-//        @Param Float8Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value >= right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class IntLessThan implements DrillSimpleFunc {
-//
-//        @Param IntHolder left;
-//        @Param IntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value < right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class BigIntLessThan implements DrillSimpleFunc {
-//
-//        @Param BigIntHolder left;
-//        @Param BigIntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value < right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float4LessThan implements DrillSimpleFunc {
-//
-//        @Param Float4Holder left;
-//        @Param Float4Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value < right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float8LessThan implements DrillSimpleFunc {
-//
-//        @Param Float8Holder left;
-//        @Param Float8Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value < right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class IntLessThanEqual implements DrillSimpleFunc {
-//
-//        @Param IntHolder left;
-//        @Param IntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value <= right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class BigIntLessThanEqual implements DrillSimpleFunc {
-//
-//        @Param BigIntHolder left;
-//        @Param BigIntHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value <= right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float4LessThanEqual implements DrillSimpleFunc {
-//
-//        @Param Float4Holder left;
-//        @Param Float4Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value <= right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class Float8LessThanEqual implements DrillSimpleFunc {
-//
-//        @Param Float8Holder left;
-//        @Param Float8Holder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = (left.value <= right.value) ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "equal", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarBinaryEqual implements DrillSimpleFunc {
-//
-//        @Param VarBinaryHolder left;
-//        @Param VarBinaryHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) == 0 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "equal", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarCharEqual implements DrillSimpleFunc {
-//
-//        @Param VarCharHolder left;
-//        @Param VarCharHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) == 0 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "equal", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarBinaryCharEqual implements DrillSimpleFunc {
-//
-//        @Param VarBinaryHolder left;
-//        @Param VarCharHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) == 0 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarBinaryGTE implements DrillSimpleFunc {
-//
-//        @Param VarBinaryHolder left;
-//        @Param VarBinaryHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) > -1 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarCharGTE implements DrillSimpleFunc {
-//
-//        @Param VarCharHolder left;
-//        @Param VarCharHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) > -1 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarBinaryCharGTE implements DrillSimpleFunc {
-//
-//        @Param VarBinaryHolder left;
-//        @Param VarCharHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) > -1 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarBinaryGT implements DrillSimpleFunc {
-//
-//        @Param VarBinaryHolder left;
-//        @Param VarBinaryHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) == 1 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarCharGT implements DrillSimpleFunc {
-//
-//        @Param VarCharHolder left;
-//        @Param VarCharHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) == 1 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "greater than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarBinaryCharGT implements DrillSimpleFunc {
-//
-//        @Param VarBinaryHolder left;
-//        @Param VarCharHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) == 1? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarBinaryLTE implements DrillSimpleFunc {
-//
-//        @Param VarBinaryHolder left;
-//        @Param VarBinaryHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) < 1 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarCharLTE implements DrillSimpleFunc {
-//
-//        @Param VarCharHolder left;
-//        @Param VarCharHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) < 1 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than or equal to", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarBinaryCharLTE implements DrillSimpleFunc {
-//
-//        @Param VarBinaryHolder left;
-//        @Param VarCharHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) < 1? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarBinaryLT implements DrillSimpleFunc {
-//
-//        @Param VarBinaryHolder left;
-//        @Param VarBinaryHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) == -1 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarCharLT implements DrillSimpleFunc {
-//
-//        @Param VarCharHolder left;
-//        @Param VarCharHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) == -1 ? 1 : 0;
-//        }
-//    }
-//
-//    @FunctionTemplate(name = "less than", scope = FunctionTemplate.FunctionScope.SIMPLE)
-//    public static class VarBinaryCharLT implements DrillSimpleFunc {
-//
-//        @Param VarBinaryHolder left;
-//        @Param VarCharHolder right;
-//        @Output BitHolder out;
-//
-//        public void setup(RecordBatch b) {}
-//
-//        public void eval() {
-//            out.value = org.apache.drill.exec.expr.fn.impl.VarHelpers.compare(left, right) == -1 ? 1 : 0;
-//        }
-//    }
-
-}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ComparisonFunctionsNullable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ComparisonFunctionsNullable.java
deleted file mode 100644
index 570aaeb05..000000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ComparisonFunctionsNullable.java
+++ /dev/null
@@ -1,250 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.drill.exec.expr.fn.impl;
-
-import org.apache.drill.exec.expr.DrillSimpleFunc;
-import org.apache.drill.exec.expr.annotations.FunctionTemplate;
-import org.apache.drill.exec.expr.annotations.Output;
-import org.apache.drill.exec.expr.annotations.Param;
-import org.apache.drill.exec.expr.holders.BitHolder;
-import org.apache.drill.exec.expr.holders.NullableBigIntHolder;
-import org.apache.drill.exec.expr.holders.NullableIntHolder;
-import org.apache.drill.exec.record.RecordBatch;
-
-public class ComparisonFunctionsNullable {
-    static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ComparisonFunctionsNullable.class);
-
-    private ComparisonFunctionsNullable() {}
-
-    @FunctionTemplate(names = {"equal", "=", "=="}, scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableIntEqual implements DrillSimpleFunc {
-
-        @Param NullableIntHolder left;
-        @Param NullableIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value == right.value) ? 1 : 0;
-            }
-        }
-    }
-
-    @FunctionTemplate(names = {"equal", "=", "=="}, scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableBigIntEqual implements DrillSimpleFunc {
-
-        @Param NullableBigIntHolder left;
-        @Param NullableBigIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value == right.value) ? 1 : 0;
-            }
-        }
-    }
-
-    @FunctionTemplate(names = {"not equal", "<>", "!="}, scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableIntNotEqual implements DrillSimpleFunc {
-
-        @Param NullableIntHolder left;
-        @Param NullableIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value != right.value) ? 1 : 0;
-            }
-        }
-    }
-
-    @FunctionTemplate(names = {"not equal", "<>", "!="}, scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableBigIntNotEqual implements DrillSimpleFunc {
-
-        @Param NullableBigIntHolder left;
-        @Param NullableBigIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value != right.value) ? 1 : 0;
-            }
-        }
-    }
-
-    @FunctionTemplate(names = {"less than","<"}, scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableIntLessThan implements DrillSimpleFunc {
-
-        @Param NullableIntHolder left;
-        @Param NullableIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value < right.value) ? 1 : 0;
-            }
-        }
-    }
-
-    @FunctionTemplate(names = {"less than","<"}, scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableBigIntLessThan implements DrillSimpleFunc {
-
-        @Param NullableBigIntHolder left;
-        @Param NullableBigIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value < right.value) ? 1 : 0;
-            }
-        }
-    }
-
-    @FunctionTemplate(names = {"less than or equal to","<="}, scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableIntLessThanEqual implements DrillSimpleFunc {
-
-        @Param NullableIntHolder left;
-        @Param NullableIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value <= right.value) ? 1 : 0;
-            }
-        }
-    }
-
-    @FunctionTemplate(names = {"less than or equal to","<="}, scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableBigIntLessThanEqual implements DrillSimpleFunc {
-
-        @Param NullableBigIntHolder left;
-        @Param NullableBigIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value <= right.value) ? 1 : 0;
-            }
-        }
-    }
-
-    @FunctionTemplate(names = {"greater than",">"}, scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableIntGreaterThan implements DrillSimpleFunc {
-
-        @Param NullableIntHolder left;
-        @Param NullableIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value > right.value) ? 1 : 0;
-            }
-        }
-    }
-
-    @FunctionTemplate(names = {"greater than",">"}, scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableBigIntGreaterThan implements DrillSimpleFunc {
-
-        @Param NullableBigIntHolder left;
-        @Param NullableBigIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value > right.value) ? 1 : 0;
-            }
-        }
-    }
-
-    @FunctionTemplate(names = {"greater than or equal to",">="}, scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableIntGreaterThanEqual implements DrillSimpleFunc {
-
-        @Param NullableIntHolder left;
-        @Param NullableIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value >= right.value) ? 1 : 0;
-            }
-        }
-    }
-
-    @FunctionTemplate(names = {"greater than or equal to",">="},
-                      scope = FunctionTemplate.FunctionScope.SIMPLE)
-    public static class NullableBigIntGreaterThanEqual implements DrillSimpleFunc {
-
-        @Param NullableBigIntHolder left;
-        @Param NullableBigIntHolder right;
-        @Output BitHolder out;
-
-        public void setup(RecordBatch b) {}
-
-        public void eval() {
-            if (left.isSet  == 0 || right.isSet == 0) {
-                out.value = 0;
-            } else {
-                out.value = (left.value >= right.value) ? 1 : 0;
-            }
-        }
-    }
-}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/TopN/TopNBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/TopN/TopNBatch.java
index 9829fc616..a73bdad9f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/TopN/TopNBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/TopN/TopNBatch.java
@@ -61,6 +61,7 @@ import org.apache.drill.exec.record.selection.SelectionVector4;
 import org.apache.drill.exec.vector.ValueVector;
 import org.apache.drill.exec.vector.complex.AbstractContainerVector;
 import org.eigenbase.rel.RelFieldCollation.Direction;
+import org.eigenbase.rel.RelFieldCollation.NullDirection;
 
 import com.google.common.base.Stopwatch;
 import com.google.common.collect.Lists;
@@ -324,7 +325,9 @@ public class TopNBatch extends AbstractRecordBatch<TopN> {
       g.setMappingSet(mainMapping);
 
       // next we wrap the two comparison sides and add the expression block for the comparison.
-      LogicalExpression fh = FunctionGenerationHelper.getComparator(left, right, context.getFunctionRegistry());
+      LogicalExpression fh =
+        FunctionGenerationHelper.getOrderingComparator(od.nullsSortHigh(), left, right,
+                                                       context.getFunctionRegistry());
       HoldingContainer out = g.addExpr(fh, false);
       JConditional jc = g.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java
index 860627d9c..4acf3d6de 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java
@@ -248,7 +248,9 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
       cg.setMappingSet(IS_SAME_I2);
       HoldingContainer second = cg.addExpr(expr, false);
 
-      LogicalExpression fh = FunctionGenerationHelper.getComparator(first, second, context.getFunctionRegistry());
+      LogicalExpression fh =
+          FunctionGenerationHelper
+          .getOrderingComparatorNullsHigh(first, second, context.getFunctionRegistry());
       HoldingContainer out = cg.addExpr(fh, false);
       cg.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)))._then()._return(JExpr.FALSE);
     }
@@ -269,7 +271,9 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
       cg.setMappingSet(ISA_B2);
       HoldingContainer second = cg.addExpr(expr, false);
 
-      LogicalExpression fh = FunctionGenerationHelper.getComparator(first, second, context.getFunctionRegistry());
+      LogicalExpression fh =
+          FunctionGenerationHelper
+          .getOrderingComparatorNullsHigh(first, second, context.getFunctionRegistry());
       HoldingContainer out = cg.addExpr(fh, false);
       cg.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)))._then()._return(JExpr.FALSE);
     }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java
index ea196457f..8f921f7f8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java
@@ -250,8 +250,10 @@ public class ChainedHashTable {
         jc._then()._return(JExpr.FALSE);
       }
 
-      // next we wrap the two comparison sides and add the expression block for the comparison.
-      LogicalExpression f = FunctionGenerationHelper.getComparator(left, right, context.getFunctionRegistry());
+      final LogicalExpression f =
+          FunctionGenerationHelper
+          .getOrderingComparatorNullsHigh(left, right, context.getFunctionRegistry());
+
       HoldingContainer out = cg.addExpr(f, false);
 
       // check if two values are not equal (comparator result != 0)
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java
index a7fa5aac8..8dfc8f1ab 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java
@@ -72,6 +72,29 @@ public final class JoinStatus {
     this.joinType = output.getJoinType();
   }
 
+  @Override
+  public String toString() {
+    return
+        super.toString()
+        + "["
+        + "leftPosition = " + leftPosition
+        + ", rightPosition = " + rightPosition
+        + ", svRightPosition = " + svRightPosition
+        + ", outputPosition = " + outputPosition
+        + ", lastLeft = " + lastLeft
+        + ", lastRight = " + lastRight
+        + ", rightSourceMode = " + rightSourceMode
+        + ", sv4 = " + sv4
+        + ", joinType = " + joinType
+        + ", ok = " + ok
+        + ", initialSet = " + initialSet
+        + ", leftRepeating = " + leftRepeating
+        + ", left = " + left
+        + ", right = " + right
+        + ", outputBatch = " + outputBatch
+        + "]";
+  }
+
   private final IterOutcome nextLeft() {
     return outputBatch.next(LEFT_INPUT, left);
   }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinTemplate.java
index 48a0996c5..4124e6db5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinTemplate.java
@@ -155,7 +155,9 @@ public abstract class JoinTemplate implements JoinWorker {
           }
           status.advanceRight();
 
-        } while ((!status.isLeftRepeating() || status.isRightPositionInCurrentBatch()) && status.isRightPositionAllowed() && doCompare(status.getLeftPosition(), status.getRightPosition()) == 0);
+        } while ((!status.isLeftRepeating() || status.isRightPositionInCurrentBatch())
+                 && status.isRightPositionAllowed()
+                 && doCompare(status.getLeftPosition(), status.getRightPosition()) == 0);
 
         if (status.getRightPosition() > initialRightPosition &&
             (status.isLeftRepeating() || ! status.isNextLeftPositionInCurrentBatch())) {
@@ -204,8 +206,8 @@ public abstract class JoinTemplate implements JoinWorker {
   /**
    * Copy the data to the new record batch (if it fits).
    *
-   * @param leftIndex  position of batch (lower 16 bits) and record (upper 16 bits) in left SV4
-   * @param outIndex position of the output record batch
+   * @param  leftIndex   position of batch (lower 16 bits) and record (upper 16 bits) in left SV4
+   * @param  outIndex  position of the output record batch
    * @return Whether or not the data was copied.
    */
   public abstract void doCopyLeft(@Named("leftIndex") int leftIndex, @Named("outIndex") int outIndex);
@@ -228,8 +230,8 @@ public abstract class JoinTemplate implements JoinWorker {
 
   /**
    * Compare the current left key to the next left key, if it's within the batch.
-   * @return  0 if both keys are equal
-   *          1 if the keys are not equal
+   * @return  0 if both keys are equal,
+   *          1 if the keys are not equal, and
    *         -1 if there are no more keys in this batch
    */
   protected abstract int doCompareNextLeftKey(@Named("leftIndex") int leftIndex);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java
index 257b93e9e..8fce52ec7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java
@@ -313,12 +313,13 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
 
       // check value equality
 
-      LogicalExpression gh = FunctionGenerationHelper.getComparator(compareThisLeftExprHolder,
-        compareNextLeftExprHolder,
-        context.getFunctionRegistry());
+      LogicalExpression gh =
+          FunctionGenerationHelper.getOrderingComparatorNullsHigh(compareThisLeftExprHolder,
+                                                                  compareNextLeftExprHolder,
+                                                                  context.getFunctionRegistry());
       HoldingContainer out = cg.addExpr(gh, false);
 
-      //If not 0, it means not equal. We return this out value.
+      // If not 0, it means not equal. We return this out value.
       JConditional jc = cg.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
       jc._then()._return(out.getValue());
     }
@@ -523,12 +524,13 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
         cg.getSetupBlock().assign(JExpr._this().ref(incomingRecordBatch), JExpr._this().ref(incomingRightRecordBatch));
         ClassGenerator.HoldingContainer compareRightExprHolder = cg.addExpr(materializedRightExpr, false);
 
-        LogicalExpression fh = FunctionGenerationHelper.getComparator(compareLeftExprHolder,
-          compareRightExprHolder,
-          context.getFunctionRegistry());
+        LogicalExpression fh =
+            FunctionGenerationHelper.getOrderingComparatorNullsHigh(compareLeftExprHolder,
+                                                                    compareRightExprHolder,
+                                                                    context.getFunctionRegistry());
         HoldingContainer out = cg.addExpr(fh, false);
 
-        // If not 0, it means not equal. We return this out value.
+        // If not 0, it means not equal.
         // Null compares to Null should returns null (unknown). In such case, we return 1 to indicate they are not equal.
         if (compareLeftExprHolder.isOptional() && compareRightExprHolder.isOptional()
             && ! areNullsEqual) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
index d78ba8ed4..804671e56 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
@@ -76,6 +76,7 @@ import org.apache.drill.exec.vector.CopyUtil;
 import org.apache.drill.exec.vector.FixedWidthVector;
 import org.apache.drill.exec.vector.ValueVector;
 import org.eigenbase.rel.RelFieldCollation.Direction;
+import org.eigenbase.rel.RelFieldCollation.NullDirection;
 
 import parquet.Preconditions;
 
@@ -636,7 +637,9 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
       g.setMappingSet(MAIN_MAPPING);
 
       // next we wrap the two comparison sides and add the expression block for the comparison.
-      LogicalExpression fh = FunctionGenerationHelper.getComparator(left, right, context.getFunctionRegistry());
+      LogicalExpression fh =
+          FunctionGenerationHelper.getOrderingComparator(od.nullsSortHigh(), left, right,
+                                                         context.getFunctionRegistry());
       HoldingContainer out = g.addExpr(fh, false);
       JConditional jc = g.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
index a06207405..352e7aedf 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
@@ -70,6 +70,7 @@ import org.apache.drill.exec.vector.AllocationHelper;
 import org.apache.drill.exec.vector.IntVector;
 import org.apache.drill.exec.vector.ValueVector;
 import org.eigenbase.rel.RelFieldCollation.Direction;
+import org.eigenbase.rel.RelFieldCollation.NullDirection;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
@@ -574,7 +575,10 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
           new ValueVectorReadExpression(new TypedFieldId(expr.getMajorType(), count++)), false);
       cg.setMappingSet(mainMapping);
 
-      LogicalExpression fh = FunctionGenerationHelper.getComparator(left, right, context.getFunctionRegistry());
+      // next we wrap the two comparison sides and add the expression block for the comparison.
+      LogicalExpression fh =
+          FunctionGenerationHelper.getOrderingComparator(od.nullsSortHigh(), left, right,
+                                                         context.getFunctionRegistry());
       ClassGenerator.HoldingContainer out = cg.addExpr(fh, false);
       JConditional jc = cg.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatch.java
index 19f542302..6f19e91d5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatch.java
@@ -43,6 +43,7 @@ import org.apache.drill.exec.record.WritableBatch;
 import org.apache.drill.exec.record.selection.SelectionVector2;
 import org.apache.drill.exec.record.selection.SelectionVector4;
 import org.eigenbase.rel.RelFieldCollation.Direction;
+import org.eigenbase.rel.RelFieldCollation.NullDirection;
 
 import com.sun.codemodel.JConditional;
 import com.sun.codemodel.JExpr;
@@ -180,7 +181,9 @@ public class SortBatch extends AbstractRecordBatch<Sort> {
       g.setMappingSet(mainMapping);
 
       // next we wrap the two comparison sides and add the expression block for the comparison.
-      LogicalExpression fh = FunctionGenerationHelper.getComparator(left, right, context.getFunctionRegistry());
+      LogicalExpression fh =
+        FunctionGenerationHelper.getOrderingComparator(od.nullsSortHigh(), left, right,
+                                                       context.getFunctionRegistry());
       HoldingContainer out = g.addExpr(fh, false);
       JConditional jc = g.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/StreamingWindowFrameRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/StreamingWindowFrameRecordBatch.java
index 26d23f231..87209ebf0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/StreamingWindowFrameRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/StreamingWindowFrameRecordBatch.java
@@ -185,7 +185,9 @@ public class StreamingWindowFrameRecordBatch extends AbstractSingleRecordBatch<W
       cg.setMappingSet(ISA_B2);
       ClassGenerator.HoldingContainer second = cg.addExpr(expr, false);
 
-      LogicalExpression fh = FunctionGenerationHelper.getComparator(first, second, context.getFunctionRegistry());
+      LogicalExpression fh =
+          FunctionGenerationHelper
+          .getOrderingComparatorNullsHigh(first, second, context.getFunctionRegistry());
       ClassGenerator.HoldingContainer out = cg.addExpr(fh, false);
       cg.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)))._then()._return(JExpr.FALSE);
     }
@@ -205,7 +207,9 @@ public class StreamingWindowFrameRecordBatch extends AbstractSingleRecordBatch<W
       cg.setMappingSet(IS_SAME_I2);
       ClassGenerator.HoldingContainer second = cg.addExpr(expr, false);
 
-      LogicalExpression fh = FunctionGenerationHelper.getComparator(first, second, context.getFunctionRegistry());
+      LogicalExpression fh =
+          FunctionGenerationHelper
+          .getOrderingComparatorNullsHigh(first, second, context.getFunctionRegistry());
       ClassGenerator.HoldingContainer out = cg.addExpr(fh, false);
       cg.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)))._then()._return(JExpr.FALSE);
     }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
index 902666141..f320bbbba 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
@@ -65,6 +65,7 @@ import org.apache.drill.exec.vector.complex.AbstractContainerVector;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.eigenbase.rel.RelFieldCollation.Direction;
+import org.eigenbase.rel.RelFieldCollation.NullDirection;
 
 import com.google.common.base.Stopwatch;
 import com.google.common.collect.Iterators;
@@ -565,7 +566,9 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
       g.setMappingSet(mainMapping);
 
       // next we wrap the two comparison sides and add the expression block for the comparison.
-      LogicalExpression fh = FunctionGenerationHelper.getComparator(left, right, context.getFunctionRegistry());
+      LogicalExpression fh =
+          FunctionGenerationHelper.getOrderingComparator(od.nullsSortHigh(), left, right,
+                                                         context.getFunctionRegistry());
       HoldingContainer out = g.addExpr(fh, false);
       JConditional jc = g.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
 
@@ -612,7 +615,9 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
       g.setMappingSet(MAIN_MAPPING);
 
       // next we wrap the two comparison sides and add the expression block for the comparison.
-      LogicalExpression fh = FunctionGenerationHelper.getComparator(left, right, context.getFunctionRegistry());
+      LogicalExpression fh =
+          FunctionGenerationHelper.getOrderingComparator(od.nullsSortHigh(), left, right,
+                                                         context.getFunctionRegistry());
       HoldingContainer out = g.addExpr(fh, false);
       JConditional jc = g.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSorter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSorter.java
index e80d30903..d97ffc06e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSorter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSorter.java
@@ -24,6 +24,8 @@ import org.apache.drill.exec.ops.FragmentContext;
 import org.apache.drill.exec.record.VectorContainer;
 import org.apache.drill.exec.record.selection.SelectionVector4;
 
+// TODO:  Doc.:  What's an MSorter?  A sorter for merge join?  something else?
+// (What's the "M" part?  Actually, rename interface to clearer.
 public interface MSorter {
   public void setup(FragmentContext context, BufferAllocator allocator, SelectionVector4 vector4, VectorContainer hyperBatch) throws SchemaChangeException;
   public void sort(VectorContainer container);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrule.java
index 732a34520..72034ed85 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrule.java
@@ -115,7 +115,7 @@ public class ProjectPrule extends Prule {
 
     for ( RelFieldCollation field : src.getFieldCollations()) {
       if (inToOut.containsKey(field.getFieldIndex())) {
-        newFields.add(new RelFieldCollation(inToOut.get(field.getFieldIndex()), field.getDirection()));
+        newFields.add(new RelFieldCollation(inToOut.get(field.getFieldIndex()), field.getDirection(), field.nullDirection));
       }
     }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
index 9015a162c..b07942873 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
@@ -65,7 +65,7 @@ public class QueryResultHandler {
     if (l == null) {
       BufferingListener bl = new BufferingListener();
       l = resultsListener.putIfAbsent(result.getQueryId(), bl);
-      // if we had a succesful insert, use that reference.  Otherwise, just throw away the new bufering listener.
+      // if we had a successful insert, use that reference.  Otherwise, just throw away the new bufering listener.
       if (l == null) {
         l = bl;
       }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestAggNullable.java b/exec/java-exec/src/test/java/org/apache/drill/TestAggNullable.java
index 4c9617bbc..34850ba40 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestAggNullable.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestAggNullable.java
@@ -47,12 +47,12 @@ public class TestAggNullable extends BaseTestQuery{
 
     enableAggr(true, false);
     actualRecordCount = testSql(query1);
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
     expectedRecordCount = 4;
     actualRecordCount = testSql(query2);
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
   }
 
@@ -68,12 +68,12 @@ public class TestAggNullable extends BaseTestQuery{
 
     enableAggr(false, true);
     actualRecordCount = testSql(query1);
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
     expectedRecordCount = 4;
     actualRecordCount = testSql(query2);
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
   }
 
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java b/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java
index 55fb59f04..a54200da2 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java
@@ -369,7 +369,7 @@ public class TestExampleQueries extends BaseTestQuery{
   public void testTopNWithSV2() throws Exception {
     int actualRecordCount = testSql("select N_NATIONKEY from cp.`tpch/nation.parquet` where N_NATIONKEY < 10 order by N_NATIONKEY limit 5");
     int expectedRecordCount = 5;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
   }
 
@@ -382,7 +382,7 @@ public class TestExampleQueries extends BaseTestQuery{
   public void testLikeEscape() throws Exception {
     int actualRecordCount = testSql("select id, name from cp.`jsoninput/specialchar.json` where name like '%#_%' ESCAPE '#'");
     int expectedRecordCount = 1;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
   }
@@ -391,7 +391,7 @@ public class TestExampleQueries extends BaseTestQuery{
   public void testSimilarEscape() throws Exception {
     int actualRecordCount = testSql("select id, name from cp.`jsoninput/specialchar.json` where name similar to '(N|S)%#_%' ESCAPE '#'");
     int expectedRecordCount = 1;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
   }
 
@@ -399,7 +399,7 @@ public class TestExampleQueries extends BaseTestQuery{
   public void testImplicitDownwardCast() throws Exception {
     int actualRecordCount = testSql("select o_totalprice from cp.`tpch/orders.parquet` where o_orderkey=60000 and o_totalprice=299402");
     int expectedRecordCount = 0;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
   }
 
@@ -408,26 +408,26 @@ public class TestExampleQueries extends BaseTestQuery{
     // cast from varchar with unknown length to a fixed length.
     int actualRecordCount = testSql("select first_name from cp.`employee.json` where cast(first_name as varchar(2)) = 'Sh'");
     int expectedRecordCount = 27;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
     // cast from varchar with unknown length to varchar(5), then to varchar(10), then to varchar(2). Should produce the same result as the first query.
     actualRecordCount = testSql("select first_name from cp.`employee.json` where cast(cast(cast(first_name as varchar(5)) as varchar(10)) as varchar(2)) = 'Sh'");
     expectedRecordCount = 27;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
 
     // this long nested cast expression should be essentially equal to substr(), meaning the query should return every row in the table.
     actualRecordCount = testSql("select first_name from cp.`employee.json` where cast(cast(cast(first_name as varchar(5)) as varchar(10)) as varchar(2)) = substr(first_name, 1, 2)");
     expectedRecordCount = 1155;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
     // cast is applied to a column from parquet file.
     actualRecordCount = testSql("select n_name from cp.`tpch/nation.parquet` where cast(n_name as varchar(2)) = 'UN'");
     expectedRecordCount = 2;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
   }
 
@@ -476,28 +476,28 @@ public class TestExampleQueries extends BaseTestQuery{
     // source is JSON
     actualRecordCount = testSql("select EMPID from ( select employee_id as empid from cp.`employee.json` limit 2)");
     expectedRecordCount = 2;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
     actualRecordCount = testSql("select EMPLOYEE_ID from ( select employee_id from cp.`employee.json` where Employee_id is not null limit 2)");
     expectedRecordCount = 2;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
     actualRecordCount = testSql("select x.EMPLOYEE_ID from ( select employee_id from cp.`employee.json` limit 2) X");
     expectedRecordCount = 2;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
     // source is PARQUET
     actualRecordCount = testSql("select NID from ( select n_nationkey as nid from cp.`tpch/nation.parquet`) where NID = 3");
     expectedRecordCount = 1;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
     actualRecordCount = testSql("select x.N_nationkey from ( select n_nationkey from cp.`tpch/nation.parquet`) X where N_NATIONKEY = 3");
     expectedRecordCount = 1;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
       // source is CSV
@@ -505,7 +505,7 @@ public class TestExampleQueries extends BaseTestQuery{
     String query = String.format("select rid, x.name from (select columns[0] as RID, columns[1] as NAME from dfs_test.`%s`) X where X.rid = 2", root);
     actualRecordCount = testSql(query);
     expectedRecordCount = 1;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
 
   }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestInList.java b/exec/java-exec/src/test/java/org/apache/drill/TestInList.java
index 03fbf9713..b6218ec44 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestInList.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestInList.java
@@ -33,7 +33,7 @@ public class TestInList extends BaseTestQuery{
         ")");
 
     int expectedRecordCount = 59;
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
+    assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
   }
 
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestJoinNullable.java b/exec/java-exec/src/test/java/org/apache/drill/TestJoinNullable.java
index c49da6cbc..320a992a2 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestJoinNullable.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestJoinNullable.java
@@ -20,6 +20,7 @@ package org.apache.drill;
 import static org.junit.Assert.assertEquals;
 
 import org.apache.drill.common.util.TestTools;
+import org.junit.Ignore;
 import org.junit.Test;
 
 public class TestJoinNullable extends BaseTestQuery{
@@ -35,105 +36,382 @@ public class TestJoinNullable extends BaseTestQuery{
     test("alter session set `planner.slice_target` = 1");
   }
 
-  @Test  // InnerJoin on nullable cols, HashJoin
+  /** InnerJoin on nullable cols, HashJoin */
+  @Test
   public void testHashInnerJoinOnNullableColumns() throws Exception {
     String query = String.format("select t1.a1, t1.b1, t2.a2, t2.b2 from dfs_test.`%s/jsoninput/nullable1.json` t1, " +
-                   " dfs_test.`%s/jsoninput/nullable2.json` t2 where t1.b1 = t2.b2", TEST_RES_PATH ,TEST_RES_PATH);
-    int actualRecordCount;
-    int expectedRecordCount = 1;
+                   " dfs_test.`%s/jsoninput/nullable2.json` t2 where t1.b1 = t2.b2", TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 1;
 
     enableJoin(true, false);
-    actualRecordCount = testSql(query);
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
-        expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
   }
 
-  @Test // InnerJoin on nullable cols, MergeJoin
-  public void testMergeInnerJoinOnNullableColumns() throws Exception {
-    String query = String.format("select t1.a1, t1.b1, t2.a2, t2.b2 from dfs_test.`%s/jsoninput/nullable1.json` t1, " +
-                   " dfs_test.`%s/jsoninput/nullable2.json` t2 where t1.b1 = t2.b2", TEST_RES_PATH ,TEST_RES_PATH);
-    int actualRecordCount;
-    int expectedRecordCount = 1;
-
-    enableJoin(false, true);
-    actualRecordCount = testSql(query);
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
-        expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
-  }
-
-  @Test // LeftOuterJoin on nullable cols, HashJoin
+  /** LeftOuterJoin on nullable cols, HashJoin */
+  @Test
   public void testHashLOJOnNullableColumns() throws Exception {
     String query = String.format("select t1.a1, t1.b1, t2.a2, t2.b2 from dfs_test.`%s/jsoninput/nullable1.json` t1 " +
                       " left outer join dfs_test.`%s/jsoninput/nullable2.json` t2 " +
-                      " on t1.b1 = t2.b2", TEST_RES_PATH ,TEST_RES_PATH);
+                      " on t1.b1 = t2.b2", TEST_RES_PATH, TEST_RES_PATH);
 
-    int actualRecordCount;
-    int expectedRecordCount = 2;
+    final int expectedRecordCount = 2;
 
     enableJoin(true, false);
-    actualRecordCount = testSql(query);
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
-        expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
   }
 
-  @Test // RightOuterJoin on nullable cols, HashJoin
+  /** RightOuterJoin on nullable cols, HashJoin */
+  @Test
   public void testHashROJOnNullableColumns() throws Exception {
     String query = String.format("select t1.a1, t1.b1, t2.a2, t2.b2 from dfs_test.`%s/jsoninput/nullable1.json` t1 " +
                       " right outer join dfs_test.`%s/jsoninput/nullable2.json` t2 " +
-                      " on t1.b1 = t2.b2", TEST_RES_PATH ,TEST_RES_PATH);
+                      " on t1.b1 = t2.b2", TEST_RES_PATH, TEST_RES_PATH);
 
-    int actualRecordCount;
-    int expectedRecordCount = 4;
+    final int expectedRecordCount = 4;
 
     enableJoin(true, false);
-    actualRecordCount = testSql(query);
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
-        expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
   }
 
-  @Test // FullOuterJoin on nullable cols, HashJoin
+  /** FullOuterJoin on nullable cols, HashJoin */
+  @Test
   public void testHashFOJOnNullableColumns() throws Exception {
     String query = String.format("select t1.a1, t1.b1, t2.a2, t2.b2 from dfs_test.`%s/jsoninput/nullable1.json` t1 " +
                       " full outer join dfs_test.`%s/jsoninput/nullable2.json` t2 " +
-                      " on t1.b1 = t2.b2", TEST_RES_PATH ,TEST_RES_PATH);
+                      " on t1.b1 = t2.b2", TEST_RES_PATH, TEST_RES_PATH);
 
-    int actualRecordCount;
-    int expectedRecordCount = 5;
+    final int expectedRecordCount = +5;
 
     enableJoin(true, false);
-    actualRecordCount = testSql(query);
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
-        expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
   }
 
-  @Test // LeftOuterJoin on nullable cols, MergeJoin
-  public void testMergeLOJOnNullableColumns() throws Exception {
+  /** InnerJoin on nullable cols, MergeJoin */
+  @Test
+  public void testMergeInnerJoinOnNullableColumns() throws Exception {
+    String query =
+        String.format(
+            "select t1.a1, t1.b1, t2.a2, t2.b2 "
+            + "  from dfs_test.`%s/jsoninput/nullable1.json` t1, "
+            + "       dfs_test.`%s/jsoninput/nullable2.json` t2 "
+            + " where t1.b1 = t2.b2", TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 1;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+  /** LeftOuterJoin on nullable cols, MergeJoin */
+  @Test
+  public void testMergeLOJNullable() throws Exception {
     String query = String.format("select t1.a1, t1.b1, t2.a2, t2.b2 from dfs_test.`%s/jsoninput/nullable1.json` t1 " +
                       " left outer join dfs_test.`%s/jsoninput/nullable2.json` t2 " +
-                      " on t1.b1 = t2.b2", TEST_RES_PATH ,TEST_RES_PATH);
+                      " on t1.b1 = t2.b2", TEST_RES_PATH, TEST_RES_PATH);
 
-    int actualRecordCount;
-    int expectedRecordCount = 2;
+    final int expectedRecordCount = 2;
 
     enableJoin(false, true);
-    actualRecordCount = testSql(query);
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
-        expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
   }
 
-  @Test // RightOuterJoin on nullable cols, MergeJoin
+  /** RightOuterJoin on nullable cols, MergeJoin */
+  @Test
   public void testMergeROJOnNullableColumns() throws Exception {
     String query = String.format("select t1.a1, t1.b1, t2.a2, t2.b2 from dfs_test.`%s/jsoninput/nullable1.json` t1 " +
                       " right outer join dfs_test.`%s/jsoninput/nullable2.json` t2 " +
-                      " on t1.b1 = t2.b2", TEST_RES_PATH ,TEST_RES_PATH);
+                      " on t1.b1 = t2.b2", TEST_RES_PATH, TEST_RES_PATH);
+
+    final int expectedRecordCount = 4;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+
+  /** Left outer join, merge, nullable col. - unordered inputs. */
+  @Test
+  public void testMergeLOJNullableNoOrderedInputs() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "FROM               dfs_test.`%s/jsoninput/nullableOrdered1.json` t1 "
+            + "   left outer join dfs_test.`%s/jsoninput/nullableOrdered2.json` t2 "
+            + "      using ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+  /** Left outer join, merge, nullable col. - ordered right, ASC NULLS FIRST (nulls low). */
+  @Test
+  public void testMergeLOJNullableOneOrderedInputAscNullsFirst() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "from         dfs_test.`%s/jsoninput/nullableOrdered1.json` t1 "
+            + "  LEFT OUTER JOIN "
+            + "    ( SELECT key, data "
+            + "        FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` t2 "
+            + "        ORDER BY 1 ASC NULLS FIRST ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+  /** Left outer join, merge, nullable col.  - ordered right, ASC NULLS LAST (nulls high). */
+  @Test
+  public void testMergeLOJNullableOneOrderedInputAscNullsLast() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "FROM         dfs_test.`%s/jsoninput/nullableOrdered1.json` t1 "
+            + "  LEFT OUTER JOIN "
+            + "    ( SELECT key, data "
+            + "        FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` t2 "
+            + "        ORDER BY 1 ASC NULLS LAST ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+  /** Left outer join, merge, nullable col. - ordered right, DESC NULLS FIRST (nulls high). */
+  @Test
+  public void testMergeLOJNullableOneOrderedInputDescNullsFirst() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "FROM         dfs_test.`%s/jsoninput/nullableOrdered1.json` t1 "
+            + "  LEFT OUTER JOIN "
+            + "    ( SELECT key, data "
+            + "        FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` t2 "
+            + "        ORDER BY 1 DESC NULLS FIRST ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+  /** Left outer join, merge, nullable col. - ordered right, DESC NULLS LAST (nulls low). */
+  @Test
+  public void testMergeLOJNullableOneOrderedInputDescNullsLast() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "FROM         dfs_test.`%s/jsoninput/nullableOrdered1.json` t1 "
+            + "  LEFT OUTER JOIN "
+            + "    ( SELECT key, data "
+            + "        FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` t2 "
+            + "        ORDER BY 1 DESC NULLS LAST ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+
+
+  /** Left outer join, merge, nullable col. - ordered inputs, both ASC NULLS FIRST (nulls low). */
+  @Test
+  public void testMergeLOJNullableBothInputsOrderedAscNullsFirstVsAscNullsFirst() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "from ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered1.json` "
+            + "         ORDER BY 1 ASC NULLS FIRST ) t1 "
+            + "  LEFT OUTER JOIN "
+            + "     ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` "
+            + "         ORDER BY 1 ASC NULLS FIRST ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+  /** Left outer join, merge, nullable col. - ordered inputs, different null order. */
+  @Test
+  public void testMergeLOJNullableBothInputsOrderedAscNullsLastVsAscNullsFirst() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "from ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered1.json` "
+            + "         ORDER BY 1 ASC NULLS LAST  ) t1 "
+            + "  LEFT OUTER JOIN "
+            + "     ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` "
+            + "         ORDER BY 1 ASC NULLS FIRST ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+  /** Left outer join, merge, nullable col. - ordered inputs, other different null order. */
+  @Test
+  public void testMergeLOJNullableBothInputsOrderedAscNullsFirstVsAscNullsLast() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "from ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered1.json` "
+            + "         ORDER BY 1 ASC NULLS FIRST ) t1 "
+            + "  LEFT OUTER JOIN "
+            + "     ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` "
+            + "         ORDER BY 1 ASC NULLS LAST  ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+  /** Left outer join, merge, nullable col. - ordered inputs, both ASC NULLS LAST (nulls high) */
+  @Test
+  public void testMergeLOJNullableBothInputsOrderedAscNullsLastVsAscNullsLast() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "from ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered1.json` "
+            + "         ORDER BY 1 ASC NULLS LAST  ) t1 "
+            + "  LEFT OUTER JOIN "
+            + "     ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` "
+            + "         ORDER BY 1 ASC NULLS LAST  ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+  /** Left outer join, merge, nullable col. - ordered inputs, DESC vs. ASC, NULLS
+      FIRST (nulls high vs. nulls low). */
+  @Test
+  public void testMergeLOJNullableBothInputsOrderedDescNullsFirstVsAscNullsFirst() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "from ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered1.json` "
+            + "         ORDER BY 1 DESC NULLS FIRST ) t1 "
+            + "  LEFT OUTER JOIN "
+            + "     ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` "
+            + "         ORDER BY 1 ASC NULLS FIRST ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+  /** Left outer join, merge, nullable col. - ordered inputs, DESC vs. ASC, NULLS
+      LAST vs. FIRST (both nulls low). */
+  @Test
+  public void testMergeLOJNullableBothInputsOrderedDescNullsLastVsAscNullsFirst() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "from ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered1.json` "
+            + "         ORDER BY 1 DESC NULLS LAST  ) t1 "
+            + "  LEFT OUTER JOIN "
+            + "     ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` "
+            + "         ORDER BY 1 ASC NULLS FIRST ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
+
+  /** Left outer join, merge, nullable col. - ordered inputs, DESC vs. ASC, NULLS
+      FIRST vs. LAST (both nulls high). */
+  @Test
+  public void testMergeLOJNullableBothInputsOrderedDescNullsFirstVsAscNullsLast() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "from ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered1.json` "
+            + "         ORDER BY 1 DESC NULLS FIRST ) t1 "
+            + "  LEFT OUTER JOIN "
+            + "     ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` "
+            + "         ORDER BY 1 ASC NULLS LAST  ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
+
+    enableJoin(false, true);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
+  }
 
-    int actualRecordCount;
-    int expectedRecordCount = 4;
+  /** Left outer join, merge, nullable col. - ordered inputs, DESC vs. ASC, NULLS
+      LAST (nulls low vs. nulls high). */
+  @Test
+  public void testMergeLOJNullableBothInputsOrderedDescNullsLastVsAscNullsLast() throws Exception {
+    String query =
+        String.format(
+            "SELECT * "
+            + "from ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered1.json` "
+            + "         ORDER BY 1 DESC NULLS LAST  ) t1 "
+            + "  LEFT OUTER JOIN "
+            + "     ( SELECT key, data "
+            + "         FROM dfs_test.`%s/jsoninput/nullableOrdered2.json` "
+            + "         ORDER BY 1 ASC NULLS LAST  ) t2 "
+            + "    USING ( key )",
+            TEST_RES_PATH, TEST_RES_PATH);
+    final int expectedRecordCount = 6;
 
     enableJoin(false, true);
-    actualRecordCount = testSql(query);
-    assertEquals(String.format("Received unexepcted number of rows in output: expected=%d, received=%s",
-        expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
+    final int actualRecordCount = testSql(query);
+    assertEquals("Number of output rows", expectedRecordCount, actualRecordCount);
   }
 
 }
diff --git a/exec/java-exec/src/test/resources/jsoninput/nullableOrdered1.json b/exec/java-exec/src/test/resources/jsoninput/nullableOrdered1.json
new file mode 100644
index 000000000..135d3a0c7
--- /dev/null
+++ b/exec/java-exec/src/test/resources/jsoninput/nullableOrdered1.json
@@ -0,0 +1,24 @@
+{
+ "key":"A",
+ "data":"L_A_1"
+}
+{
+ "key":"B",
+ "data":"L_B_1"
+}
+{
+ "key":null,
+ "data":"L_null_1"
+}
+{
+ "key":"A",
+ "data":"L_A_2"
+}
+{
+ "key":null,
+ "data":"L_null_2"
+}
+{
+ "key":"B",
+ "data":"L_B_2"
+}
diff --git a/exec/java-exec/src/test/resources/jsoninput/nullableOrdered2.json b/exec/java-exec/src/test/resources/jsoninput/nullableOrdered2.json
new file mode 100644
index 000000000..ebaedd132
--- /dev/null
+++ b/exec/java-exec/src/test/resources/jsoninput/nullableOrdered2.json
@@ -0,0 +1,16 @@
+{
+ "key":null,
+ "data":"R_null_1"
+}
+{
+ "key":"A",
+ "data":"R_A_1"
+}
+{
+ "key":null,
+ "data":"R_null_2"
+}
+{
+ "key":null,
+ "data":"R_null_3"
+}
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillDatabaseMetaData.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillDatabaseMetaData.java
new file mode 100644
index 000000000..ca1648d9b
--- /dev/null
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillDatabaseMetaData.java
@@ -0,0 +1,53 @@
+/**
+ * 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.drill.jdbc;
+
+import net.hydromatic.avatica.AvaticaConnection;
+import net.hydromatic.avatica.AvaticaDatabaseMetaData;
+
+public class DrillDatabaseMetaData extends AvaticaDatabaseMetaData {
+
+  protected DrillDatabaseMetaData( AvaticaConnection connection ) {
+    super( connection );
+  }
+
+
+  // For omitted NULLS FIRST/NULLS HIGH, Drill sort NULL sorts as highest value:
+
+  @Override
+  public boolean nullsAreSortedHigh() {
+    return true;
+  }
+
+  @Override
+  public boolean nullsAreSortedLow() {
+    return false;
+  }
+
+  @Override
+  public boolean nullsAreSortedAtStart() {
+    return false;
+  }
+
+  @Override
+  public boolean nullsAreSortedAtEnd() {
+    return false;
+  }
+
+
+}
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillFactory.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillFactory.java
index 9b9eb7b55..c674f8e6b 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillFactory.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillFactory.java
@@ -38,14 +38,17 @@ public abstract class DrillFactory implements AvaticaFactory {
     this.minor = minor;
   }
 
+  @Override
   public int getJdbcMajorVersion() {
     return major;
   }
 
+  @Override
   public int getJdbcMinorVersion() {
     return minor;
   }
 
+  @Override
   public final AvaticaConnection newConnection(UnregisteredDriver driver, AvaticaFactory factory, String url,
       Properties info)  throws SQLException{
     return newDrillConnection((Driver) driver, (DrillFactory) factory, url, info);
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillJdbc41Factory.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillJdbc41Factory.java
index 7fb0a34eb..4da26e9ef 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillJdbc41Factory.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillJdbc41Factory.java
@@ -57,16 +57,20 @@ public class DrillJdbc41Factory extends DrillFactory {
     return new DrillJdbc41Connection((Driver) driver, factory, url, info);
   }
 
+  @Override
   public DrillJdbc41DatabaseMetaData newDatabaseMetaData(AvaticaConnection connection) {
     return new DrillJdbc41DatabaseMetaData((DrillConnectionImpl) connection);
   }
 
+
+  @Override
   public DrillJdbc41Statement newStatement(AvaticaConnection connection, int resultSetType, int resultSetConcurrency,
       int resultSetHoldability) {
     return new DrillJdbc41Statement((DrillConnectionImpl) connection, resultSetType, resultSetConcurrency,
         resultSetHoldability);
   }
 
+  @Override
   public AvaticaPreparedStatement newPreparedStatement(AvaticaConnection connection,
       AvaticaPrepareResult prepareResult, int resultSetType, int resultSetConcurrency, int resultSetHoldability)
       throws SQLException {
@@ -74,11 +78,13 @@ public class DrillJdbc41Factory extends DrillFactory {
         resultSetType, resultSetConcurrency, resultSetHoldability);
   }
 
+  @Override
   public DrillResultSet newResultSet(AvaticaStatement statement, AvaticaPrepareResult prepareResult, TimeZone timeZone) {
     final ResultSetMetaData metaData = newResultSetMetaData(statement, prepareResult.getColumnList());
     return new DrillResultSet(statement, (DrillPrepareResult) prepareResult, metaData, timeZone);
   }
 
+  @Override
   public ResultSetMetaData newResultSetMetaData(AvaticaStatement statement, List<ColumnMetaData> columnMetaDataList) {
     return new AvaticaResultSetMetaData(statement, null, columnMetaDataList);
   }
@@ -176,7 +182,7 @@ public class DrillJdbc41Factory extends DrillFactory {
     }
   }
 
-  private static class DrillJdbc41DatabaseMetaData extends AvaticaDatabaseMetaData {
+  private static class DrillJdbc41DatabaseMetaData extends DrillDatabaseMetaData {
     DrillJdbc41DatabaseMetaData(DrillConnectionImpl connection) {
       super(connection);
     }
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/JdbcNullOrderingAndGroupingTest.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/JdbcNullOrderingAndGroupingTest.java
new file mode 100644
index 000000000..d14b34467
--- /dev/null
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/JdbcNullOrderingAndGroupingTest.java
@@ -0,0 +1,1290 @@
+/**
+ * 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.drill.jdbc.test;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+
+import static org.junit.Assert.assertThat;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.hamcrest.CoreMatchers;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.*;
+
+import com.google.common.base.Function;
+
+public class JdbcNullOrderingAndGroupingTest extends JdbcTestQueryBase {
+  static final org.slf4j.Logger logger =
+      org.slf4j.LoggerFactory.getLogger(JdbcNullOrderingAndGroupingTest.class);
+
+  // TODO:  Move this to where is covers more tests:  HACK: Disable Jetty
+  // status(?) server so unit tests run (without Maven setup).
+  @BeforeClass
+  public static void setUpClass() {
+    System.setProperty( "drill.exec.http.enabled", "false" );
+  }
+
+
+  @Test
+  public void testNullsOrderInJdbcMetaData() throws SQLException {
+    try (Connection conn =
+             DriverManager.getConnection( "jdbc:drill:zk=local", null ) ) {
+      DatabaseMetaData dbMetaData = conn.getMetaData();
+
+      assertThat( "DatabaseMetadata.nullsAreSortedHigh()",
+                  dbMetaData.nullsAreSortedHigh(), equalTo( true ) );
+
+      assertThat( "DatabaseMetadata.nullsAreSortedLow()",
+                  dbMetaData.nullsAreSortedLow(), equalTo( false ) );
+      assertThat( "DatabaseMetadata.nullsAreSortedAtEnd()",
+                  dbMetaData.nullsAreSortedAtEnd(), equalTo( false ) );
+      assertThat( "DatabaseMetadata.nullsAreSortedAtStart()",
+                  dbMetaData.nullsAreSortedAtStart(), equalTo( false ) );
+    }
+  }
+
+
+
+  ////////////////////
+  // DonutsTopping3:
+
+  @Test
+  public void testOrderDonutsTopping3AscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, tbl.topping[3].type AS topping3 \n" +
+              "FROM cp.`donuts.json` AS tbl \n" +
+              "ORDER BY topping3 ASC NULLS FIRST" )
+        .returns( "id=0005; topping3=null\n" +
+                  "id=0002; topping3=Chocolate\n" +
+                  "id=0003; topping3=Maple\n" +
+                  "id=0001; topping3=Powdered Sugar\n" +
+                  "id=0004; topping3=Powdered Sugar" );
+  }
+
+  @Test
+  public void testOrderDonutsTopping3AscNullsLast() throws Exception {
+     JdbcAssert
+         .withNoDefaultSchema()
+         .sql( "SELECT tbl.id, tbl.topping[3].type AS topping3 \n" +
+               "FROM cp.`donuts.json` AS tbl \n" +
+               "ORDER BY topping3 ASC NULLS LAST" )
+         .returns( "id=0002; topping3=Chocolate\n" +
+                   "id=0003; topping3=Maple\n" +
+                   "id=0001; topping3=Powdered Sugar\n" +
+                   "id=0004; topping3=Powdered Sugar\n" +
+                   "id=0005; topping3=null"
+                   );
+  }
+
+  @Test
+  public void testOrderDonutsTopping3AscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, tbl.topping[3].type AS topping3 \n" +
+              "FROM cp.`donuts.json` AS tbl \n" +
+              "ORDER BY topping3 ASC" )
+        .returns( "id=0002; topping3=Chocolate\n" +
+                  "id=0003; topping3=Maple\n" +
+                  "id=0001; topping3=Powdered Sugar\n" +
+                  "id=0004; topping3=Powdered Sugar\n" +
+                  "id=0005; topping3=null" );
+  }
+
+  @Test
+  public void testOrderDonutsTopping3DescNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, tbl.topping[3].type AS topping3 \n" +
+              "FROM cp.`donuts.json` AS tbl \n" +
+              "ORDER BY topping3 DESC NULLS FIRST" )
+        .returns( "id=0005; topping3=null\n" +
+                  "id=0001; topping3=Powdered Sugar\n" +
+                  "id=0004; topping3=Powdered Sugar\n" +
+                  "id=0003; topping3=Maple\n" +
+                  "id=0002; topping3=Chocolate" );
+  }
+
+  @Test
+  public void testOrderDonutsTopping3DescNullsLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, tbl.topping[3].type AS topping3 \n" +
+              "FROM cp.`donuts.json` AS tbl \n" +
+              "ORDER BY topping3 DESC NULLS LAST" )
+        .returns( "id=0001; topping3=Powdered Sugar\n" +
+                  "id=0004; topping3=Powdered Sugar\n" +
+                  "id=0003; topping3=Maple\n" +
+                  "id=0002; topping3=Chocolate\n" +
+                  "id=0005; topping3=null" );
+  }
+
+  @Test
+  public void testOrderDonutsTopping3DescNullsDefaultFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, tbl.topping[3].type AS topping3 \n" +
+              "FROM cp.`donuts.json` AS tbl \n" +
+              "ORDER BY topping3 DESC" )
+        .returns( "id=0005; topping3=null\n" +
+                  "id=0001; topping3=Powdered Sugar\n" +
+                  "id=0004; topping3=Powdered Sugar\n" +
+                  "id=0003; topping3=Maple\n" +
+                  "id=0002; topping3=Chocolate" );
+  }
+
+  @Test
+  public void testOrderDonutsTopping3DefaultedAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, tbl.topping[3].type AS topping3 \n" +
+              "FROM cp.`donuts.json` AS tbl \n" +
+              "ORDER BY topping3 NULLS FIRST" )
+        .returns( "id=0005; topping3=null\n" +
+                  "id=0002; topping3=Chocolate\n" +
+                  "id=0003; topping3=Maple\n" +
+                  "id=0001; topping3=Powdered Sugar\n" +
+                  "id=0004; topping3=Powdered Sugar" );
+  }
+
+  @Test
+  public void testOrderDonutsTopping3DefaultedAscNullsLast() throws Exception {
+    JdbcAssert.withNoDefaultSchema()
+        .sql( "SELECT tbl.id, tbl.topping[3].type AS topping3 \n" +
+              "FROM cp.`donuts.json` AS tbl \n" +
+              "ORDER BY topping3 NULLS LAST" )
+        .returns( "id=0002; topping3=Chocolate\n" +
+                  "id=0003; topping3=Maple\n" +
+                  "id=0001; topping3=Powdered Sugar\n" +
+                  "id=0004; topping3=Powdered Sugar\n" +
+                  "id=0005; topping3=null" );
+  }
+
+  @Test
+  public void testOrderDonutsTopping3DefaultedAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, tbl.topping[3].type AS topping3 \n" +
+              "FROM cp.`donuts.json` AS tbl \n" +
+              "ORDER BY topping3" )
+        .returns( "id=0002; topping3=Chocolate\n" +
+                  "id=0003; topping3=Maple\n" +
+                  "id=0001; topping3=Powdered Sugar\n" +
+                  "id=0004; topping3=Powdered Sugar\n" +
+                  "id=0005; topping3=null" );
+  }
+
+
+  ////////////////////
+  // type VARCHAR? / VarChar:
+
+  @Test
+  public void testOrderVarCharAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_VarChar AS VARCHAR ) AS as_VARCHAR \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_VARCHAR ASC NULLS FIRST" )
+        .returns( "id=2; as_VARCHAR=null\n" +
+                  "id=3; as_VARCHAR=A\n" +
+                  "id=1; as_VARCHAR=B" );  // TODO: Revisit VARCHAR's acting as VARCHAR(1) in cast
+  }
+
+  @Test
+  public void testOrderVarCharAscNullsLast() throws Exception {
+     JdbcAssert
+         .withNoDefaultSchema()
+         .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_VarChar AS VARCHAR ) AS as_VARCHAR \n" +
+               "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+               "ORDER BY as_VARCHAR ASC NULLS LAST" )
+         .returns( "id=3; as_VARCHAR=A\n" +
+                   "id=1; as_VARCHAR=B\n" +  // TODO: Revisit VARCHAR's acting as VARCHAR(1)
+                   "id=2; as_VARCHAR=null"
+                   );
+  }
+
+  @Test
+  public void testOrderVarCharAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_VarChar AS VARCHAR ) AS as_VARCHAR \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_VARCHAR ASC" )
+        .returns( "id=3; as_VARCHAR=A\n" +
+                  "id=1; as_VARCHAR=B\n" +  // TODO: Revisit VARCHAR's acting as VARCHAR(1)
+                  "id=2; as_VARCHAR=null" );
+  }
+
+  @Test
+  public void testOrderVarCharDescNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_VarChar AS VARCHAR ) AS as_VARCHAR \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_VARCHAR DESC NULLS FIRST" )
+        .returns( "id=2; as_VARCHAR=null\n" +
+                  "id=1; as_VARCHAR=B\n" +  // TODO: Revisit VARCHAR's acting as VARCHAR(1)
+                  "id=3; as_VARCHAR=A" );
+  }
+
+  @Test
+  public void testOrderVarCharDescNullsLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_VarChar AS VARCHAR ) AS as_VARCHAR \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_VARCHAR DESC NULLS LAST" )
+        .returns( "id=1; as_VARCHAR=B\n" +  // TODO: Revisit VARCHAR's acting as VARCHAR(1)
+                  "id=3; as_VARCHAR=A\n" +
+                  "id=2; as_VARCHAR=null" );
+  }
+
+  @Test
+  public void testOrderVarCharDescNullsDefaultFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_VarChar AS VARCHAR ) AS as_VARCHAR \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_VARCHAR DESC" )
+        .returns( "id=2; as_VARCHAR=null\n" +
+                  "id=1; as_VARCHAR=B\n" +  // TODO: Revisit VARCHAR's acting as VARCHAR(1)
+                  "id=3; as_VARCHAR=A" );
+  }
+
+  @Test
+  public void testOrderVarCharDefaultedAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_VarChar AS VARCHAR ) AS as_VARCHAR \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_VARCHAR NULLS FIRST" )
+        .returns( "id=2; as_VARCHAR=null\n" +
+                  "id=3; as_VARCHAR=A\n" +
+                  "id=1; as_VARCHAR=B" );  // TODO: Revisit VARCHAR's acting as VARCHAR(1)
+  }
+
+  @Test
+  public void testOrderVarCharDefaultedAscNullsLast() throws Exception {
+    JdbcAssert.withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_VarChar AS VARCHAR ) AS as_VARCHAR \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_VARCHAR NULLS LAST" )
+        .returns( "id=3; as_VARCHAR=A\n" +
+                  "id=1; as_VARCHAR=B\n" +  // TODO: Revisit VARCHAR's acting as VARCHAR(1)
+                  "id=2; as_VARCHAR=null" );
+  }
+
+  @Test
+  public void testOrderVarCharDefaultedAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_VarChar AS VARCHAR ) AS as_VARCHAR \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_VARCHAR" )
+        .returns( "id=3; as_VARCHAR=A\n" +
+                  "id=1; as_VARCHAR=B\n" +  // TODO: Revisit VARCHAR's acting as VARCHAR(1)
+                  "id=2; as_VARCHAR=null" );
+  }
+
+
+  ////////////////////
+  // type INTEGER / Int:
+
+  @Test
+  public void testOrderIntAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Int AS INT ) AS as_INT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INT ASC NULLS FIRST" )
+        .returns( "id=2; as_INT=null\n" +
+                  "id=3; as_INT=19\n" +
+                  "id=1; as_INT=180" );
+  }
+
+  @Test
+  public void testOrderIntAscNullsLast() throws Exception {
+     JdbcAssert
+         .withNoDefaultSchema()
+         .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Int AS INT ) AS as_INT \n" +
+               "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+               "ORDER BY as_INT ASC NULLS LAST" )
+         .returns( "id=3; as_INT=19\n" +
+                   "id=1; as_INT=180\n" +
+                   "id=2; as_INT=null"
+                   );
+  }
+
+  @Test
+  public void testOrderIntAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Int AS INT ) AS as_INT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INT ASC" )
+        .returns( "id=3; as_INT=19\n" +
+                  "id=1; as_INT=180\n" +
+                  "id=2; as_INT=null" );
+  }
+
+  @Test
+  public void testOrderIntDescNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Int AS INT ) AS as_INT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INT DESC NULLS FIRST" )
+        .returns( "id=2; as_INT=null\n" +
+                  "id=1; as_INT=180\n" +
+                  "id=3; as_INT=19" );
+  }
+
+  @Test
+  public void testOrderIntDescNullsLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Int AS INT ) AS as_INT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INT DESC NULLS LAST" )
+        .returns( "id=1; as_INT=180\n" +
+                  "id=3; as_INT=19\n" +
+                  "id=2; as_INT=null" );
+  }
+
+  @Test
+  public void testOrderIntDescNullsDefaultFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Int AS INT ) AS as_INT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INT DESC" )
+        .returns( "id=2; as_INT=null\n" +
+                  "id=1; as_INT=180\n" +
+                  "id=3; as_INT=19" );
+  }
+
+  @Test
+  public void testOrderIntDefaultedAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Int AS INT ) AS as_INT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INT NULLS FIRST" )
+        .returns( "id=2; as_INT=null\n" +
+                  "id=3; as_INT=19\n" +
+                  "id=1; as_INT=180" );
+  }
+
+  @Test
+  public void testOrderIntDefaultedAscNullsLast() throws Exception {
+    JdbcAssert.withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Int AS INT ) AS as_INT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INT NULLS LAST" )
+        .returns( "id=3; as_INT=19\n" +
+                  "id=1; as_INT=180\n" +
+                  "id=2; as_INT=null" );
+  }
+
+  @Test
+  public void testOrderIntDefaultedAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Int AS INT ) AS as_INT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INT" )
+        .returns( "id=3; as_INT=19\n" +
+                  "id=1; as_INT=180\n" +
+                  "id=2; as_INT=null" );
+  }
+
+
+  ////////////////////
+  // type FLOAT? / Float?:
+
+  @Test
+  public void testOrderFloatAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Float AS FLOAT ) AS as_FLOAT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_FLOAT ASC NULLS FIRST" )
+        .returns( "id=2; as_FLOAT=null\n" +
+                  "id=3; as_FLOAT=19.0\n" +
+                  "id=1; as_FLOAT=180.0" );
+  }
+
+  @Test
+  public void testOrderFloatAscNullsLast() throws Exception {
+     JdbcAssert
+         .withNoDefaultSchema()
+         .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Float AS FLOAT ) AS as_FLOAT \n" +
+               "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+               "ORDER BY as_FLOAT ASC NULLS LAST" )
+         .returns( "id=3; as_FLOAT=19.0\n" +
+                   "id=1; as_FLOAT=180.0\n" +
+                   "id=2; as_FLOAT=null"
+                   );
+  }
+
+  @Test
+  public void testOrderFloatAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Float AS FLOAT ) AS as_FLOAT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_FLOAT ASC" )
+        .returns( "id=3; as_FLOAT=19.0\n" +
+                  "id=1; as_FLOAT=180.0\n" +
+                  "id=2; as_FLOAT=null" );
+  }
+
+  @Test
+  public void testOrderFloatDescNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Float AS FLOAT ) AS as_FLOAT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_FLOAT DESC NULLS FIRST" )
+        .returns( "id=2; as_FLOAT=null\n" +
+                  "id=1; as_FLOAT=180.0\n" +
+                  "id=3; as_FLOAT=19.0" );
+  }
+
+  @Test
+  public void testOrderFloatDescNullsLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Float AS FLOAT ) AS as_FLOAT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_FLOAT DESC NULLS LAST" )
+        .returns( "id=1; as_FLOAT=180.0\n" +
+                  "id=3; as_FLOAT=19.0\n" +
+                  "id=2; as_FLOAT=null" );
+  }
+
+  @Test
+  public void testOrderFloatDescNullsDefaultFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Float AS FLOAT ) AS as_FLOAT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_FLOAT DESC" )
+        .returns( "id=2; as_FLOAT=null\n" +
+                  "id=1; as_FLOAT=180.0\n" +
+                  "id=3; as_FLOAT=19.0" );
+  }
+
+  @Test
+  public void testOrderFloatDefaultedAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Float AS FLOAT ) AS as_FLOAT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_FLOAT NULLS FIRST" )
+        .returns( "id=2; as_FLOAT=null\n" +
+                  "id=3; as_FLOAT=19.0\n" +
+                  "id=1; as_FLOAT=180.0" );
+  }
+
+  @Test
+  public void testOrderFloatDefaultedAscNullsLast() throws Exception {
+    JdbcAssert.withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Float AS FLOAT ) AS as_FLOAT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_FLOAT NULLS LAST" )
+        .returns( "id=3; as_FLOAT=19.0\n" +
+                  "id=1; as_FLOAT=180.0\n" +
+                  "id=2; as_FLOAT=null" );
+  }
+
+  @Test
+  public void testOrderFloatDefaultedAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Float AS FLOAT ) AS as_FLOAT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_FLOAT" )
+        .returns( "id=3; as_FLOAT=19.0\n" +
+                  "id=1; as_FLOAT=180.0\n" +
+                  "id=2; as_FLOAT=null" );
+  }
+
+
+  ////////////////////
+  // type BIGINT / BigInt:
+
+  @Test
+  public void testOrderBigIntAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_BigInt AS BIGINT ) AS as_BIGINT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_BIGINT ASC NULLS FIRST" )
+        .returns( "id=2; as_BIGINT=null\n" +
+                  "id=3; as_BIGINT=19\n" +
+                  "id=1; as_BIGINT=180" );
+  }
+
+  @Test
+  public void testOrderBigIntAscNullsLast() throws Exception {
+     JdbcAssert
+         .withNoDefaultSchema()
+         .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_BigInt AS BIGINT ) AS as_BIGINT \n" +
+               "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+               "ORDER BY as_BIGINT ASC NULLS LAST" )
+         .returns( "id=3; as_BIGINT=19\n" +
+                   "id=1; as_BIGINT=180\n" +
+                   "id=2; as_BIGINT=null"
+                   );
+  }
+
+  @Test
+  public void testOrderBigIntAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_BigInt AS BIGINT ) AS as_BIGINT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_BIGINT ASC" )
+        .returns( "id=3; as_BIGINT=19\n" +
+                  "id=1; as_BIGINT=180\n" +
+                  "id=2; as_BIGINT=null" );
+  }
+
+  @Test
+  public void testOrderBigIntDescNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_BigInt AS BIGINT ) AS as_BIGINT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_BIGINT DESC NULLS FIRST" )
+        .returns( "id=2; as_BIGINT=null\n" +
+                  "id=1; as_BIGINT=180\n" +
+                  "id=3; as_BIGINT=19" );
+  }
+
+  @Test
+  public void testOrderBigIntDescNullsLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_BigInt AS BIGINT ) AS as_BIGINT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_BIGINT DESC NULLS LAST" )
+        .returns( "id=1; as_BIGINT=180\n" +
+                  "id=3; as_BIGINT=19\n" +
+                  "id=2; as_BIGINT=null" );
+  }
+
+  @Test
+  public void testOrderBigIntDescNullsDefaultFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_BigInt AS BIGINT ) AS as_BIGINT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_BIGINT DESC" )
+        .returns( "id=2; as_BIGINT=null\n" +
+                  "id=1; as_BIGINT=180\n" +
+                  "id=3; as_BIGINT=19" );
+  }
+
+  @Test
+  public void testOrderBigIntDefaultedAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_BigInt AS BIGINT ) AS as_BIGINT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_BIGINT NULLS FIRST" )
+        .returns( "id=2; as_BIGINT=null\n" +
+                  "id=3; as_BIGINT=19\n" +
+                  "id=1; as_BIGINT=180" );
+  }
+
+  @Test
+  public void testOrderBigIntDefaultedAscNullsLast() throws Exception {
+    JdbcAssert.withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_BigInt AS BIGINT ) AS as_BIGINT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_BIGINT NULLS LAST" )
+        .returns( "id=3; as_BIGINT=19\n" +
+                  "id=1; as_BIGINT=180\n" +
+                  "id=2; as_BIGINT=null" );
+  }
+
+  @Test
+  public void testOrderBigIntDefaultedAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_BigInt AS BIGINT ) AS as_BIGINT \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_BIGINT" )
+        .returns( "id=3; as_BIGINT=19\n" +
+                  "id=1; as_BIGINT=180\n" +
+                  "id=2; as_BIGINT=null" );
+  }
+
+
+  ////////////////////
+  // Type DATE? / Date?:
+
+  @Test
+  public void testOrderDateAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Date AS DATE ) AS as_DATE \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DATE ASC NULLS FIRST" )
+        .returns( "id=2; as_DATE=null\n" +
+                  "id=3; as_DATE=2014-01-01\n" +
+                  "id=1; as_DATE=2014-12-31" );
+  }
+
+  @Test
+  public void testOrderDateAscNullsLast() throws Exception {
+     JdbcAssert
+         .withNoDefaultSchema()
+         .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Date AS DATE ) AS as_DATE \n" +
+               "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+               "ORDER BY as_DATE ASC NULLS LAST" )
+         .returns( "id=3; as_DATE=2014-01-01\n" +
+                   "id=1; as_DATE=2014-12-31\n" +
+                   "id=2; as_DATE=null"
+                   );
+  }
+
+  @Test
+  public void testOrderDateAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Date AS DATE ) AS as_DATE \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DATE ASC" )
+        .returns( "id=3; as_DATE=2014-01-01\n" +
+                  "id=1; as_DATE=2014-12-31\n" +
+                  "id=2; as_DATE=null" );
+  }
+
+  @Test
+  public void testOrderDateDescNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Date AS DATE ) AS as_DATE \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DATE DESC NULLS FIRST" )
+        .returns( "id=2; as_DATE=null\n" +
+                  "id=1; as_DATE=2014-12-31\n" +
+                  "id=3; as_DATE=2014-01-01" );
+  }
+
+  @Test
+  public void testOrderDateDescNullsLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Date AS DATE ) AS as_DATE \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DATE DESC NULLS LAST" )
+        .returns( "id=1; as_DATE=2014-12-31\n" +
+                  "id=3; as_DATE=2014-01-01\n" +
+                  "id=2; as_DATE=null" );
+  }
+
+  @Test
+  public void testOrderDateDescNullsDefaultFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Date AS DATE ) AS as_DATE \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DATE DESC" )
+        .returns( "id=2; as_DATE=null\n" +
+                  "id=1; as_DATE=2014-12-31\n" +
+                  "id=3; as_DATE=2014-01-01" );
+  }
+
+  @Test
+  public void testOrderDateDefaultedAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Date AS DATE ) AS as_DATE \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DATE NULLS FIRST" )
+        .returns( "id=2; as_DATE=null\n" +
+                  "id=3; as_DATE=2014-01-01\n" +
+                  "id=1; as_DATE=2014-12-31" );
+  }
+
+  @Test
+  public void testOrderDateDefaultedAscNullsLast() throws Exception {
+    JdbcAssert.withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Date AS DATE ) AS as_DATE \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DATE NULLS LAST" )
+        .returns( "id=3; as_DATE=2014-01-01\n" +
+                  "id=1; as_DATE=2014-12-31\n" +
+                  "id=2; as_DATE=null" );
+  }
+
+  @Test
+  public void testOrderDateDefaultedAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Date AS DATE ) AS as_DATE \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DATE" )
+        .returns( "id=3; as_DATE=2014-01-01\n" +
+                  "id=1; as_DATE=2014-12-31\n" +
+                  "id=2; as_DATE=null" );
+  }
+
+
+  ////////////////////
+  // Type INTERVAL? / Interval?:
+
+  @Test
+  public void testOrderIntervalAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Interval AS INTERVAL HOUR ) AS as_INTERVAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INTERVAL ASC NULLS FIRST" )
+        .returns( "id=2; as_INTERVAL=null\n" +
+                  "id=3; as_INTERVAL=PT3600S\n" +
+                  "id=1; as_INTERVAL=PT7200S" );
+  }
+
+  @Test
+  public void testOrderIntervalAscNullsLast() throws Exception {
+     JdbcAssert
+         .withNoDefaultSchema()
+         .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Interval AS INTERVAL HOUR ) AS as_INTERVAL \n" +
+               "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+               "ORDER BY as_INTERVAL ASC NULLS LAST" )
+         .returns( "id=3; as_INTERVAL=PT3600S\n" +
+                   "id=1; as_INTERVAL=PT7200S\n" +
+                   "id=2; as_INTERVAL=null"
+                   );
+  }
+
+  @Test
+  public void testOrderIntervalAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Interval AS INTERVAL HOUR ) AS as_INTERVAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INTERVAL ASC" )
+        .returns( "id=3; as_INTERVAL=PT3600S\n" +
+                  "id=1; as_INTERVAL=PT7200S\n" +
+                  "id=2; as_INTERVAL=null" );
+  }
+
+  @Test
+  public void testOrderIntervalDescNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Interval AS INTERVAL HOUR ) AS as_INTERVAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INTERVAL DESC NULLS FIRST" )
+        .returns( "id=2; as_INTERVAL=null\n" +
+                  "id=1; as_INTERVAL=PT7200S\n" +
+                  "id=3; as_INTERVAL=PT3600S" );
+  }
+
+  @Test
+  public void testOrderIntervalDescNullsLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Interval AS INTERVAL HOUR ) AS as_INTERVAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INTERVAL DESC NULLS LAST" )
+        .returns( "id=1; as_INTERVAL=PT7200S\n" +
+                  "id=3; as_INTERVAL=PT3600S\n" +
+                  "id=2; as_INTERVAL=null" );
+  }
+
+  @Test
+  public void testOrderIntervalDescNullsDefaultFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Interval AS INTERVAL HOUR ) AS as_INTERVAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INTERVAL DESC" )
+        .returns( "id=2; as_INTERVAL=null\n" +
+                  "id=1; as_INTERVAL=PT7200S\n" +
+                  "id=3; as_INTERVAL=PT3600S" );
+  }
+
+  @Test
+  public void testOrderIntervalDefaultedAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Interval AS INTERVAL HOUR ) AS as_INTERVAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INTERVAL NULLS FIRST" )
+        .returns( "id=2; as_INTERVAL=null\n" +
+                  "id=3; as_INTERVAL=PT3600S\n" +
+                  "id=1; as_INTERVAL=PT7200S" );
+  }
+
+  @Test
+  public void testOrderIntervalDefaultedAscNullsLast() throws Exception {
+    JdbcAssert.withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Interval AS INTERVAL HOUR ) AS as_INTERVAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INTERVAL NULLS LAST" )
+        .returns( "id=3; as_INTERVAL=PT3600S\n" +
+                  "id=1; as_INTERVAL=PT7200S\n" +
+                  "id=2; as_INTERVAL=null" );
+  }
+
+  @Test
+  public void testOrderIntervalDefaultedAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Interval AS INTERVAL HOUR ) AS as_INTERVAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_INTERVAL" )
+        .returns( "id=3; as_INTERVAL=PT3600S\n" +
+                  "id=1; as_INTERVAL=PT7200S\n" +
+                  "id=2; as_INTERVAL=null" );
+  }
+
+
+  ////////////////////
+  // type DECIMAL / Decimal9?:
+
+  @Test
+  public void testOrderDecimalAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal AS DECIMAL ) AS as_DECIMAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL ASC NULLS FIRST" )
+        .returns( "id=2; as_DECIMAL=null\n" +
+                  "id=3; as_DECIMAL=19\n" +
+                  "id=1; as_DECIMAL=180" );
+  }
+
+  @Test
+  public void testOrderDecimalAscNullsLast() throws Exception {
+     JdbcAssert
+         .withNoDefaultSchema()
+         .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal AS DECIMAL ) AS as_DECIMAL \n" +
+               "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+               "ORDER BY as_DECIMAL ASC NULLS LAST" )
+         .returns( "id=3; as_DECIMAL=19\n" +
+                   "id=1; as_DECIMAL=180\n" +
+                   "id=2; as_DECIMAL=null"
+                   );
+  }
+
+  @Test
+  public void testOrderDecimalAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal AS DECIMAL ) AS as_DECIMAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL ASC" )
+        .returns( "id=3; as_DECIMAL=19\n" +
+                  "id=1; as_DECIMAL=180\n" +
+                  "id=2; as_DECIMAL=null" );
+  }
+
+  @Test
+  public void testOrderDecimalDescNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal AS DECIMAL ) AS as_DECIMAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL DESC NULLS FIRST" )
+        .returns( "id=2; as_DECIMAL=null\n" +
+                  "id=1; as_DECIMAL=180\n" +
+                  "id=3; as_DECIMAL=19" );
+  }
+
+  @Test
+  public void testOrderDecimalDescNullsLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal AS DECIMAL ) AS as_DECIMAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL DESC NULLS LAST" )
+        .returns( "id=1; as_DECIMAL=180\n" +
+                  "id=3; as_DECIMAL=19\n" +
+                  "id=2; as_DECIMAL=null" );
+  }
+
+  @Test
+  public void testOrderDecimalDescNullsDefaultFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal AS DECIMAL ) AS as_DECIMAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL DESC" )
+        .returns( "id=2; as_DECIMAL=null\n" +
+                  "id=1; as_DECIMAL=180\n" +
+                  "id=3; as_DECIMAL=19" );
+  }
+
+  @Test
+  public void testOrderDecimalDefaultedAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal AS DECIMAL ) AS as_DECIMAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL NULLS FIRST" )
+        .returns( "id=2; as_DECIMAL=null\n" +
+                  "id=3; as_DECIMAL=19\n" +
+                  "id=1; as_DECIMAL=180" );
+  }
+
+  @Test
+  public void testOrderDecimalDefaultedAscNullsLast() throws Exception {
+    JdbcAssert.withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal AS DECIMAL ) AS as_DECIMAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL NULLS LAST" )
+        .returns( "id=3; as_DECIMAL=19\n" +
+                  "id=1; as_DECIMAL=180\n" +
+                  "id=2; as_DECIMAL=null" );
+  }
+
+  @Test
+  public void testOrderDecimalDefaultedAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal AS DECIMAL ) AS as_DECIMAL \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL" )
+        .returns( "id=3; as_DECIMAL=19\n" +
+                  "id=1; as_DECIMAL=180\n" +
+                  "id=2; as_DECIMAL=null" );
+  }
+
+  ////////////////////
+  // type DECIMAL(5) / Decimal9(?):
+  // Note: DECIMAL(5) and DECIMAL(35) are both tested because Decimal9 (and
+  // Decimal18) and Decimal38Sparse (and Decimal28Sparse) involve different
+  // code paths for NULLS FIRST/LAST handling.
+
+  @Test
+  public void testOrderDecimal5AscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal5 AS DECIMAL(5) ) AS as_DECIMAL5 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL5 ASC NULLS FIRST" )
+        .returns( "id=2; as_DECIMAL5=null\n" +
+                  "id=3; as_DECIMAL5=1235\n" +
+                  "id=1; as_DECIMAL5=9877" );
+  }
+
+  @Test
+  public void testOrderDecimal5AscNullsLast() throws Exception {
+     JdbcAssert
+         .withNoDefaultSchema()
+         .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal5 AS DECIMAL(5) ) AS as_DECIMAL5 \n" +
+               "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+               "ORDER BY as_DECIMAL5 ASC NULLS LAST" )
+         .returns( "id=3; as_DECIMAL5=1235\n" +
+                   "id=1; as_DECIMAL5=9877\n" +
+                   "id=2; as_DECIMAL5=null"
+                   );
+  }
+
+  @Test
+  public void testOrderDecimal5AscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal5 AS DECIMAL(5) ) AS as_DECIMAL5 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL5 ASC" )
+        .returns( "id=3; as_DECIMAL5=1235\n" +
+                  "id=1; as_DECIMAL5=9877\n" +
+                  "id=2; as_DECIMAL5=null" );
+  }
+
+  @Test
+  public void testOrderDecimal5DescNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal5 AS DECIMAL(5) ) AS as_DECIMAL5 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL5 DESC NULLS FIRST" )
+        .returns( "id=2; as_DECIMAL5=null\n" +
+                  "id=1; as_DECIMAL5=9877\n" +
+                  "id=3; as_DECIMAL5=1235" );
+  }
+
+  @Test
+  public void testOrderDecimal5DescNullsLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal5 AS DECIMAL(5) ) AS as_DECIMAL5 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL5 DESC NULLS LAST" )
+        .returns( "id=1; as_DECIMAL5=9877\n" +
+                  "id=3; as_DECIMAL5=1235\n" +
+                  "id=2; as_DECIMAL5=null" );
+  }
+
+  @Test
+  public void testOrderDecimal5DescNullsDefaultFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal5 AS DECIMAL(5) ) AS as_DECIMAL5 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL5 DESC" )
+        .returns( "id=2; as_DECIMAL5=null\n" +
+                  "id=1; as_DECIMAL5=9877\n" +
+                  "id=3; as_DECIMAL5=1235" );
+  }
+
+  @Test
+  public void testOrderDecimal5DefaultedAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal5 AS DECIMAL(5) ) AS as_DECIMAL5 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL5 NULLS FIRST" )
+        .returns( "id=2; as_DECIMAL5=null\n" +
+                  "id=3; as_DECIMAL5=1235\n" +
+                  "id=1; as_DECIMAL5=9877" );
+  }
+
+  @Test
+  public void testOrderDecimal5DefaultedAscNullsLast() throws Exception {
+    JdbcAssert.withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal5 AS DECIMAL(5) ) AS as_DECIMAL5 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL5 NULLS LAST" )
+        .returns( "id=3; as_DECIMAL5=1235\n" +
+                  "id=1; as_DECIMAL5=9877\n" +
+                  "id=2; as_DECIMAL5=null" );
+  }
+
+  @Test
+  public void testOrderDecimal5DefaultedAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal5 AS DECIMAL(5) ) AS as_DECIMAL5 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL5" )
+        .returns( "id=3; as_DECIMAL5=1235\n" +
+                  "id=1; as_DECIMAL5=9877\n" +
+                  "id=2; as_DECIMAL5=null" );
+  }
+
+  ////////////////////
+  // type DECIMAL(35) / Decimal39Sparse(?):
+  // Note: DECIMAL(5) and DECIMAL(35) are both tested because Decimal9 (and
+  // Decimal18) and Decimal38Sparse (and Decimal28Sparse) involve different
+  // code paths for NULLS FIRST/LAST handling.
+
+  @Test
+  public void testOrderDecimal35AscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal35 AS DECIMAL(35) ) AS as_DECIMAL35 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL35 ASC NULLS FIRST" )
+        .returns( "id=2; as_DECIMAL35=null\n" +
+                  "id=3; as_DECIMAL35=12345678901234567890123456789012345\n" +
+                  "id=1; as_DECIMAL35=43210987654321098765432109876543210" );
+  }
+
+  @Test
+  public void testOrderDecimal35AscNullsLast() throws Exception {
+     JdbcAssert
+         .withNoDefaultSchema()
+         .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal35 AS DECIMAL(35) ) AS as_DECIMAL35 \n" +
+               "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+               "ORDER BY as_DECIMAL35 ASC NULLS LAST" )
+         .returns( "id=3; as_DECIMAL35=12345678901234567890123456789012345\n" +
+                   "id=1; as_DECIMAL35=43210987654321098765432109876543210\n" +
+                   "id=2; as_DECIMAL35=null"
+                   );
+  }
+
+  @Test
+  public void testOrderDecimal35AscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal35 AS DECIMAL(35) ) AS as_DECIMAL35 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL35 ASC" )
+        .returns( "id=3; as_DECIMAL35=12345678901234567890123456789012345\n" +
+                  "id=1; as_DECIMAL35=43210987654321098765432109876543210\n" +
+                  "id=2; as_DECIMAL35=null" );
+  }
+
+  @Test
+  public void testOrderDecimal35DescNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal35 AS DECIMAL(35) ) AS as_DECIMAL35 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL35 DESC NULLS FIRST" )
+        .returns( "id=2; as_DECIMAL35=null\n" +
+                  "id=1; as_DECIMAL35=43210987654321098765432109876543210\n" +
+                  "id=3; as_DECIMAL35=12345678901234567890123456789012345" );
+  }
+
+  @Test
+  public void testOrderDecimal35DescNullsLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal35 AS DECIMAL(35) ) AS as_DECIMAL35 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL35 DESC NULLS LAST" )
+        .returns( "id=1; as_DECIMAL35=43210987654321098765432109876543210\n" +
+                  "id=3; as_DECIMAL35=12345678901234567890123456789012345\n" +
+                  "id=2; as_DECIMAL35=null" );
+  }
+
+  @Test
+  public void testOrderDecimal35DescNullsDefaultFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal35 AS DECIMAL(35) ) AS as_DECIMAL35 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL35 DESC" )
+        .returns( "id=2; as_DECIMAL35=null\n" +
+                  "id=1; as_DECIMAL35=43210987654321098765432109876543210\n" +
+                  "id=3; as_DECIMAL35=12345678901234567890123456789012345" );
+  }
+
+  @Test
+  public void testOrderDecimal35DefaultedAscNullsFirst() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal35 AS DECIMAL(35) ) AS as_DECIMAL35 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL35 NULLS FIRST" )
+        .returns( "id=2; as_DECIMAL35=null\n" +
+                  "id=3; as_DECIMAL35=12345678901234567890123456789012345\n" +
+                  "id=1; as_DECIMAL35=43210987654321098765432109876543210" );
+  }
+
+  @Test
+  public void testOrderDecimal35DefaultedAscNullsLast() throws Exception {
+    JdbcAssert.withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal35 AS DECIMAL(35) ) AS as_DECIMAL35 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL35 NULLS LAST" )
+        .returns( "id=3; as_DECIMAL35=12345678901234567890123456789012345\n" +
+                  "id=1; as_DECIMAL35=43210987654321098765432109876543210\n" +
+                  "id=2; as_DECIMAL35=null" );
+  }
+
+  @Test
+  public void testOrderDecimal35DefaultedAscNullsDefaultLast() throws Exception {
+    JdbcAssert
+        .withNoDefaultSchema()
+        .sql( "SELECT tbl.id, \n" +
+              "       CAST( tbl.for_Decimal35 AS DECIMAL(35) ) AS as_DECIMAL35 \n" +
+              "FROM cp.`null_ordering_and_grouping_data.json` AS tbl \n" +
+              "ORDER BY as_DECIMAL35" )
+        .returns( "id=3; as_DECIMAL35=12345678901234567890123456789012345\n" +
+                  "id=1; as_DECIMAL35=43210987654321098765432109876543210\n" +
+                  "id=2; as_DECIMAL35=null" );
+  }
+
+
+}
diff --git a/exec/jdbc/src/test/resources/donuts.json b/exec/jdbc/src/test/resources/donuts.json
new file mode 100644
index 000000000..fea1fbd18
--- /dev/null
+++ b/exec/jdbc/src/test/resources/donuts.json
@@ -0,0 +1,130 @@
+{
+    "id": "0001",
+    "type": "donut",
+    "name": "Cake",
+    "ppu": 0.55,
+    "sales": 35,
+
+    "batters":
+      {
+        "batter":
+          [
+            { "id": "1001", "type": "Regular" },
+            { "id": "1002", "type": "Chocolate" },
+            { "id": "1003", "type": "Blueberry" },
+            { "id": "1004", "type": "Devil's Food" }
+          ]
+      },
+    "topping":
+      [
+        { "id": "5001", "type": "None" },
+        { "id": "5002", "type": "Glazed" },
+        { "id": "5005", "type": "Sugar" },
+        { "id": "5007", "type": "Powdered Sugar" },
+        { "id": "5006", "type": "Chocolate with Sprinkles" },
+        { "id": "5003", "type": "Chocolate" },
+        { "id": "5004", "type": "Maple" }
+      ]
+}
+{
+    "id": "0002",
+    "type": "donut",
+    "name": "Raised",
+    "ppu": 0.69,
+    "sales": 145,
+    "batters":
+      {
+        "batter":
+          [
+            { "id": "1001", "type": "Regular" }
+          ]
+      },
+    "topping":
+      [
+        { "id": "5001", "type": "None" },
+        { "id": "5002", "type": "Glazed" },
+        { "id": "5005", "type": "Sugar" },
+        { "id": "5003", "type": "Chocolate" },
+        { "id": "5004", "type": "Maple" }
+      ]
+}
+{
+    "id": "0003",
+    "type": "donut",
+    "name": "Old Fashioned",
+    "ppu": 0.55,
+    "sales": 300,
+
+
+    "batters":
+      {
+        "batter":
+          [
+            { "id": "1001", "type": "Regular" },
+            { "id": "1002", "type": "Chocolate" }
+          ]
+      },
+    "topping":
+      [
+        { "id": "5001", "type": "None" },
+        { "id": "5002", "type": "Glazed" },
+        { "id": "5003", "type": "Chocolate" },
+        { "id": "5004", "type": "Maple" }
+      ]
+}
+{
+    "id": "0004",
+    "type": "donut",
+    "name": "Filled",
+    "ppu": 0.69,
+    "sales": 14,
+
+    "batters":
+      {
+        "batter":
+          [
+            { "id": "1001", "type": "Regular" },
+            { "id": "1002", "type": "Chocolate" },
+            { "id": "1003", "type": "Blueberry" },
+            { "id": "1004", "type": "Devil's Food" }
+          ]
+      },
+    "topping":
+      [
+        { "id": "5001", "type": "None" },
+        { "id": "5002", "type": "Glazed" },
+        { "id": "5005", "type": "Sugar" },
+        { "id": "5007", "type": "Powdered Sugar" },
+        { "id": "5006", "type": "Chocolate with Sprinkles" },
+        { "id": "5003", "type": "Chocolate" },
+        { "id": "5004", "type": "Maple" }
+      ],
+    "filling":
+      [
+        { "id": "6001", "type": "None" },
+        { "id": "6002", "type": "Raspberry" },
+        { "id": "6003", "type": "Lemon" },
+        { "id": "6004", "type": "Chocolate" },
+        { "id": "6005", "type": "Kreme" }
+      ]
+}
+{
+    "id": "0005",
+    "type": "donut",
+    "name": "Apple Fritter",
+    "ppu": 1.00,
+    "sales": 700,
+
+
+    "batters":
+      {
+        "batter":
+          [
+            { "id": "1001", "type": "Regular" }
+          ]
+      },
+    "topping":
+      [
+        { "id": "5002", "type": "Glazed" }
+      ]
+}
diff --git a/exec/jdbc/src/test/resources/null_ordering_and_grouping_data.json b/exec/jdbc/src/test/resources/null_ordering_and_grouping_data.json
new file mode 100644
index 000000000..aa701aa0f
--- /dev/null
+++ b/exec/jdbc/src/test/resources/null_ordering_and_grouping_data.json
@@ -0,0 +1,63 @@
+{
+    "id": "1"
+    ,
+    "for_VarChar": "Bb"
+    ,
+    "for_Int": 180
+    ,
+    "for_Float": "180"
+    ,
+    "for_BigInt": "180"
+    ,
+    "for_Date": "2014-12-31"
+    ,
+    "for_Interval": "PT7200S"
+    ,
+    "for_Decimal": "180"
+    ,
+    "for_Decimal5": "9876.54321"
+    ,
+    "for_Decimal35": "43210987654321098765432109876543210"
+}
+{
+    "id": "2"
+    ,
+    "for_VarChar": null
+    ,
+    "for_Int": null
+    ,
+    "for_Float": null
+    ,
+    "for_BigInt": null
+    ,
+    "for_Date": null
+    ,
+    "for_Interval": null
+    ,
+    "for_Decimal": null
+    ,
+    "for_Decimal5": null
+    ,
+    "for_Decimal35": null
+}
+{
+    "id": "3"
+    ,
+    "for_VarChar": "A"
+    ,
+    "for_Int": 19
+    ,
+    "for_Float": "19"
+    ,
+    "for_BigInt": "19"
+    ,
+    "for_Date": "2014-01-01"
+    ,
+    "for_Interval": "PT3600S"
+    ,
+    "for_Decimal": "19"
+    ,
+    "for_Decimal5": "1234.56789"
+    ,
+    "for_Decimal35": "12345678901234567890123456789012345"
+}