<commit_before>package org.sagebionetworks.tool.migration.v4;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.sagebionetworks.client.SynapseAdminClient;
import org.sagebionetworks.client.exceptions.SynapseException;
import org.sagebionetworks.repo.model.migration.MigrationType;
import org.sagebionetworks.repo.model.migration.MigrationTypeCount;
import org.sagebionetworks.repo.model.status.StackStatus;
import org.sagebionetworks.repo.model.status.StatusEnum;
import org.sagebionetworks.schema.adapter.JSONObjectAdapterException;
import org.sagebionetworks.tool.migration.v3.DeleteWorker;
import org.sagebionetworks.tool.migration.v3.SynapseClientFactory;
import org.sagebionetworks.tool.migration.v3.stream.BufferedRowMetadataReader;
import org.sagebionetworks.tool.migration.v3.stream.BufferedRowMetadataWriter;
import org.sagebionetworks.tool.migration.v4.delta.DeltaFinder;
import org.sagebionetworks.tool.migration.v4.delta.DeltaRanges;
import org.sagebionetworks.tool.migration.v4.utils.MigrationTypeCountDiff;
import org.sagebionetworks.tool.migration.v4.utils.ToolMigrationUtils;
import org.sagebionetworks.tool.migration.v4.utils.TypeToMigrateMetadata;
import org.sagebionetworks.tool.progress.BasicProgress;

/**
 * The V4 migration client.
 *
 */
public class MigrationClient {
	
	static private Log log = LogFactory.getLog(MigrationClient.class);

	SynapseClientFactory factory;
	ExecutorService threadPool;
	List<Exception> deferredExceptions;
	final int MAX_DEFERRED_EXCEPTIONS = 10;
	
	/**
	 * New migration client.
	 * @param factory
	 */
	public MigrationClient(SynapseClientFactory factory) {
		if(factory == null) throw new IllegalArgumentException("Factory cannot be null");
		this.factory = factory;
		threadPool = Executors.newFixedThreadPool(1);
		deferredExceptions = new ArrayList<Exception>();
	}

	/**
	 * Migrate all data from the source to destination.
	 * 
	 * @throws Exception 
	 */
	public boolean migrate(int maxRetries, long batchSize, long timeoutMS, int retryDenominator) throws Exception {
		boolean failed = false;
		try {
			// First set the destination stack status to down
			setDestinationStatus(StatusEnum.READ_ONLY, "Staging is down for data migration");
			this.migrateAllTypes(batchSize, timeoutMS, retryDenominator);
		} catch (Exception e) {
			log.error("Migration failed", e);
		} finally {
			// After migration is complete, re-enable read/write
			setDestinationStatus(StatusEnum.READ_WRITE, "Synapse is ready for read/write");
		}
		return failed;
	}
	
	/**
	 * 
	 * @param status
	 * @param message
	 * @throws SynapseException
	 * @throws JSONObjectAdapterException
	 */
	public void setDestinationStatus(StatusEnum status, String message) throws SynapseException, JSONObjectAdapterException {
		setStatus(this.factory.createNewDestinationClient(), status, message);
	}
	
	/**
	 * 
	 * @param status
	 * @param message
	 * @throws SynapseException
	 * @throws JSONObjectAdapterException
	 */
	public void setSourceStatus(StatusEnum status, String message) throws SynapseException, JSONObjectAdapterException {
		setStatus(this.factory.createNewSourceClient(), status, message);
	}
	
	/**
	 * 
	 * @param client
	 * @param status
	 * @param message
	 * @throws SynapseException 
	 * @throws JSONObjectAdapterException 
	 */
	private static void setStatus(SynapseAdminClient client, StatusEnum status, String message) throws JSONObjectAdapterException, SynapseException{
		StackStatus destStatus = client.getCurrentStackStatus();
		destStatus.setStatus(status);
		destStatus.setCurrentMessage(message);
		destStatus = client.updateCurrentStackStatus(destStatus);
	}
	
