Bug fix (#9)

README.md

@@ -26,7 +26,7 @@ ctx.sql('select sex, smoker, avg(tip/total_bill) as tip_pct from tips group by s
 
 ## Status
 
-- Proof-of-concept. Not producing correct results yet.
+- Seems to be working, but only tested with a few queries so far!
 
 ## Features
 

examples/tips.py

@@ -12,6 +12,9 @@
 ctx = RaySqlContext(workers)
 ctx.register_csv('tips', 'tips.csv', True)
 
-ctx.sql('select day, sum(total_bill) from tips group by day')
+# Parquet is also supported
+# ctx.register_parquet('tips', 'tips.parquet')
 
-#ctx.sql('select sex, smoker, avg(tip/total_bill) as tip_pct from tips group by sex, smoker')
+ctx.sql('select sex, smoker, avg(tip/total_bill) as tip_pct from tips group by sex, smoker')
+
+# ctx.sql('select day, sum(total_bill) from tips group by day')

src/planner.rs

@@ -189,7 +189,11 @@ fn generate_query_stages(
             &Partitioning::Hash(_, _) => {
                 // create a shuffle query stage for this repartition
                 let stage_id = graph.next_id();
-                let shuffle_writer = ShuffleWriterExec::new(stage_id, plan.clone());
+                // we remove the RepartitionExec because it isn't designed for our
+                // distributed use case and the ShuffleWriterExec contains the repartitioning
+                // logic
+                let shuffle_writer_input = plan.children()[0].clone();
+                let shuffle_writer = ShuffleWriterExec::new(stage_id, shuffle_writer_input);
                 let stage_id = graph.add_query_stage(stage_id, Arc::new(shuffle_writer));
                 // replace the plan with a shuffle reader
                 Ok(Arc::new(ShuffleReaderExec::new(stage_id, plan.schema())))