<commit_before>package org.sagebionetworks.table.worker;

import static org.sagebionetworks.repo.model.ACCESS_TYPE.CREATE;
import static org.sagebionetworks.repo.model.ACCESS_TYPE.DELETE;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Callable;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.sagebionetworks.StackConfiguration;
import org.sagebionetworks.asynchronous.workers.sqs.MessageUtils;
import org.sagebionetworks.asynchronous.workers.sqs.Worker;
import org.sagebionetworks.asynchronous.workers.sqs.WorkerProgress;
import org.sagebionetworks.repo.manager.NodeInheritanceManager;
import org.sagebionetworks.repo.manager.table.TableRowManager;
import org.sagebionetworks.repo.manager.trash.EntityInTrashCanException;
import org.sagebionetworks.repo.model.ConflictingUpdateException;
import org.sagebionetworks.repo.model.DatastoreException;
import org.sagebionetworks.repo.model.ObjectType;
import org.sagebionetworks.repo.model.exception.LockUnavilableException;
import org.sagebionetworks.repo.model.jdo.KeyFactory;
import org.sagebionetworks.repo.model.message.ChangeMessage;
import org.sagebionetworks.repo.model.message.ChangeType;
import org.sagebionetworks.repo.model.table.ColumnModel;
import org.sagebionetworks.repo.model.table.RowSet;
import org.sagebionetworks.repo.model.table.TableRowChange;
import org.sagebionetworks.repo.model.table.TableStatus;
import org.sagebionetworks.repo.model.table.TableUnavilableException;
import org.sagebionetworks.repo.web.NotFoundException;
import org.sagebionetworks.table.cluster.ConnectionFactory;
import org.sagebionetworks.table.cluster.TableIndexDAO;
import org.sagebionetworks.table.cluster.utils.TableModelUtils;
import org.springframework.beans.factory.annotation.Autowired;

import com.amazonaws.services.sqs.model.Message;
import com.google.common.collect.SetMultimap;

/**
 * This worker updates the index used to support the tables features. It will
 * listen to table changes and apply them to the RDS table that acts as the
 * index of the table feature.
 * 
 * @author John
 * 
 */
public class TableWorker implements Worker {

	private static final Long TRASH_FOLDER_ID = Long.parseLong(StackConfiguration.getTrashFolderEntityIdStatic());

	enum State {
		SUCCESS, UNRECOVERABLE_FAILURE, RECOVERABLE_FAILURE,
	}

	private static final long BATCH_SIZE = 16000;

	static private Logger log = LogManager.getLogger(TableWorker.class);

	@Autowired
	ConnectionFactory tableConnectionFactory;
	@Autowired
	TableRowManager tableRowManager;
	@Autowired
	StackConfiguration configuration;
	@Autowired
	NodeInheritanceManager nodeInheritanceManager;

	List<Message> messages;
	WorkerProgress workerProgress;

	public TableWorker() {
	}

	// unit testing only
	TableWorker(ConnectionFactory tableConnectionFactory, TableRowManager tableRowManager, StackConfiguration configuration,
			NodeInheritanceManager nodeInheritanceManager) {
		this.tableConnectionFactory = tableConnectionFactory;
		this.tableRowManager = tableRowManager;
		this.configuration = configuration;
		this.nodeInheritanceManager = nodeInheritanceManager;
	}

	@Override
	public void setMessages(List<Message> messages) {
		this.messages = messages;
	}

	@Override
	public void setWorkerProgress(WorkerProgress workerProgress) {
		this.workerProgress = workerProgress;
	}