	/**
	 * Migrate all types.
	 * @param batchSize - Max batch size
	 * @param timeoutMS - max time to wait for a deamon job.
	 * @param retryDenominator - how to divide a batch into sub-batches when errors occur.
	 * @throws Exception
	 */
	public void migrateAllTypes(long batchSize, long timeoutMS, int retryDenominator) throws Exception {

		SynapseAdminClient source = factory.createNewSourceClient();
		SynapseAdminClient destination = factory.createNewDestinationClient();
		
		// Get the counts for all type from both the source and destination
		List<MigrationTypeCount> startSourceCounts = ToolMigrationUtils.getTypeCounts(source);
		List<MigrationTypeCount> startDestCounts = ToolMigrationUtils.getTypeCounts(destination);
		log.info("Starting diffs in counts:");
		printDiffsInCounts(startSourceCounts, startDestCounts);
		
		// Get the primary types for src and dest
		List<MigrationType> srcPrimaryTypes = source.getPrimaryTypes().getList();
		List<MigrationType> destPrimaryTypes = destination.getPrimaryTypes().getList();
		// Only migrate the src primary types that are at destination
		List<MigrationType> primaryTypesToMigrate = new LinkedList<MigrationType>();
		for (MigrationType pt: destPrimaryTypes) {
			if (srcPrimaryTypes.contains(pt)) {
				primaryTypesToMigrate.add(pt);
			}
		}
		
		// Get the metadata for the (primary) types to migrate
		List<TypeToMigrateMetadata> typesToMigrateMetadata = ToolMigrationUtils.buildTypeToMigrateMetadata(startSourceCounts, startDestCounts, primaryTypesToMigrate);
		
		// Do the actual migration.
		migrateAll(batchSize, timeoutMS, retryDenominator, typesToMigrateMetadata);
		
		// Print the final counts
		List<MigrationTypeCount> endSourceCounts = ToolMigrationUtils.getTypeCounts(source);
		List<MigrationTypeCount> endDestCounts = ToolMigrationUtils.getTypeCounts(destination);
		log.info("Ending diffs in  counts:");
		printDiffsInCounts(endSourceCounts, endDestCounts);
		
	}

