#NEWLINE# @ skip_if_cached #NEWLINE# def run_CreatePCollectionView( self , transform_node ) : #NEWLINE# #INDENT# transform = transform_node . transform #NEWLINE# view = transform . view #NEWLINE# values = self . _cache . get_pvalue ( transform_node . inputs [ 0 ] ) #NEWLINE# if isinstance ( view , SingletonPCollectionView ) : #NEWLINE# #INDENT# ( has_default , default_value ) = view . _view_options ( ) #NEWLINE# if ( len ( values ) == 0 ) : #NEWLINE# #INDENT# if has_default : #NEWLINE# #INDENT# result = default_value #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# result = EmptySideInput ( ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# if ( len ( values ) == 1 ) : #NEWLINE# #INDENT# result = values [ 0 ] . value #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# raise ValueError ( ( 'PCollection with more than one element accessed as a singleton view: %s.' % view ) ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# if isinstance ( view , IterablePCollectionView ) : #NEWLINE# #INDENT# result = [ v . value for v in values ] #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if isinstance ( has_default , ListPCollectionView ) : #NEWLINE# #INDENT# result = [ v . value for v in values ] #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# raise NotImplementedError #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# self . _cache . cache_output ( transform_node , result )