private void initializeGenerationFromDisk ( final CatalogMap < Connector > connectors , final ExportDataProcessor processor , File [ ] files , List < Pair < Integer , Integer > > localPartitionsToSites , long genId ) { List < Integer > onDiskPartitions = new ArrayList < Integer > ( ) ; NavigableSet < Table > streams = CatalogUtil . getExportTablesExcludeViewOnly ( connectors ) ; Set < String > exportedTables = new HashSet <> ( ) ; for ( Table stream : streams ) { exportedTables . add ( stream . getTypeName ( ) ) ; } /* * Find all the data files. Once one is found, extract the nonce * and check for any advertisements related to the data files. If * there are orphaned advertisements, delete them. */ Map < String , File > dataFiles = new HashMap <> ( ) ; for ( File data : files ) { if ( data . getName ( ) . endsWith ( ".pbd" ) ) { PbdSegmentName pbdName = PbdSegmentName . parseFile ( exportLog , data ) ; if ( pbdName . m_nonce != null ) { String nonce = pbdName . m_nonce ; String streamName = getStreamNameFromNonce ( nonce ) ; if ( exportedTables . contains ( streamName ) ) { dataFiles . put ( nonce , data ) ; } else { // ENG-15740, stream can be dropped while node is offline, delete .pbd files // if stream is no longer in catalog data . delete ( ) ; } } else if ( pbdName . m_result == Result . NOT_PBD ) { exportLog . warn ( data . getAbsolutePath ( ) + " is not a PBD file." ) ; } else if ( pbdName . m_result == Result . INVALID_NAME ) { exportLog . warn ( data . getAbsolutePath ( ) + " doesn't have valid PBD name." ) ; } } } for ( File ad : files ) { if ( ad . getName ( ) . endsWith ( ".ad" ) ) { String nonce = getNonceFromAdFile ( ad ) ; File dataFile = dataFiles . get ( nonce ) ; if ( dataFile != null ) { try { addDataSource ( ad , localPartitionsToSites , onDiskPartitions , processor , genId ) ; } catch ( IOException e ) { VoltDB . crashLocalVoltDB ( "Error intializing export datasource " + ad , true , e ) ; } } else { //Delete ads that have no data ad . delete ( ) ; } } } // Count unique partitions only Set < Integer > allLocalPartitions = localPartitionsToSites . stream ( ) . map ( p -> p . getFirst ( ) ) . collect ( Collectors . toSet ( ) ) ; Set < Integer > onDIskPartitionsSet = new HashSet < Integer > ( onDiskPartitions ) ; onDIskPartitionsSet . removeAll ( allLocalPartitions ) ; // One export mailbox per node, since we only keep one generation if ( ! onDIskPartitionsSet . isEmpty ( ) ) { createAckMailboxesIfNeeded ( onDIskPartitionsSet ) ; } }
Initialize generation from disk creating data sources from the PBD files .