	/**
	 * Does the actaul migration work.
	 * @param batchSize
	 * @param timeoutMS
	 * @param retryDenominator
	 * @param primaryTypes
	 * @throws Exception
	 */
	private void migrateAll(long batchSize, long timeoutMS,	int retryDenominator, List<TypeToMigrateMetadata> primaryTypes)
			throws Exception {

		// Each migration uses a different salt (same for each type)
		String salt = UUID.randomUUID().toString();
		
		List<DeltaData> deltaList = new LinkedList<DeltaData>();
		for (TypeToMigrateMetadata tm: primaryTypes) {
			DeltaData dd = calculateDeltaForType(tm, salt, batchSize);<commit_msg>Add table checksum if src RO + fix.<commit_after>package org.sagebionetworks.tool.migration.v4;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.sagebionetworks.client.SynapseAdminClient;
import org.sagebionetworks.client.exceptions.SynapseException;
import org.sagebionetworks.repo.model.migration.MigrationType;
import org.sagebionetworks.repo.model.migration.MigrationTypeCount;
import org.sagebionetworks.repo.model.status.StackStatus;
import org.sagebionetworks.repo.model.status.StatusEnum;
import org.sagebionetworks.schema.adapter.JSONObjectAdapterException;
import org.sagebionetworks.tool.migration.v3.DeleteWorker;
import org.sagebionetworks.tool.migration.v3.SynapseClientFactory;
import org.sagebionetworks.tool.migration.v3.stream.BufferedRowMetadataReader;
import org.sagebionetworks.tool.migration.v3.stream.BufferedRowMetadataWriter;
import org.sagebionetworks.tool.migration.v4.delta.DeltaFinder;
import org.sagebionetworks.tool.migration.v4.delta.DeltaRanges;
import org.sagebionetworks.tool.migration.v4.utils.MigrationTypeCountDiff;
import org.sagebionetworks.tool.migration.v4.utils.ToolMigrationUtils;
import org.sagebionetworks.tool.migration.v4.utils.TypeToMigrateMetadata;
import org.sagebionetworks.tool.progress.BasicProgress;

/**
 * The V4 migration client.
 *
 */
public class MigrationClient {
	
	static private Log log = LogFactory.getLog(MigrationClient.class);

	SynapseClientFactory factory;
	ExecutorService threadPool;
	List<Exception> deferredExceptions;
	final int MAX_DEFERRED_EXCEPTIONS = 10;
	
	/**
	 * New migration client.
	 * @param factory
	 */
	public MigrationClient(SynapseClientFactory factory) {
		if(factory == null) throw new IllegalArgumentException("Factory cannot be null");
		this.factory = factory;
		threadPool = Executors.newFixedThreadPool(1);
		deferredExceptions = new ArrayList<Exception>();
	}

	/**
	 * Migrate all data from the source to destination.
	 * 
	 * @throws Exception 
	 */
	public boolean migrate(int maxRetries, long batchSize, long timeoutMS, int retryDenominator) throws Exception {
		boolean failed = false;
		try {
			// First set the destination stack status to down
			setDestinationStatus(StatusEnum.READ_ONLY, "Staging is down for data migration");
			this.migrateAllTypes(batchSize, timeoutMS, retryDenominator);
		} catch (Exception e) {
			log.error("Migration failed", e);
			failed = true;
		} finally {
			// After migration is complete, re-enable read/write
			setDestinationStatus(StatusEnum.READ_WRITE, "Synapse is ready for read/write");
		}
		return failed;
	}
	
	/**
	 * 
	 * @param status
	 * @param message
	 * @throws SynapseException
	 * @throws JSONObjectAdapterException
	 */
	public void setDestinationStatus(StatusEnum status, String message) throws SynapseException, JSONObjectAdapterException {
		setStatus(this.factory.createNewDestinationClient(), status, message);
	}
	
	/**
	 * 
	 * @param status
	 * @param message
	 * @throws SynapseException
	 * @throws JSONObjectAdapterException
	 */
	public void setSourceStatus(StatusEnum status, String message) throws SynapseException, JSONObjectAdapterException {
		setStatus(this.factory.createNewSourceClient(), status, message);
	}
	
	/**
	 * 
	 * @param client
	 * @param status
	 * @param message
	 * @throws SynapseException 
	 * @throws JSONObjectAdapterException 
	 */
	private static void setStatus(SynapseAdminClient client, StatusEnum status, String message) throws JSONObjectAdapterException, SynapseException{
		StackStatus destStatus = client.getCurrentStackStatus();
		destStatus.setStatus(status);
		destStatus.setCurrentMessage(message);
		destStatus = client.updateCurrentStackStatus(destStatus);
	}

	/**
	 * Migrate all types.
	 * @param batchSize - Max batch size
	 * @param timeoutMS - max time to wait for a deamon job.
	 * @param retryDenominator - how to divide a batch into sub-batches when errors occur.
	 * @throws Exception
	 */
	public void migrateAllTypes(long batchSize, long timeoutMS, int retryDenominator) throws Exception {

		SynapseAdminClient source = factory.createNewSourceClient();
		SynapseAdminClient destination = factory.createNewDestinationClient();
		
		// Get the counts for all type from both the source and destination
		List<MigrationTypeCount> startSourceCounts = ToolMigrationUtils.getTypeCounts(source);
		List<MigrationTypeCount> startDestCounts = ToolMigrationUtils.getTypeCounts(destination);
		log.info("Starting diffs in counts:");
		printDiffsInCounts(startSourceCounts, startDestCounts);
		
		// Get the primary types for src and dest
		List<MigrationType> srcPrimaryTypes = source.getPrimaryTypes().getList();
		List<MigrationType> destPrimaryTypes = destination.getPrimaryTypes().getList();
		// Only migrate the src primary types that are at destination
		List<MigrationType> primaryTypesToMigrate = new LinkedList<MigrationType>();
		for (MigrationType pt: destPrimaryTypes) {
			if (srcPrimaryTypes.contains(pt)) {
				primaryTypesToMigrate.add(pt);
			}
		}
		
		// Get the metadata for the (primary) types to migrate
		List<TypeToMigrateMetadata> typesToMigrateMetadata = ToolMigrationUtils.buildTypeToMigrateMetadata(startSourceCounts, startDestCounts, primaryTypesToMigrate);
		
		// Do the actual migration.
		migrateAll(batchSize, timeoutMS, retryDenominator, typesToMigrateMetadata);
		
		// Print the final counts
		List<MigrationTypeCount> endSourceCounts = ToolMigrationUtils.getTypeCounts(source);
		List<MigrationTypeCount> endDestCounts = ToolMigrationUtils.getTypeCounts(destination);
		log.info("Ending diffs in  counts:");
		printDiffsInCounts(endSourceCounts, endDestCounts);
		
		// If final sync (source is in read-only mode) then do a table checksum
		// Note: Destination is always in read-only during migration
		if (source.getCurrentStackStatus().getStatus() == StatusEnum.READ_ONLY) {
			log.info("Final migration, checking table checksums");
			boolean isChecksumDiff = false;
			for (TypeToMigrateMetadata t: typesToMigrateMetadata) {
				String srcTableChecksum = source.getChecksumForType(t.getType()).getChecksum();
				String destTableChecksum =destination.getChecksumForType(t.getType()).getChecksum();
				if (! srcTableChecksum.equals(destTableChecksum)) {
					isChecksumDiff = true;
					log.info("Table checksum difference for type: " + t);
				}
			}
			if (isChecksumDiff) {
				throw new RuntimeException("Table checksum differences in final sync.");
			}
		}
		
	}

	/**
	 * Does the actaul migration work.
	 * @param batchSize
	 * @param timeoutMS
	 * @param retryDenominator
	 * @param primaryTypes
	 * @throws Exception
	 */
	private void migrateAll(long batchSize, long timeoutMS,	int retryDenominator, List<TypeToMigrateMetadata> primaryTypes)
			throws Exception {

		// Each migration uses a different salt (same for each type)
		String salt = UUID.randomUUID().toString();
		
		List<DeltaData> deltaList = new LinkedList<DeltaData>();
		for (TypeToMigrateMetadata tm: primaryTypes) {
			DeltaData dd = calculateDeltaForType(tm, salt, batchSize);
{'fraction_non_alphanumeric': 0.091616125579736, 'fraction_numerical': 0.0019978594363182305, 'mean_word_length': 5.276757725033587, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1122052', 'n_tokens_mistral': 4377, 'n_tokens_neox': 4072, 'n_words': 953}