Не обновлять строку в ResultSet, если данные были изменены

мы извлекаем данные из разных типов баз данных (Oracle, MySQL, SQL-Server,...). Как только он будет успешно записан в файл, мы хотим пометить его как переданный, поэтому мы обновляем конкретный столбец.

Наша проблема заключается в том, что пользователь имеет возможность изменять данные за это время, но может забыть совершить. Запись блокируется с помощью инструкции select for update. Так может случиться, что мы отмечаем что-то переданное, а это не так.

Это выдержка из нашего кода:

Statement stmt = conn.createStatement(ResultSet.TYPE_SCROLL_SENSITIVE, ResultSet.CONCUR_UPDATABLE);
ResultSet extractedData = stmt.executeQuery(sql);
writeDataToFile(extractedData);
extractedData.beforeFirst();
while (extractedData.next()) {
    if (!extractedData.rowUpdated()) {
        extractedData.updateString("COLUMNNAME", "TRANSMITTED");
        // code will stop here if user has changed data but did not commit
        extractedData.updateRow();
        // once committed the changed data is marked as transmitted
    }
}

Метод extractedData.rowUpdated() возвращает false, потому что технически пользователь еще ничего не изменил. Есть ли способ не обновить строку и определить, были ли данные изменены на этом позднем этапе?

К сожалению, я не могу изменить программу, которую пользователь использует для изменения данных.

Ответ 1

Итак, вы хотите

Запуск всех строк таблицы, которые не были экспортированы
Экспортировать эти данные где-нибудь
Отметьте эти строки экспортированными, чтобы ваша следующая итерация не экспортировала их снова
Поскольку могут быть ожидающие изменения в строке, вы не хотите связываться с этой информацией.

Как насчет:

You iterate over all rows. 

for every row 
   generate a hash value for the contents of the row
   compare column "UPDATE_STATUS" with calulated hash
   if no match
     export row
     store hash into "UPDATE_STATUS" 
      if store fails (row locked) 
         -> no worries, will be exported again next time
      if store succeeds (on data already changed by user) 
         -> no worries, will be exported again as hash will not match

Это может еще больше замедлить ваш экспорт, поскольку вам придется перебирать все, а не обо всем WHERE UPDATE_STATUS IS NULL, но вы могли бы выполнить два задания - один (быстрый) итерация над WHERE UPDATE_STATUS IS NULL и одна медленная и тщательная WHERE UPDATE_STATUS IS NOT NULL (с хеш-перепроверкой на месте)

Если вы хотите избежать сбоев/ожиданий хранилища, вы можете захотеть сохранить хэш/обновленную информацию во вторую таблицу, скопировав первичный ключ плюс значение поля хэша - таким образом пользователь блокировки на главной таблице не будут мешать вашим обновлениям (так как они будут на другой таблице)

Ответ 2

"пользователь [...] может забыть совершить" > Пользователь либо совершает, либо не делает этого. "Забыть" для фиксации равносильно ошибке в его программном обеспечении.

Чтобы обойти это, вам нужно либо:

Запустите транзакцию с уровнем изоляции SERIALIZABLE и внутри этой транзакции:
- Прочитайте данные и экспортируйте их. Данные, прочитанные таким образом, блокируются от обновления.
- Обновление данных, которые вы обработали. Примечание: не делайте этого с обновляемым ResultSet, выполняйте это с помощью оператора UPDATE. Таким образом вам не нужен CONCUR_UPDATABLE + TYPE_SCROLL_SENSITIVE, который намного медленнее, чем CONCUR_READ_ONLY + TYPE_FORWARD_ONLY.
Зафиксировать транзакцию.

Таким образом, багги-программное обеспечение будет заблокировано от обновления данных, которые вы обрабатываете.

Другой способ

Запустите TRANSACTION на более низком уровне изоляции (по умолчанию READ COMMITTED) и внутри этой транзакции
- Выберите данные с соответствующими подсказками таблиц. Например, для SQL Server эти: TABLOCKX + HOLDLOCK (большие наборы данных) или ROWLOCK + XLOCK + HOLDLOCK (небольшие наборы данных), или PAGLOCK + XLOCK + HOLDLOCK. Наличие HOLDLOCK в качестве подсказки таблицы практически эквивалентно транзакции SERIALIZABLE. Обратите внимание, что эскалация блокировки может перевести последние два в блокировки таблиц, если количество блокировок становится слишком высоким.
- Обновление данных, которые вы обработали; Примечание. Используйте оператор UPDATE. Потеряте обновляемый /scroll _sensitive набор результатов.
Зафиксируйте транзакцию.

