<s> import collections <EOL> import random <EOL> import sys <EOL> import re <EOL> from raco . algebra import * <EOL> from raco . expression import NamedAttributeRef as AttRef <EOL> from raco . expression import UnnamedAttributeRef as AttIndex <EOL> from raco . expression import StateVar <EOL> from raco . expression import aggregate <EOL> from raco . backends . myria import ( <EOL> MyriaShuffleConsumer , MyriaShuffleProducer , MyriaHyperShuffleProducer , <EOL> MyriaBroadcastConsumer , MyriaQueryScan , MyriaSplitConsumer , MyriaDupElim , <EOL> MyriaGroupBy ) <EOL> from raco . backends . myria import ( MyriaLeftDeepTreeAlgebra , <EOL> MyriaHyperCubeAlgebra ) <EOL> from raco . compile import optimize <EOL> from raco import relation_key <EOL> from raco . catalog import FakeCatalog <EOL> import raco . scheme as scheme <EOL> import raco . myrial . myrial_test as myrial_test <EOL> from raco import types <EOL> class OptimizerTest ( myrial_test . MyrialTestCase ) : <EOL> x_scheme = scheme . Scheme ( [ ( "<STR_LIT:a>" , types . LONG_TYPE ) , ( "<STR_LIT:b>" , types . LONG_TYPE ) , ( "<STR_LIT:c>" , types . LONG_TYPE ) ] ) <EOL> y_scheme = scheme . Scheme ( [ ( "<STR_LIT:d>" , types . LONG_TYPE ) , ( "<STR_LIT:e>" , types . LONG_TYPE ) , ( "<STR_LIT:f>" , types . LONG_TYPE ) ] ) <EOL> part_scheme = scheme . Scheme ( [ ( "<STR_LIT:g>" , types . LONG_TYPE ) , ( "<STR_LIT:h>" , types . LONG_TYPE ) , ( "<STR_LIT:i>" , types . LONG_TYPE ) ] ) <EOL> x_key = relation_key . RelationKey . from_string ( "<STR_LIT>" ) <EOL> y_key = relation_key . RelationKey . from_string ( "<STR_LIT>" ) <EOL> part_key = relation_key . RelationKey . from_string ( "<STR_LIT>" ) <EOL> part_partition = RepresentationProperties ( <EOL> hash_partitioned = frozenset ( [ AttIndex ( <NUM_LIT:1> ) ] ) ) <EOL> def setUp ( self ) : <EOL> super ( OptimizerTest , self ) . setUp ( ) <EOL> random . seed ( <NUM_LIT> ) <EOL> rng = <NUM_LIT:20> <EOL> count = <NUM_LIT:30> <EOL> self . x_data = collections . Counter ( <EOL> [ ( random . randrange ( rng ) , random . randrange ( rng ) , <EOL> random . randrange ( rng ) ) for _ in range ( count ) ] ) <EOL> self . y_data = collections . Counter ( <EOL> [ ( random . randrange ( rng ) , random . randrange ( rng ) , <EOL> random . randrange ( rng ) ) for _ in range ( count ) ] ) <EOL> self . part_data = collections . Counter ( <EOL> [ ( random . randrange ( rng ) , random . randrange ( rng ) , <EOL> random . randrange ( rng ) ) for _ in range ( count ) ] ) <EOL> self . db . ingest ( OptimizerTest . x_key , <EOL> self . x_data , <EOL> OptimizerTest . x_scheme ) <EOL> self . db . ingest ( OptimizerTest . y_key , <EOL> self . y_data , <EOL> OptimizerTest . y_scheme ) <EOL> self . db . ingest ( OptimizerTest . part_key , <EOL> self . part_data , <EOL> OptimizerTest . part_scheme , <EOL> self . part_partition ) <EOL> self . expected = collections . Counter ( <EOL> [ ( a , b , c , d , e , f ) for ( a , b , c ) in self . x_data <EOL> for ( d , e , f ) in self . y_data if a > b and e <= f and c == d ] ) <EOL> self . z_key = relation_key . RelationKey . from_string ( "<STR_LIT>" ) <EOL> self . z_data = collections . Counter ( [ ( <NUM_LIT:1> , <NUM_LIT:2> ) , ( <NUM_LIT:2> , <NUM_LIT:3> ) , ( <NUM_LIT:1> , <NUM_LIT:2> ) , ( <NUM_LIT:3> , <NUM_LIT:4> ) ] ) <EOL> self . z_scheme = scheme . Scheme ( [ ( '<STR_LIT:src>' , types . LONG_TYPE ) , ( '<STR_LIT>' , types . LONG_TYPE ) ] ) <EOL> self . db . ingest ( '<STR_LIT>' , self . z_data , self . z_scheme ) <EOL> self . expected2 = collections . Counter ( <EOL> [ ( s1 , d3 ) for ( s1 , d1 ) in self . z_data . elements ( ) <EOL> for ( s2 , d2 ) in self . z_data . elements ( ) <EOL> for ( s3 , d3 ) in self . z_data . elements ( ) if d1 == s2 and d2 == s3 ] ) <EOL> @ staticmethod <EOL> def logical_to_physical ( lp , ** kwargs ) : <EOL> if kwargs . get ( '<STR_LIT>' , False ) : <EOL> algebra = MyriaHyperCubeAlgebra ( FakeCatalog ( <NUM_LIT:64> ) ) <EOL> else : <EOL> algebra = MyriaLeftDeepTreeAlgebra ( ) <EOL> return optimize ( lp , algebra , ** kwargs ) <EOL> @ staticmethod <EOL> def get_count ( op , claz ) : <EOL> """<STR_LIT>""" <EOL> def count ( _op ) : <EOL> if isinstance ( _op , claz ) : <EOL> yield <NUM_LIT:1> <EOL> else : <EOL> yield <NUM_LIT:0> <EOL> return sum ( op . postorder ( count ) ) <EOL> @ staticmethod <EOL> def get_num_select_conjuncs ( op ) : <EOL> """<STR_LIT>""" <EOL> def count ( _op ) : <EOL> if isinstance ( _op , Select ) : <EOL> yield len ( expression . extract_conjuncs ( _op . condition ) ) <EOL> else : <EOL> yield <NUM_LIT:0> <EOL> return sum ( op . postorder ( count ) ) <EOL> def test_push_selects ( self ) : <EOL> """<STR_LIT>""" <EOL> lp = StoreTemp ( '<STR_LIT>' , <EOL> Select ( expression . LTEQ ( AttRef ( "<STR_LIT:e>" ) , AttRef ( "<STR_LIT:f>" ) ) , <EOL> Select ( expression . EQ ( AttRef ( "<STR_LIT:c>" ) , AttRef ( "<STR_LIT:d>" ) ) , <EOL> Select ( expression . GT ( AttRef ( "<STR_LIT:a>" ) , AttRef ( "<STR_LIT:b>" ) ) , <EOL> CrossProduct ( Scan ( self . x_key , self . x_scheme ) , <EOL> Scan ( self . y_key , self . y_scheme ) ) ) ) ) ) <EOL> self . assertEquals ( self . get_count ( lp , Select ) , <NUM_LIT:3> ) <EOL> self . assertEquals ( self . get_count ( lp , CrossProduct ) , <NUM_LIT:1> ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertIsInstance ( pp . input , MyriaSplitConsumer ) <EOL> self . assertIsInstance ( pp . input . input . input , Join ) <EOL> self . assertEquals ( self . get_count ( pp , Select ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . get_count ( pp , CrossProduct ) , <NUM_LIT:0> ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_temp_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , self . expected ) <EOL> def test_collapse_applies ( self ) : <EOL> """<STR_LIT>""" <EOL> lp = StoreTemp ( '<STR_LIT>' , <EOL> Apply ( [ ( None , AttIndex ( <NUM_LIT:1> ) ) , ( '<STR_LIT:w>' , expression . PLUS ( AttIndex ( <NUM_LIT:0> ) , AttIndex ( <NUM_LIT:0> ) ) ) ] , <EOL> Apply ( [ ( None , AttIndex ( <NUM_LIT:1> ) ) , ( None , AttIndex ( <NUM_LIT:0> ) ) , ( None , AttIndex ( <NUM_LIT:1> ) ) ] , <EOL> Apply ( [ ( '<STR_LIT:x>' , AttIndex ( <NUM_LIT:0> ) ) , ( '<STR_LIT:y>' , expression . PLUS ( AttIndex ( <NUM_LIT:1> ) , AttIndex ( <NUM_LIT:0> ) ) ) ] , <EOL> Apply ( [ ( None , AttIndex ( <NUM_LIT:0> ) ) , ( None , AttIndex ( <NUM_LIT:1> ) ) ] , <EOL> Scan ( self . x_key , self . x_scheme ) ) ) ) ) ) <EOL> self . assertEquals ( self . get_count ( lp , Apply ) , <NUM_LIT:4> ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertIsInstance ( pp . input , Apply ) <EOL> self . assertEquals ( self . get_count ( pp , Apply ) , <NUM_LIT:1> ) <EOL> expected = collections . Counter ( <EOL> [ ( b , a + a ) for ( a , b , c ) in <EOL> [ ( b , a , b ) for ( a , b ) in <EOL> [ ( a , b + a ) for ( a , b ) in <EOL> [ ( a , b ) for ( a , b , c ) in self . x_data ] ] ] ] ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_temp_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_select_count_star ( self ) : <EOL> """<STR_LIT>""" <EOL> lp = StoreTemp ( '<STR_LIT>' , <EOL> GroupBy ( [ ] , [ expression . COUNTALL ( ) ] , <EOL> Scan ( self . x_key , self . x_scheme ) ) ) <EOL> self . assertEquals ( self . get_count ( lp , GroupBy ) , <NUM_LIT:1> ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertIsInstance ( pp . input . input . input , GroupBy ) <EOL> apply = pp . input . input . input . input . input . input . input <EOL> self . assertIsInstance ( apply , Apply ) <EOL> self . assertEquals ( self . get_count ( pp , Apply ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( len ( apply . scheme ( ) ) , <NUM_LIT:1> ) <EOL> expected = collections . Counter ( [ ( len ( self . x_data ) , ) ] ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_temp_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_projects_apply_join ( self ) : <EOL> """<STR_LIT>""" <EOL> lp = StoreTemp ( '<STR_LIT>' , <EOL> Apply ( [ ( None , AttIndex ( <NUM_LIT:1> ) ) ] , <EOL> ProjectingJoin ( expression . EQ ( AttIndex ( <NUM_LIT:0> ) , AttIndex ( <NUM_LIT:3> ) ) , <EOL> Scan ( self . x_key , self . x_scheme ) , <EOL> Scan ( self . x_key , self . x_scheme ) , <EOL> [ AttIndex ( i ) for i in xrange ( <NUM_LIT:2> * len ( self . x_scheme ) ) ] ) ) ) <EOL> self . assertIsInstance ( lp . input . input , ProjectingJoin ) <EOL> self . assertEquals ( <NUM_LIT:2> * len ( self . x_scheme ) , <EOL> len ( lp . input . input . scheme ( ) ) ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertIsInstance ( pp . input , MyriaSplitConsumer ) <EOL> proj_join = pp . input . input . input <EOL> self . assertIsInstance ( proj_join , ProjectingJoin ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( proj_join . scheme ( ) ) ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( proj_join . left . scheme ( ) ) ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( proj_join . right . scheme ( ) ) ) <EOL> expected = collections . Counter ( <EOL> [ ( b , ) <EOL> for ( a , b , c ) in self . x_data <EOL> for ( d , e , f ) in self . x_data <EOL> if a == d ] ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_temp_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_push_selects_apply ( self ) : <EOL> """<STR_LIT>""" <EOL> lp = StoreTemp ( '<STR_LIT>' , <EOL> Select ( expression . LTEQ ( AttRef ( "<STR_LIT:c>" ) , AttRef ( "<STR_LIT:a>" ) ) , <EOL> Select ( expression . LTEQ ( AttRef ( "<STR_LIT:b>" ) , AttRef ( "<STR_LIT:c>" ) ) , <EOL> Apply ( [ ( '<STR_LIT:b>' , AttIndex ( <NUM_LIT:1> ) ) , <EOL> ( '<STR_LIT:c>' , AttIndex ( <NUM_LIT:2> ) ) , <EOL> ( '<STR_LIT:a>' , AttIndex ( <NUM_LIT:0> ) ) ] , <EOL> Scan ( self . x_key , self . x_scheme ) ) ) ) ) <EOL> expected = collections . Counter ( <EOL> [ ( b , c , a ) for ( a , b , c ) in self . x_data if c <= a and b <= c ] ) <EOL> self . assertEquals ( self . get_count ( lp , Select ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . get_count ( lp , Scan ) , <NUM_LIT:1> ) <EOL> self . assertIsInstance ( lp . input , Select ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertIsInstance ( pp . input , Apply ) <EOL> self . assertEquals ( self . get_count ( pp , Select ) , <NUM_LIT:1> ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_temp_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_push_selects_groupby ( self ) : <EOL> """<STR_LIT>""" <EOL> lp = StoreTemp ( '<STR_LIT>' , <EOL> Select ( expression . LTEQ ( AttRef ( "<STR_LIT:c>" ) , AttRef ( "<STR_LIT:a>" ) ) , <EOL> Select ( expression . LTEQ ( AttRef ( "<STR_LIT:b>" ) , AttRef ( "<STR_LIT:c>" ) ) , <EOL> GroupBy ( [ AttIndex ( <NUM_LIT:1> ) , AttIndex ( <NUM_LIT:2> ) , AttIndex ( <NUM_LIT:0> ) ] , <EOL> [ expression . COUNTALL ( ) ] , <EOL> Scan ( self . x_key , self . x_scheme ) ) ) ) ) <EOL> expected = collections . Counter ( <EOL> [ ( b , c , a ) for ( a , b , c ) in self . x_data if c <= a and b <= c ] ) <EOL> expected = collections . Counter ( k + ( v , ) for k , v in expected . items ( ) ) <EOL> self . assertEquals ( self . get_count ( lp , Select ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . get_count ( lp , Scan ) , <NUM_LIT:1> ) <EOL> self . assertIsInstance ( lp . input , Select ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertIsInstance ( pp . input , MyriaSplitConsumer ) <EOL> self . assertIsInstance ( pp . input . input . input , GroupBy ) <EOL> self . assertEquals ( self . get_count ( pp , Select ) , <NUM_LIT:1> ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_temp_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_noop_apply_removed ( self ) : <EOL> lp = StoreTemp ( '<STR_LIT>' , <EOL> Apply ( [ ( None , AttIndex ( <NUM_LIT:1> ) ) ] , <EOL> ProjectingJoin ( expression . EQ ( AttIndex ( <NUM_LIT:0> ) , AttIndex ( <NUM_LIT:3> ) ) , <EOL> Scan ( self . x_key , self . x_scheme ) , <EOL> Scan ( self . x_key , self . x_scheme ) , <EOL> [ AttIndex ( i ) for i in xrange ( <NUM_LIT:2> * len ( self . x_scheme ) ) ] ) ) ) <EOL> self . assertIsInstance ( lp . input , Apply ) <EOL> lp_scheme = lp . scheme ( ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertNotIsInstance ( pp . input , Apply ) <EOL> self . assertEquals ( lp_scheme , pp . scheme ( ) ) <EOL> def test_not_noop_apply_not_removed ( self ) : <EOL> lp = StoreTemp ( '<STR_LIT>' , <EOL> Apply ( [ ( '<STR_LIT>' , AttIndex ( <NUM_LIT:1> ) ) ] , <EOL> ProjectingJoin ( expression . EQ ( AttIndex ( <NUM_LIT:0> ) , AttIndex ( <NUM_LIT:3> ) ) , <EOL> Scan ( self . x_key , self . x_scheme ) , <EOL> Scan ( self . x_key , self . x_scheme ) , <EOL> [ AttIndex ( i ) for i in xrange ( <NUM_LIT:2> * len ( self . x_scheme ) ) ] ) ) ) <EOL> self . assertIsInstance ( lp . input , Apply ) <EOL> lp_scheme = lp . scheme ( ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertIsInstance ( pp . input , Apply ) <EOL> self . assertEquals ( lp_scheme , pp . scheme ( ) ) <EOL> def test_extract_join ( self ) : <EOL> """<STR_LIT>""" <EOL> s = expression . AND ( expression . LTEQ ( AttRef ( "<STR_LIT:e>" ) , AttRef ( "<STR_LIT:f>" ) ) , <EOL> expression . AND ( <EOL> expression . EQ ( AttRef ( "<STR_LIT:c>" ) , AttRef ( "<STR_LIT:d>" ) ) , <EOL> expression . GT ( AttRef ( "<STR_LIT:a>" ) , AttRef ( "<STR_LIT:b>" ) ) ) ) <EOL> lp = StoreTemp ( '<STR_LIT>' , Select ( s , CrossProduct ( <EOL> Scan ( self . x_key , self . x_scheme ) , <EOL> Scan ( self . y_key , self . y_scheme ) ) ) ) <EOL> self . assertEquals ( self . get_num_select_conjuncs ( lp ) , <NUM_LIT:3> ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertIsInstance ( pp . input , MyriaSplitConsumer ) <EOL> self . assertIsInstance ( pp . input . input . input , Join ) <EOL> self . assertEquals ( self . get_count ( pp , CrossProduct ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , Select ) , <NUM_LIT:2> ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_temp_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , self . expected ) <EOL> def test_multi_condition_join ( self ) : <EOL> s = expression . AND ( expression . EQ ( AttRef ( "<STR_LIT:c>" ) , AttRef ( "<STR_LIT:d>" ) ) , <EOL> expression . EQ ( AttRef ( "<STR_LIT:a>" ) , AttRef ( "<STR_LIT:f>" ) ) ) <EOL> lp = StoreTemp ( '<STR_LIT>' , Select ( s , CrossProduct ( <EOL> Scan ( self . x_key , self . x_scheme ) , <EOL> Scan ( self . y_key , self . y_scheme ) ) ) ) <EOL> self . assertEquals ( self . get_num_select_conjuncs ( lp ) , <NUM_LIT:2> ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , CrossProduct ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , Select ) , <NUM_LIT:0> ) <EOL> expected = collections . Counter ( <EOL> [ ( a , b , c , d , e , f ) for ( a , b , c ) in self . x_data <EOL> for ( d , e , f ) in self . y_data if a == f and c == d ] ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_temp_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_multiway_join_left_deep ( self ) : <EOL> query = """<STR_LIT>""" <EOL> lp = self . get_logical_plan ( query ) <EOL> self . assertEquals ( self . get_count ( lp , CrossProduct ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . get_count ( lp , Join ) , <NUM_LIT:0> ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , CrossProduct ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , Join ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:4> ) <EOL> self . assertEquals ( self . get_count ( pp , NaryJoin ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaHyperShuffleProducer ) , <NUM_LIT:0> ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , self . expected2 ) <EOL> def test_multiway_join_hyper_cube ( self ) : <EOL> query = """<STR_LIT>""" <EOL> lp = self . get_logical_plan ( query ) <EOL> self . assertEquals ( self . get_count ( lp , CrossProduct ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . get_count ( lp , Join ) , <NUM_LIT:0> ) <EOL> pp = self . logical_to_physical ( lp , hypercube = True ) <EOL> self . assertEquals ( self . get_count ( pp , CrossProduct ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , Join ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , NaryJoin ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaHyperShuffleProducer ) , <NUM_LIT:3> ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , self . expected2 ) <EOL> def test_hyper_cube_tie_breaking_heuristic ( self ) : <EOL> query = """<STR_LIT>""" <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp , hypercube = True ) <EOL> def get_max_dim_size ( _op ) : <EOL> if isinstance ( _op , MyriaHyperShuffleProducer ) : <EOL> yield max ( _op . hyper_cube_dimensions ) <EOL> self . assertTrue ( max ( pp . postorder ( get_max_dim_size ) ) <= <NUM_LIT:4> ) <EOL> def test_naryjoin_merge ( self ) : <EOL> query = """<STR_LIT>""" <EOL> statements = self . parser . parse ( query ) <EOL> self . processor . evaluate ( statements ) <EOL> lp = self . processor . get_logical_plan ( ) <EOL> pp = self . logical_to_physical ( lp , hypercube = True ) <EOL> self . assertEquals ( self . get_count ( pp , NaryJoin ) , <NUM_LIT:0> ) <EOL> def test_right_deep_join ( self ) : <EOL> """<STR_LIT>""" <EOL> s = expression . AND ( expression . EQ ( AttIndex ( <NUM_LIT:1> ) , AttIndex ( <NUM_LIT:2> ) ) , <EOL> expression . EQ ( AttIndex ( <NUM_LIT:3> ) , AttIndex ( <NUM_LIT:4> ) ) ) <EOL> lp = Apply ( [ ( '<STR_LIT:x>' , AttIndex ( <NUM_LIT:0> ) ) , ( '<STR_LIT:y>' , AttIndex ( <NUM_LIT:5> ) ) ] , <EOL> Select ( s , <EOL> CrossProduct ( Scan ( self . z_key , self . z_scheme ) , <EOL> CrossProduct ( <EOL> Scan ( self . z_key , self . z_scheme ) , <EOL> Scan ( self . z_key , self . z_scheme ) ) ) ) ) <EOL> lp = StoreTemp ( '<STR_LIT>' , lp ) <EOL> self . assertEquals ( self . get_count ( lp , CrossProduct ) , <NUM_LIT:2> ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , CrossProduct ) , <NUM_LIT:0> ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_temp_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , self . expected2 ) <EOL> def test_explicit_shuffle ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" <EOL> statements = self . parser . parse ( query ) <EOL> self . processor . evaluate ( statements ) <EOL> lp = self . processor . get_logical_plan ( ) <EOL> self . assertEquals ( self . get_count ( lp , Shuffle ) , <NUM_LIT:1> ) <EOL> for op in lp . walk ( ) : <EOL> if isinstance ( op , Shuffle ) : <EOL> self . assertEquals ( op . columnlist , [ AttIndex ( <NUM_LIT:2> ) , AttIndex ( <NUM_LIT:1> ) ] ) <EOL> def test_shuffle_before_distinct ( self ) : <EOL> query = """<STR_LIT>""" <EOL> pp = self . get_physical_plan ( query ) <EOL> self . assertEquals ( self . get_count ( pp , Distinct ) , <NUM_LIT:2> ) <EOL> first = True <EOL> for op in pp . walk ( ) : <EOL> if isinstance ( op , Distinct ) : <EOL> self . assertIsInstance ( op . input , MyriaShuffleConsumer ) <EOL> self . assertIsInstance ( op . input . input , MyriaShuffleProducer ) <EOL> break <EOL> def test_shuffle_before_difference ( self ) : <EOL> query = """<STR_LIT>""" <EOL> pp = self . get_physical_plan ( query ) <EOL> self . assertEquals ( self . get_count ( pp , Difference ) , <NUM_LIT:1> ) <EOL> for op in pp . walk ( ) : <EOL> if isinstance ( op , Difference ) : <EOL> self . assertIsInstance ( op . left , MyriaShuffleConsumer ) <EOL> self . assertIsInstance ( op . left . input , MyriaShuffleProducer ) <EOL> self . assertIsInstance ( op . right , MyriaShuffleConsumer ) <EOL> self . assertIsInstance ( op . right . input , MyriaShuffleProducer ) <EOL> def test_bug_240_broken_remove_unused_columns_rule ( self ) : <EOL> query = """<STR_LIT>""" <EOL> pp = self . execute_query ( query , output = '<STR_LIT>' ) <EOL> def test_broadcast_cardinality_right ( self ) : <EOL> query = """<STR_LIT>""" . format ( x = self . x_key , y = self . y_key , z = self . z_key ) <EOL> pp = self . get_physical_plan ( query ) <EOL> counter = <NUM_LIT:0> <EOL> for op in pp . walk ( ) : <EOL> if isinstance ( op , CrossProduct ) : <EOL> counter += <NUM_LIT:1> <EOL> self . assertIsInstance ( op . right , MyriaBroadcastConsumer ) <EOL> self . assertEquals ( counter , <NUM_LIT:1> ) <EOL> def test_broadcast_cardinality_left ( self ) : <EOL> query = """<STR_LIT>""" . format ( x = self . x_key , y = self . y_key , z = self . z_key ) <EOL> pp = self . get_physical_plan ( query ) <EOL> counter = <NUM_LIT:0> <EOL> for op in pp . walk ( ) : <EOL> if isinstance ( op , CrossProduct ) : <EOL> counter += <NUM_LIT:1> <EOL> self . assertIsInstance ( op . left , MyriaBroadcastConsumer ) <EOL> self . assertEquals ( counter , <NUM_LIT:1> ) <EOL> def test_broadcast_cardinality_with_agg ( self ) : <EOL> query = """<STR_LIT>""" . format ( x = self . x_key , y = self . y_key , z = self . z_key ) <EOL> pp = self . get_physical_plan ( query ) <EOL> counter = <NUM_LIT:0> <EOL> for op in pp . walk ( ) : <EOL> if isinstance ( op , CrossProduct ) : <EOL> counter += <NUM_LIT:1> <EOL> self . assertIsInstance ( op . left , MyriaBroadcastConsumer ) <EOL> self . assertEquals ( counter , <NUM_LIT:1> ) <EOL> def test_relation_cardinality ( self ) : <EOL> query = """<STR_LIT>""" . format ( x = self . x_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> self . assertIsInstance ( lp , Sequence ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( lp . children ( ) ) ) <EOL> self . assertEquals ( sum ( self . x_data . values ( ) ) ** <NUM_LIT:2> , <EOL> lp . children ( ) [ <NUM_LIT:0> ] . num_tuples ( ) ) <EOL> def test_relation_physical_cardinality ( self ) : <EOL> query = """<STR_LIT>""" . format ( x = self . x_key ) <EOL> pp = self . get_physical_plan ( query ) <EOL> self . assertEquals ( sum ( self . x_data . values ( ) ) ** <NUM_LIT:2> , <EOL> pp . num_tuples ( ) ) <EOL> def test_catalog_cardinality ( self ) : <EOL> self . assertEquals ( sum ( self . x_data . values ( ) ) , <EOL> self . db . num_tuples ( self . x_key ) ) <EOL> self . assertEquals ( sum ( self . y_data . values ( ) ) , <EOL> self . db . num_tuples ( self . y_key ) ) <EOL> self . assertEquals ( sum ( self . z_data . values ( ) ) , <EOL> self . db . num_tuples ( self . z_key ) ) <EOL> def test_groupby_to_distinct ( self ) : <EOL> query = """<STR_LIT>""" . format ( x = self . x_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> self . assertEquals ( self . get_count ( lp , GroupBy ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( lp , Distinct ) , <NUM_LIT:0> ) <EOL> pp = self . logical_to_physical ( copy . deepcopy ( lp ) ) <EOL> self . assertEquals ( self . get_count ( pp , GroupBy ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , Distinct ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . db . evaluate ( lp ) , self . db . evaluate ( pp ) ) <EOL> def test_groupby_to_lesser_groupby ( self ) : <EOL> query = """<STR_LIT>""" . format ( x = self . x_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> self . assertEquals ( self . get_count ( lp , GroupBy ) , <NUM_LIT:1> ) <EOL> for op in lp . walk ( ) : <EOL> if isinstance ( op , GroupBy ) : <EOL> self . assertEquals ( len ( op . grouping_list ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( len ( op . aggregate_list ) , <NUM_LIT:2> ) <EOL> pp = self . logical_to_physical ( copy . deepcopy ( lp ) ) <EOL> self . assertEquals ( self . get_count ( pp , GroupBy ) , <NUM_LIT:2> ) <EOL> for op in pp . walk ( ) : <EOL> if isinstance ( op , GroupBy ) : <EOL> self . assertEquals ( len ( op . grouping_list ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( len ( op . aggregate_list ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . db . evaluate ( lp ) , self . db . evaluate ( pp ) ) <EOL> def __run_uda_test ( self , uda_state = None ) : <EOL> scan = Scan ( self . x_key , self . x_scheme ) <EOL> init_ex = expression . NumericLiteral ( <NUM_LIT:0> ) <EOL> update_ex = expression . PLUS ( expression . NamedStateAttributeRef ( "<STR_LIT:value>" ) , <EOL> AttIndex ( <NUM_LIT:1> ) ) <EOL> emit_ex = expression . UdaAggregateExpression ( <EOL> expression . NamedStateAttributeRef ( "<STR_LIT:value>" ) , uda_state ) <EOL> statemods = [ StateVar ( "<STR_LIT:value>" , init_ex , update_ex ) ] <EOL> log_gb = GroupBy ( [ AttIndex ( <NUM_LIT:0> ) ] , [ emit_ex ] , scan , statemods ) <EOL> lp = StoreTemp ( '<STR_LIT>' , log_gb ) <EOL> pp = self . logical_to_physical ( copy . deepcopy ( lp ) ) <EOL> self . db . evaluate ( lp ) <EOL> log_result = self . db . get_temp_table ( '<STR_LIT>' ) <EOL> self . db . delete_temp_table ( '<STR_LIT>' ) <EOL> self . db . evaluate ( pp ) <EOL> phys_result = self . db . get_temp_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( log_result , phys_result ) <EOL> self . assertEquals ( len ( log_result ) , <NUM_LIT:15> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:1> ) <EOL> return pp <EOL> def test_non_decomposable_uda ( self ) : <EOL> """<STR_LIT>""" <EOL> pp = self . __run_uda_test ( ) <EOL> for op in pp . walk ( ) : <EOL> if isinstance ( op , MyriaShuffleProducer ) : <EOL> self . assertEquals ( op . hash_columns , [ AttIndex ( <NUM_LIT:0> ) ] ) <EOL> self . assertEquals ( self . get_count ( op , GroupBy ) , <NUM_LIT:0> ) <EOL> def test_decomposable_uda ( self ) : <EOL> """<STR_LIT>""" <EOL> lemits = [ expression . UdaAggregateExpression ( <EOL> expression . NamedStateAttributeRef ( "<STR_LIT:value>" ) ) ] <EOL> remits = copy . deepcopy ( lemits ) <EOL> init_ex = expression . NumericLiteral ( <NUM_LIT:0> ) <EOL> update_ex = expression . PLUS ( expression . NamedStateAttributeRef ( "<STR_LIT:value>" ) , <EOL> AttIndex ( <NUM_LIT:1> ) ) <EOL> lstatemods = [ StateVar ( "<STR_LIT:value>" , init_ex , update_ex ) ] <EOL> rstatemods = copy . deepcopy ( lstatemods ) <EOL> uda_state = expression . DecomposableAggregateState ( <EOL> lemits , lstatemods , remits , rstatemods ) <EOL> pp = self . __run_uda_test ( uda_state ) <EOL> self . assertEquals ( self . get_count ( pp , GroupBy ) , <NUM_LIT:2> ) <EOL> for op in pp . walk ( ) : <EOL> if isinstance ( op , MyriaShuffleProducer ) : <EOL> self . assertEquals ( op . hash_columns , [ AttIndex ( <NUM_LIT:0> ) ] ) <EOL> self . assertEquals ( self . get_count ( op , GroupBy ) , <NUM_LIT:1> ) <EOL> def test_successful_append ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( x = self . x_key ) <EOL> lp = self . get_logical_plan ( query , apply_chaining = False ) <EOL> self . assertEquals ( self . get_count ( lp , ScanTemp ) , <NUM_LIT:5> ) <EOL> self . assertEquals ( self . get_count ( lp , StoreTemp ) , <NUM_LIT:4> ) <EOL> self . assertEquals ( self . get_count ( lp , AppendTemp ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( lp , Store ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( lp , Scan ) , <NUM_LIT:1> ) <EOL> pp = self . logical_to_physical ( copy . deepcopy ( lp ) ) <EOL> self . assertEquals ( self . get_count ( pp , ScanTemp ) , <NUM_LIT:4> ) <EOL> self . assertEquals ( self . get_count ( pp , StoreTemp ) , <NUM_LIT:3> ) <EOL> self . assertEquals ( self . get_count ( pp , AppendTemp ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , Store ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , Scan ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . db . evaluate ( lp ) , self . db . evaluate ( pp ) ) <EOL> def test_failed_append ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( x = self . x_key ) <EOL> lp = self . get_logical_plan ( query , dead_code_elimination = False ) <EOL> self . assertEquals ( self . get_count ( lp , AppendTemp ) , <NUM_LIT:0> ) <EOL> pp = self . logical_to_physical ( copy . deepcopy ( lp ) ) <EOL> self . assertEquals ( self . get_count ( pp , AppendTemp ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . db . evaluate ( lp ) , self . db . evaluate ( pp ) ) <EOL> def test_push_work_into_sql ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( x = self . x_key ) <EOL> pp = self . get_physical_plan ( query , push_sql = True ) <EOL> self . assertEquals ( self . get_count ( pp , Operator ) , <NUM_LIT:2> ) <EOL> self . assertTrue ( isinstance ( pp . input , MyriaQueryScan ) ) <EOL> expected = collections . Counter ( [ ( a , c ) for ( a , b , c ) in self . x_data ] ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_push_work_into_sql_2 ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( x = self . x_key ) <EOL> pp = self . get_physical_plan ( query , push_sql = True ) <EOL> self . assertEquals ( self . get_count ( pp , Operator ) , <NUM_LIT:2> ) <EOL> self . assertTrue ( isinstance ( pp . input , MyriaQueryScan ) ) <EOL> expected = collections . Counter ( [ ( a , c ) <EOL> for ( a , b , c ) in self . x_data <EOL> if b < <NUM_LIT:5> ] ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_no_push_when_shuffle ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( x = self . x_key , y = self . y_key ) <EOL> pp = self . get_physical_plan ( query , push_sql = True ) <EOL> self . assertEquals ( self . get_count ( pp , Join ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaQueryScan ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . get_count ( pp , Apply ) , <NUM_LIT:0> ) <EOL> expected = collections . Counter ( [ ( a , f ) <EOL> for ( a , b , c ) in self . x_data <EOL> for ( d , e , f ) in self . y_data <EOL> if b == e ] ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_table ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_no_push_when_random ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( x = self . x_key , y = self . y_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> self . assertEquals ( self . get_count ( lp , Select ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( lp , CrossProduct ) , <NUM_LIT:1> ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , Select ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , CrossProduct ) , <NUM_LIT:1> ) <EOL> for op in pp . walk ( ) : <EOL> if isinstance ( op , Select ) : <EOL> self . assertIsInstance ( op . input , MyriaSplitConsumer ) <EOL> self . assertIsInstance ( op . input . input . input , CrossProduct ) <EOL> def test_partitioning_from_shuffle ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( x = self . x_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> tail = lp . args [ <NUM_LIT:0> ] . input <EOL> lp . args [ <NUM_LIT:0> ] . input = Shuffle ( tail , [ AttIndex ( <NUM_LIT:0> ) ] ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( pp . partitioning ( ) . hash_partitioned , <EOL> frozenset ( [ AttIndex ( <NUM_LIT:0> ) ] ) ) <EOL> def test_partitioning_from_scan ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( pp . partitioning ( ) . hash_partitioned , <EOL> self . part_partition . hash_partitioned ) <EOL> def test_repartitioning ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> tail = lp . args [ <NUM_LIT:0> ] . input <EOL> lp . args [ <NUM_LIT:0> ] . input = Shuffle ( tail , [ AttIndex ( <NUM_LIT:2> ) ] ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( pp . partitioning ( ) . hash_partitioned , <EOL> frozenset ( [ AttIndex ( <NUM_LIT:2> ) ] ) ) <EOL> def test_remove_shuffle ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:0> ) <EOL> def test_do_not_remove_shuffle_left ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:1> ) <EOL> def test_do_not_remove_shuffle_both ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:2> ) <EOL> def test_apply_removes_partitioning ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( pp . partitioning ( ) . hash_partitioned , <EOL> frozenset ( ) ) <EOL> def test_apply_maintains_partitioning ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( pp . partitioning ( ) . hash_partitioned , <EOL> frozenset ( [ AttIndex ( <NUM_LIT:0> ) ] ) ) <EOL> def test_swapping_apply_maintains_partitioning ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( pp . partitioning ( ) . hash_partitioned , <EOL> frozenset ( [ AttIndex ( <NUM_LIT:1> ) ] ) ) <EOL> def test_projecting_join_maintains_partitioning ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( pp . partitioning ( ) . hash_partitioned , <EOL> frozenset ( [ AttIndex ( <NUM_LIT:0> ) ] ) ) <EOL> def test_no_shuffle_for_partitioned_distinct ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaDupElim ) , <NUM_LIT:1> ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_table ( '<STR_LIT>' ) <EOL> expected = dict ( [ ( ( h , ) , <NUM_LIT:1> ) for _ , h , _ in self . part_data ] ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_no_shuffle_for_partitioned_groupby ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaGroupBy ) , <NUM_LIT:1> ) <EOL> def test_partition_aware_groupby_into_sql ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp , push_sql = True , <EOL> push_sql_grouping = True ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaGroupBy ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaQueryScan ) , <NUM_LIT:1> ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_table ( '<STR_LIT>' ) <EOL> temp = dict ( [ ( h , sys . maxsize ) for _ , h , _ in self . part_data ] ) <EOL> for _ , h , i in self . part_data : <EOL> temp [ h ] = min ( temp [ h ] , i ) <EOL> expected = dict ( ( ( h , i ) , <NUM_LIT:1> ) for h , i in temp . items ( ) ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_partition_aware_distinct_into_sql ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp , push_sql = True ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaGroupBy ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaDupElim ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaQueryScan ) , <NUM_LIT:1> ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_table ( '<STR_LIT>' ) <EOL> expected = dict ( [ ( ( h , ) , <NUM_LIT:1> ) for _ , h , _ in self . part_data ] ) <EOL> self . assertEquals ( result , expected ) <EOL> def test_push_half_groupby_into_sql ( self ) : <EOL> """<STR_LIT>""" <EOL> query = """<STR_LIT>""" . format ( part = self . part_key ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp , push_sql = True , <EOL> push_sql_grouping = True ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleConsumer ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaShuffleProducer ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaGroupBy ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaQueryScan ) , <NUM_LIT:1> ) <EOL> self . db . evaluate ( pp ) <EOL> result = self . db . get_table ( '<STR_LIT>' ) <EOL> temp = dict ( [ ( i , sys . maxsize ) for _ , _ , i in self . part_data ] ) <EOL> for _ , h , i in self . part_data : <EOL> temp [ i ] = min ( temp [ i ] , h ) <EOL> expected = dict ( ( ( k , v ) , <NUM_LIT:1> ) for k , v in temp . items ( ) ) <EOL> self . assertEquals ( result , expected ) <EOL> def _check_aggregate_functions_pushed ( <EOL> self , <EOL> func , <EOL> expected , <EOL> override = False ) : <EOL> if override : <EOL> agg = func <EOL> else : <EOL> agg = "<STR_LIT>" . format ( func = func ) <EOL> query = """<STR_LIT>""" . format ( part = self . part_key , agg = agg ) <EOL> lp = self . get_logical_plan ( query ) <EOL> pp = self . logical_to_physical ( lp , push_sql = True , <EOL> push_sql_grouping = True ) <EOL> self . assertEquals ( self . get_count ( pp , MyriaQueryScan ) , <NUM_LIT:1> ) <EOL> for op in pp . walk ( ) : <EOL> if isinstance ( op , MyriaQueryScan ) : <EOL> self . assertTrue ( re . search ( expected , op . sql ) ) <EOL> def test_aggregate_AVG_pushed ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _check_aggregate_functions_pushed ( <EOL> aggregate . AVG . __name__ , '<STR_LIT>' ) <EOL> def test_aggregate_STDDEV_pushed ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _check_aggregate_functions_pushed ( <EOL> aggregate . STDEV . __name__ , '<STR_LIT>' ) <EOL> def test_aggregate_COUNTALL_pushed ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _check_aggregate_functions_pushed ( <EOL> '<STR_LIT>' , r'<STR_LIT>' , True ) </s>
