merge

table.merge

                                                                                
 Documentation                                                                  
                          Create a table from other tables and/or arrays.       
                                                                                
 Origin                                                                         
                          Authors   Markus Binsteiner (markus@frkl.io)          
                                                                                
 Context                                                                        
                          Tags         core                                     
                          Labels       package: kiara_modules.core              
                          References   source_repo:                             
                                       https://github.com/DHARPA-Project/kia…   
                                       documentation:                           
                                       https://dharpa.org/kiara_modules.core/   
                                       module_doc:                              
                                       https://dharpa.org/kiara_modules.core…   
                                       source_url:                              
                                       https://github.com/DHARPA-Project/kia…   
                                                                                
 Module config                                                                  
                          Field          Type     Description        Required   
                         ─────────────────────────────────────────────────────  
                          constants      object   Value constants    no         
                                                  for this module.              
                          defaults       object   Value defaults     no         
                                                  for this module.              
                          input_schema   object   A dict             yes        
                                                  describing the                
                                                  inputs for this               
                                                  merge process.                
                                                                                
 Module config          -- no config --                                         
 Python class                                                                   
                          class_name    MergeTableModule                        
                          module_name   kiara_modules.core.table                
                          full_name     kiara_modules.core.table.MergeTableM…   
                                                                                
 Processing source code  ─────────────────────────────────────────────────────  
                          def process(self, inputs: ValueSet, outputs: Value…   
                                                                                
                              import pyarrow as pa                              
                                                                                
                              input_schema: typing.Dict[str, typing.Any] = s…   
                                  "input_schema"                                
                              )                                                 
                                                                                
                              sources = {}                                      
                              for field_name in input_schema.keys():            
                                  sources[field_name] = inputs.get_value_dat…   
                                                                                
                              len_dict = {}                                     
                              arrays = []                                       
                              column_names = []                                 
                              for source_key, table_or_column in sources.ite…   
                                                                                
                                  if isinstance(table_or_column, pa.Table):     
                                      rows = table_or_column.num_rows           
                                      for name in table_or_column.schema.nam…   
                                          column = table_or_column.column(na…   
                                          arrays.append(column)                 
                                          column_names.append(name)             
                                                                                
                                  elif isinstance(table_or_column, (pa.Array…   
                                      rows = len(table_or_column)               
                                      arrays.append(table_or_column)            
                                      column_names.append(source_key)           
                                  else:                                         
                                      raise KiaraProcessingException(           
                                          f"Can't merge table: invalid type …   
                                      )                                         
                                                                                
                                  len_dict[source_key] = rows                   
                                                                                
                              all_rows = None                                   
                              for source_key, rows in len_dict.items():         
                                  if all_rows is None:                          
                                      all_rows = rows                           
                                  else:                                         
                                      if all_rows != rows:                      
                                          all_rows = None                       
                                          break                                 
                                                                                
                              if all_rows is None:                              
                                  len_str = ""                                  
                                  for name, rows in len_dict.items():           
                                      len_str = f" {name} ({rows})"             
                                                                                
                                  raise KiaraProcessingException(               
                                      f"Can't merge table, sources have diff…   
                                  )                                             
                                                                                
                              table = pa.Table.from_arrays(arrays=arrays, na…   
                                                                                
                              outputs.set_value("table", table)                 
                                                                                
                         ─────────────────────────────────────────────────────