То же самое дело, багги-программное обеспечение будет заблокировано от обновления данных, которые вы обрабатываете.

Ответ 3

В итоге нам пришлось реализовать оптимистичную блокировку. В некоторых таблицах у нас уже есть столбец, в котором хранится номер версии. В некоторых других таблицах есть столбец временной метки, который содержит время последнего изменения (с помощью триггера).

Хотя временная метка не всегда может быть надежным источником оптимистической блокировки, мы все равно пошли с ней. Несколько изменений за одну секунду не очень реалистичны в нашей среде.

Поскольку мы должны знать первичный ключ, не описывая его перед рукой, нам приходилось обращаться к метаданным результатов. Некоторые из наших баз данных не поддерживают это (например, старые таблицы DB/2). Мы все еще используем для них старую систему.

Примечание. tableMetaData - это файл XML-config, в котором хранится наше описание таблицы. Это напрямую не связано с метаданными таблицы в базе данных.

Statement stmt = conn.createStatement(ResultSet.TYPE_SCROLL_SENSITIVE, ResultSet.CONCUR_UPDATABLE);
ResultSet extractedData = stmt.executeQuery(sql);
writeDataToFile(extractedData);
extractedData.beforeFirst();
while (extractedData.next()) {
    if (tableMetaData.getVersion() != null) {
        markDataAsExported(extractedData, tableMetaData);
    } else {
        markResultSetAsExported(extractedData, tableMetaData);
    }
}

// new way with building of an update statement including the version column in the where clause
private void markDataAsExported(ResultSet extractedData, TableMetaData tableMetaData) throws SQLException {
    ResultSet resultSetPrimaryKeys = null;
    PreparedStatement versionedUpdateStatement = null;
    try {
        ResultSetMetaData extractedMetaData = extractedData.getMetaData();
        resultSetPrimaryKeys = conn.getMetaData().getPrimaryKeys(null, null, tableMetaData.getTable());
        ArrayList<String> primaryKeyList = new ArrayList<String>();
        String sqlStatement = "update " + tableMetaData.getTable() + " set " + tableMetaData.getUpdateColumn()
                + " = ? where ";
        if (resultSetPrimaryKeys.isBeforeFirst()) {
            while (resultSetPrimaryKeys.next()) {
                primaryKeyList.add(resultSetPrimaryKeys.getString(4));
                sqlStatement += resultSetPrimaryKeys.getString(4) + " = ? and ";
            }
            sqlStatement += tableMetaData.getVersionColumn() + " = ?";
            versionedUpdateStatement = conn.prepareStatement(sqlStatement);
            while (extractedData.next()) {
                versionedUpdateStatement.setString(1, tableMetaData.getUpdateValue());
                for (int i = 0; i < primaryKeyList.size(); i++) {
                    versionedUpdateStatement.setObject(i + 2, extractedData.getObject(primaryKeyList.get(i)),
                            extractedMetaData.getColumnType(extractedData.findColumn(primaryKeyList.get(i))));
                }
                versionedUpdateStatement.setObject(primaryKeyList.size() + 2,
                        extractedData.getObject(tableMetaData.getVersionColumn()), tableMetaData.getVersionType());
                if (versionedUpdateStatement.executeUpdate() == 0) {
                    logger.warn(Message.COLLECTOR_DATA_CHANGED, tableMetaData.getTable());
                }
            }
        } else {
            logger.warn(Message.COLLECTOR_PK_ERROR, tableMetaData.getTable());
            markResultSetAsExported(extractedData, tableMetaData);
        }
    } finally {
        if (resultSetPrimaryKeys != null) {
            resultSetPrimaryKeys.close();
        }
        if (versionedUpdateStatement != null) {
            versionedUpdateStatement.close();
        }
    }
}

//the old way as fallback
private void markResultSetAsExported(ResultSet extractedData, TableMetaData tableMetaData) throws SQLException {
    while (extractedData.next()) {
        extractedData.updateString(tableMetaData.getUpdateColumn(), tableMetaData.getUpdateValue());
        extractedData.updateRow();
    }
}