	@Override
	public List<Message> call() throws Exception {
		List<Message> processedMessages = new LinkedList<Message>();
		// If the feature is disabled then we simply swallow all messages
		if (!configuration.getTableEnabled()) {
			return messages;
		}
		// process each message
		for (Message message : messages) {
			// Extract the ChangeMessage
			ChangeMessage change = MessageUtils.extractMessageBody(message);
			// We only care about entity messages here
			if (ObjectType.TABLE.equals((change.getObjectType()))) {
				if (ChangeType.DELETE.equals(change.getChangeType())) {
					// Delete the table in the index
					TableIndexDAO indexDao = tableConnectionFactory
							.getConnection(change.getObjectId());
					if (indexDao != null) {
						indexDao.deleteTable(change.getObjectId());
						indexDao.deleteStatusTable(change.getObjectId());
					}
					processedMessages.add(message);
				} else {
					// Create or update.
					String tableId = change.getObjectId();
					// make sure the table is not in the trash
					if (nodeInheritanceManager.isNodeInTrash(tableId)) {
						processedMessages.add(message);
						continue;
					}
					// this method does the real work.
					State state = createOrUpdateTable(tableId,
							change.getObjectEtag(), message);
					if (!State.RECOVERABLE_FAILURE.equals(state)) {
						// Only recoverable failures should remain in the queue.
						// All other must be removed.
						processedMessages.add(message);
					}
				}
			} else {
				// Non-table messages must be returned so they can be removed
				// from the queue.
				processedMessages.add(message);
			}
		}
		return processedMessages;<commit_msg>Don't retry if the table no longer exists<commit_after>package org.sagebionetworks.table.worker;

import static org.sagebionetworks.repo.model.ACCESS_TYPE.CREATE;
import static org.sagebionetworks.repo.model.ACCESS_TYPE.DELETE;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Callable;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.sagebionetworks.StackConfiguration;
import org.sagebionetworks.asynchronous.workers.sqs.MessageUtils;
import org.sagebionetworks.asynchronous.workers.sqs.Worker;
import org.sagebionetworks.asynchronous.workers.sqs.WorkerProgress;
import org.sagebionetworks.repo.manager.NodeInheritanceManager;
import org.sagebionetworks.repo.manager.table.TableRowManager;
import org.sagebionetworks.repo.manager.trash.EntityInTrashCanException;
import org.sagebionetworks.repo.model.ConflictingUpdateException;
import org.sagebionetworks.repo.model.DatastoreException;
import org.sagebionetworks.repo.model.ObjectType;
import org.sagebionetworks.repo.model.exception.LockUnavilableException;
import org.sagebionetworks.repo.model.jdo.KeyFactory;
import org.sagebionetworks.repo.model.message.ChangeMessage;
import org.sagebionetworks.repo.model.message.ChangeType;
import org.sagebionetworks.repo.model.table.ColumnModel;
import org.sagebionetworks.repo.model.table.RowSet;
import org.sagebionetworks.repo.model.table.TableRowChange;
import org.sagebionetworks.repo.model.table.TableStatus;
import org.sagebionetworks.repo.model.table.TableUnavilableException;
import org.sagebionetworks.repo.web.NotFoundException;
import org.sagebionetworks.table.cluster.ConnectionFactory;
import org.sagebionetworks.table.cluster.TableIndexDAO;
import org.sagebionetworks.table.cluster.utils.TableModelUtils;
import org.springframework.beans.factory.annotation.Autowired;

import com.amazonaws.services.sqs.model.Message;
import com.google.common.collect.SetMultimap;

/**
 * This worker updates the index used to support the tables features. It will
 * listen to table changes and apply them to the RDS table that acts as the
 * index of the table feature.
 * 
 * @author John
 * 
 */
public class TableWorker implements Worker {

	private static final Long TRASH_FOLDER_ID = Long.parseLong(StackConfiguration.getTrashFolderEntityIdStatic());

	enum State {
		SUCCESS, UNRECOVERABLE_FAILURE, RECOVERABLE_FAILURE,
	}

	private static final long BATCH_SIZE = 16000;

	static private Logger log = LogManager.getLogger(TableWorker.class);

	@Autowired
	ConnectionFactory tableConnectionFactory;
	@Autowired
	TableRowManager tableRowManager;
	@Autowired
	StackConfiguration configuration;
	@Autowired
	NodeInheritanceManager nodeInheritanceManager;

	List<Message> messages;
	WorkerProgress workerProgress;

	public TableWorker() {
	}

	// unit testing only
	TableWorker(ConnectionFactory tableConnectionFactory, TableRowManager tableRowManager, StackConfiguration configuration,
			NodeInheritanceManager nodeInheritanceManager) {
		this.tableConnectionFactory = tableConnectionFactory;
		this.tableRowManager = tableRowManager;
		this.configuration = configuration;
		this.nodeInheritanceManager = nodeInheritanceManager;
	}

	@Override
	public void setMessages(List<Message> messages) {
		this.messages = messages;
	}

	@Override
	public void setWorkerProgress(WorkerProgress workerProgress) {
		this.workerProgress = workerProgress;
	}

	@Override
	public List<Message> call() throws Exception {
		List<Message> processedMessages = new LinkedList<Message>();
		// If the feature is disabled then we simply swallow all messages
		if (!configuration.getTableEnabled()) {
			return messages;
		}
		// process each message
		for (Message message : messages) {
			// Extract the ChangeMessage
			ChangeMessage change = MessageUtils.extractMessageBody(message);
			// We only care about entity messages here
			if (ObjectType.TABLE.equals((change.getObjectType()))) {
				if (ChangeType.DELETE.equals(change.getChangeType())) {
					// Delete the table in the index
					TableIndexDAO indexDao = tableConnectionFactory
							.getConnection(change.getObjectId());
					if (indexDao != null) {
						indexDao.deleteTable(change.getObjectId());
						indexDao.deleteStatusTable(change.getObjectId());
					}
					processedMessages.add(message);
				} else {
					// Create or update.
					String tableId = change.getObjectId();
					// make sure the table is not in the trash
					try {
						if (nodeInheritanceManager.isNodeInTrash(tableId)) {
							processedMessages.add(message);
							continue;
						}
					} catch (NotFoundException e) {
						// if the table no longer exists, we want to stop trying
						processedMessages.add(message);
						continue;
					}
					// this method does the real work.
					State state = createOrUpdateTable(tableId,
							change.getObjectEtag(), message);
					if (!State.RECOVERABLE_FAILURE.equals(state)) {
						// Only recoverable failures should remain in the queue.
						// All other must be removed.
						processedMessages.add(message);
					}
				}
			} else {
				// Non-table messages must be returned so they can be removed
				// from the queue.
				processedMessages.add(message);
			}
		}
		return processedMessages;
{'fraction_non_alphanumeric': 0.08831241805581569, 'fraction_numerical': 0.001311106948866829, 'mean_word_length': 4.83551912568306, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, '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': '5399267', 'n_tokens_mistral': 3517, 'n_tokens_neox': 3148, 'n_words': 675}