private void addDataSources ( Table table , int hostId , List < Pair < Integer , Integer > > localPartitionsToSites , Set < Integer > partitionsInUse , final ExportDataProcessor processor , final long genId , boolean isCatalogUpdate ) { for ( Pair < Integer , Integer > partitionAndSiteId : localPartitionsToSites ) { /* * IOException can occur if there is a problem * with the persistent aspects of the datasource storage */ int partition = partitionAndSiteId . getFirst ( ) ; int siteId = partitionAndSiteId . getSecond ( ) ; synchronized ( m_dataSourcesByPartition ) { try { Map < String , ExportDataSource > dataSourcesForPartition = m_dataSourcesByPartition . get ( partition ) ; if ( dataSourcesForPartition == null ) { dataSourcesForPartition = new HashMap < String , ExportDataSource > ( ) ; m_dataSourcesByPartition . put ( partition , dataSourcesForPartition ) ; } final String key = table . getTypeName ( ) ; if ( ! dataSourcesForPartition . containsKey ( key ) ) { ExportDataSource exportDataSource = new ExportDataSource ( this , processor , "database" , key , partition , siteId , genId , table . getColumns ( ) , table . getPartitioncolumn ( ) , m_directory . getPath ( ) ) ; int migrateBatchSize = CatalogUtil . getPersistentMigrateBatchSize ( key ) ; exportDataSource . setupMigrateRowsDeleter ( migrateBatchSize ) ; if ( exportLog . isDebugEnabled ( ) ) { exportLog . debug ( "Creating ExportDataSource for table in catalog " + key + " partition " + partition + " site " + siteId ) ; } dataSourcesForPartition . put ( key , exportDataSource ) ; if ( isCatalogUpdate ) { exportDataSource . updateCatalog ( table , genId ) ; } } else { // Associate any existing EDS to the export client in the new processor ExportDataSource eds = dataSourcesForPartition . get ( key ) ; ExportClientBase client = processor . getExportClient ( key ) ; if ( client != null ) { // Associate to an existing export client eds . setClient ( client ) ; eds . setRunEveryWhere ( client . isRunEverywhere ( ) ) ; } else { // Reset to no export client eds . setClient ( null ) ; eds . setRunEveryWhere ( false ) ; } // Mark in catalog only if partition is in use eds . markInCatalog ( partitionsInUse . contains ( partition ) ) ; if ( isCatalogUpdate ) { eds . updateCatalog ( table , genId ) ; } } } catch ( IOException e ) { VoltDB . crashLocalVoltDB ( "Error creating datasources for table " + table . getTypeName ( ) + " host id " + hostId , true , e ) ; } } } }
Add datasources for a catalog table in all partitions