Support qualified columns in queries (#55)

* support qualified columns in queries

* handle coalesced hash join partition in HashJoinStream

* implement Into<Column> for &str

* add todo for ARROW-10971

* fix cross join handling in production push down optimizer

When a projection is pushed down to cross join inputs, fields from
resulting plan's schema need to be trimmed to only contain projected
fields.

* maintain field order during plan optimization using projections

* change TableScane name from Option<String> to String

* WIP: fix ballista

* separate logical and physical expressions in proto, fix ballista build

* fix join schema handling in production push down optimizer

schema needs to be recalculated based on newly optimized inputs

* tpch 7 & 8 are now passing!

* fix roundtrip_join test

* fix clippy warnings

* fix sql planner test error checking with matches

`format("{:?}", err)` yields different results between stable and
nightly rust.

* address FIXMEs

* honor datafusion field name semantic

strip qualifer name in physical field names

* add more comment

* enable more queries in benchmark/run.sh

* use unzip to avoid unnecessary iterators

* reduce diff by discarding style related changes

* simplify hash_join tests

* reduce diff for easier revuew

* fix unnecessary reference clippy error

* incorporate code review feedback

* fix window schema handling in projection pushdown optimizer

Author: QP Hou

ballista/rust/core/proto/ballista.proto

@@ -28,11 +28,29 @@ option java_outer_classname = "BallistaProto";
 // Ballista Logical Plan
 ///////////////////////////////////////////////////////////////////////////////////////////////////
 
+message ColumnRelation {
+  string relation = 1;
+}
+
+message Column {
+  string name = 1;
+  ColumnRelation relation = 2;
+}
+
+message DfField{
+  Field field = 1;
+  ColumnRelation qualifier = 2;
+}
+
+message DfSchema {
+  repeated DfField columns = 1;
+}
+
 // logical expressions
 message LogicalExprNode {
   oneof ExprType {
     // column references
-    string column_name = 1;
+    Column column = 1;
 
     // alias
     AliasNode alias = 2;
@@ -295,7 +313,7 @@ message CreateExternalTableNode{
   string location = 2;
   FileType file_type = 3;
   bool has_header = 4;
-  Schema schema = 5;
+  DfSchema schema = 5;
 }
 
 enum FileType{
@@ -309,11 +327,6 @@ message ExplainNode{
   bool verbose = 2;
 }
 
-message DfField{
-  string qualifier = 2;
-  Field field = 1;
-}
-
 message AggregateNode {
   LogicalPlanNode input = 1;
   repeated LogicalExprNode group_expr = 2;
@@ -369,8 +382,8 @@ message JoinNode {
   LogicalPlanNode left = 1;
   LogicalPlanNode right = 2;
   JoinType join_type = 3;
-  repeated string left_join_column = 4;
-  repeated string right_join_column = 5;
+  repeated Column left_join_column = 4;
+  repeated Column right_join_column = 5;
 }
 
 message LimitNode {
@@ -408,6 +421,119 @@ message PhysicalPlanNode {
   }
 }
 
+// physical expressions
+message PhysicalExprNode {
+  oneof ExprType {
+    // column references
+    PhysicalColumn column = 1;
+
+    ScalarValue literal = 2;
+
+    // binary expressions
+    PhysicalBinaryExprNode binary_expr = 3;
+
+    // aggregate expressions
+    PhysicalAggregateExprNode aggregate_expr = 4;
+
+    // null checks
+    PhysicalIsNull is_null_expr = 5;
+    PhysicalIsNotNull is_not_null_expr = 6;
+    PhysicalNot not_expr = 7;
+
+    PhysicalCaseNode case_ = 8;
+    PhysicalCastNode cast = 9;
+    PhysicalSortExprNode sort = 10;
+    PhysicalNegativeNode negative = 11;
+    PhysicalInListNode in_list = 12;
+    PhysicalScalarFunctionNode scalar_function = 13;
+    PhysicalTryCastNode try_cast = 14;
+
+    // window expressions
+    PhysicalWindowExprNode window_expr = 15;
+  }
+}
+
+message PhysicalAggregateExprNode {
+  AggregateFunction aggr_function = 1;
+  PhysicalExprNode expr = 2;
+}
+
+message PhysicalWindowExprNode {
+  oneof window_function {
+    AggregateFunction aggr_function = 1;
+    BuiltInWindowFunction built_in_function = 2;
+    // udaf = 3
+  }
+  PhysicalExprNode expr = 4;
+}
+
+message PhysicalIsNull {
+  PhysicalExprNode expr = 1;
+}
+
+message PhysicalIsNotNull {
+  PhysicalExprNode expr = 1;
+}
+
+message PhysicalNot {
+  PhysicalExprNode expr = 1;
+}
+
+message PhysicalAliasNode {
+  PhysicalExprNode expr = 1;
+  string alias = 2;
+}
+
+message PhysicalBinaryExprNode {
+  PhysicalExprNode l = 1;
+  PhysicalExprNode r = 2;
+  string op = 3;
+}
+
+message PhysicalSortExprNode {
+  PhysicalExprNode expr = 1;
+  bool asc = 2;
+  bool nulls_first = 3;
+}
+
+message PhysicalWhenThen {
+  PhysicalExprNode when_expr = 1;
+  PhysicalExprNode then_expr = 2;
+}
+
+message PhysicalInListNode {
+  PhysicalExprNode expr = 1;
+  repeated PhysicalExprNode list = 2;
+  bool negated = 3;
+}
+
+message PhysicalCaseNode {
+  PhysicalExprNode expr = 1;
+  repeated PhysicalWhenThen when_then_expr = 2;
+  PhysicalExprNode else_expr = 3;
+}
+
+message PhysicalScalarFunctionNode {
+  string name = 1;
+  ScalarFunction fun = 2;
+  repeated PhysicalExprNode args = 3;
+  ArrowType return_type = 4;
+}
+
+message PhysicalTryCastNode {
+  PhysicalExprNode expr = 1;
+  ArrowType arrow_type = 2;
+}
+
+message PhysicalCastNode {
+  PhysicalExprNode expr = 1;
+  ArrowType arrow_type = 2;
+}
+
+message PhysicalNegativeNode {
+  PhysicalExprNode expr = 1;
+}
+
 message UnresolvedShuffleExecNode {
   repeated uint32 query_stage_ids = 1;
   Schema schema = 2;
@@ -416,7 +542,7 @@ message UnresolvedShuffleExecNode {
 
 message FilterExecNode {
   PhysicalPlanNode input = 1;
-  LogicalExprNode expr = 2;
+  PhysicalExprNode expr = 2;
 }
 
 message ParquetScanExecNode {
@@ -447,11 +573,15 @@ message HashJoinExecNode {
 
 }
 
-message JoinOn {
-   string left = 1;
-   string right = 2;
+message PhysicalColumn {
+  string name = 1;
+  uint32 index = 2;
 }
 
+message JoinOn {
+  PhysicalColumn left = 1;
+  PhysicalColumn right = 2;
+}
 
 message EmptyExecNode {
   bool produce_one_row = 1;
@@ -460,7 +590,7 @@ message EmptyExecNode {
 
 message ProjectionExecNode {
   PhysicalPlanNode input = 1;
-  repeated LogicalExprNode expr = 2;
+  repeated PhysicalExprNode expr = 2;
   repeated string expr_name = 3;
 }
 
@@ -472,14 +602,14 @@ enum AggregateMode {
 
 message WindowAggExecNode {
   PhysicalPlanNode input = 1;
-  repeated LogicalExprNode window_expr = 2;
+  repeated PhysicalExprNode window_expr = 2;
   repeated string window_expr_name = 3;
   Schema input_schema = 4;
 }
 
 message HashAggregateExecNode {
-  repeated LogicalExprNode group_expr = 1;
-  repeated LogicalExprNode aggr_expr = 2;
+  repeated PhysicalExprNode group_expr = 1;
+  repeated PhysicalExprNode aggr_expr = 2;
   AggregateMode mode = 3;
   PhysicalPlanNode input = 4;
   repeated string group_expr_name = 5;
@@ -510,7 +640,7 @@ message LocalLimitExecNode {
 
 message SortExecNode {
   PhysicalPlanNode input = 1;
-  repeated LogicalExprNode expr = 2;
+  repeated PhysicalExprNode expr = 2;
 }
 
 message CoalesceBatchesExecNode {
@@ -522,11 +652,16 @@ message MergeExecNode {
   PhysicalPlanNode input = 1;
 }
 
+message PhysicalHashRepartition {
+  repeated PhysicalExprNode hash_expr = 1;
+  uint64 partition_count = 2;
+}
+
 message RepartitionExecNode{
   PhysicalPlanNode input = 1;
   oneof partition_method {
     uint64 round_robin = 2;
-    HashRepartition hash = 3;
+    PhysicalHashRepartition hash = 3;
     uint64 unknown = 4;
   }
 }
@@ -803,7 +938,7 @@ message ScalarListValue{
 
 
 message ScalarValue{
-    oneof value{
+    oneof value {
         bool   bool_value = 1;
         string utf8_value = 2;
         string large_utf8_value = 3;