private void handlePartitioningTurnedOffOrOn ( SortedSetMultimap < TableContext , TableRuntimeContext > reconstructedPartitions ) { for ( TableContext tableContext : reconstructedPartitions . keySet ( ) ) { final SortedSet < TableRuntimeContext > partitions = reconstructedPartitions . get ( tableContext ) ; final TableRuntimeContext lastPartition = partitions . last ( ) ; final TableContext sourceTableContext = lastPartition . getSourceTableContext ( ) ; Utils . checkState ( sourceTableContext . equals ( tableContext ) , String . format ( "Source table context for %s should match TableContext map key of %s" , lastPartition . getDescription ( ) , tableContext . getQualifiedName ( ) ) ) ; final boolean partitioningTurnedOff = lastPartition . isPartitioned ( ) && sourceTableContext . getPartitioningMode ( ) == PartitioningMode . DISABLED ; final boolean partitioningTurnedOn = ! lastPartition . isPartitioned ( ) && sourceTableContext . isPartitionable ( ) && sourceTableContext . getPartitioningMode ( ) != PartitioningMode . DISABLED ; if ( ! partitioningTurnedOff && ! partitioningTurnedOn ) { continue ; } final Map < String , String > nextStartingOffsets = new HashMap <> ( ) ; final Map < String , String > nextMaxOffsets = new HashMap <> ( ) ; final int newPartitionSequence = lastPartition . getPartitionSequence ( ) > 0 ? lastPartition . getPartitionSequence ( ) + 1 : 1 ; if ( partitioningTurnedOff ) { LOG . info ( "Table {} has switched from partitioned to non-partitioned; partition sequence {} will be the last (with" + " no max offsets)" , sourceTableContext . getQualifiedName ( ) , newPartitionSequence ) ; lastPartition . getStartingPartitionOffsets ( ) . forEach ( ( col , off ) -> { String basedOnStartOffset = lastPartition . generateNextPartitionOffset ( col , off ) ; nextStartingOffsets . put ( col , basedOnStartOffset ) ; } ) ; } else if ( partitioningTurnedOn ) { lastPartition . getStartingPartitionOffsets ( ) . forEach ( ( col , off ) -> { String basedOnStoredOffset = lastPartition . getInitialStoredOffsets ( ) . get ( col ) ; nextStartingOffsets . put ( col , basedOnStoredOffset ) ; } ) ; nextStartingOffsets . forEach ( ( col , off ) -> nextMaxOffsets . put ( col , lastPartition . generateNextPartitionOffset ( col , off ) ) ) ; if ( ! reconstructedPartitions . remove ( sourceTableContext , lastPartition ) ) { throw new IllegalStateException ( String . format ( "Failed to remove partition %s for table %s in switching partitioning from off to on" , lastPartition . getDescription ( ) , sourceTableContext . getQualifiedName ( ) ) ) ; } LOG . info ( "Table {} has switched from non-partitioned to partitioned; using last stored offsets as the starting" + " offsets for the new partition {}" , sourceTableContext . getQualifiedName ( ) , newPartitionSequence ) ; } final TableRuntimeContext nextPartition = new TableRuntimeContext ( sourceTableContext , lastPartition . isUsingNonIncrementalLoad ( ) , ( lastPartition . isPartitioned ( ) && ! partitioningTurnedOff ) || partitioningTurnedOn , newPartitionSequence , nextStartingOffsets , nextMaxOffsets ) ; reconstructedPartitions . put ( sourceTableContext , nextPartition ) ; } }
Checks whether any tables have had partitioning turned off or not and updates the partition map appropriately