bugfix : check if table meta cache should be refreshed in AT mode

rm-datasource/src/main/java/io/seata/rm/datasource/DataSourceProxy.java

@@ -115,6 +115,7 @@ private void init(DataSource dataSource, String resourceGroupId) {
         initResourceId();
         DefaultResourceManager.get().registerResource(this);
         if (ENABLE_TABLE_META_CHECKER_ENABLE) {
+            //todo todo_4572
             tableMetaExecutor.scheduleAtFixedRate(() -> {
                 try (Connection connection = dataSource.getConnection()) {
                     TableMetaCacheFactory.getTableMetaCache(DataSourceProxy.this.getDbType())

rm-datasource/src/main/java/io/seata/rm/datasource/exception/RmTableMetaException.java

@@ -0,0 +1,41 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package io.seata.rm.datasource.exception;
+
+import io.seata.rm.datasource.sql.struct.TableMeta;
+
+/**
+ * The type RmTableMetaException exception.
+ *
+ * @author Bughue
+ */
+public class RmTableMetaException extends RuntimeException {
+    private String columnName;
+    private TableMeta tableMeta;
+
+    public RmTableMetaException(String columnName, TableMeta tableMeta) {
+        this.columnName = columnName;
+        this.tableMeta = tableMeta;
+    }
+
+    public TableMeta getTableMeta() {
+        return tableMeta;
+    }
+
+    public String getColumnName() {
+        return columnName;
+    }
+}

rm-datasource/src/main/java/io/seata/rm/datasource/exec/BaseTransactionalExecutor.java

@@ -85,7 +85,6 @@ public abstract class BaseTransactionalExecutor<T, S extends Statement> implemen
      */
     protected List<SQLRecognizer> sqlRecognizers;
 
-    private TableMeta tableMeta;
 
     /**
      * Instantiates a new Base transactional executor.
@@ -252,13 +251,9 @@ protected TableMeta getTableMeta() {
      * @return the table meta
      */
     protected TableMeta getTableMeta(String tableName) {
-        if (tableMeta != null) {
-            return tableMeta;
-        }
         ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
-        tableMeta = TableMetaCacheFactory.getTableMetaCache(connectionProxy.getDbType())
+        return TableMetaCacheFactory.getTableMetaCache(connectionProxy.getDbType())
             .getTableMeta(connectionProxy.getTargetConnection(), tableName, connectionProxy.getDataSourceProxy().getResourceId());
-        return tableMeta;
     }
 
     /**
@@ -406,7 +401,7 @@ protected TableRecords buildTableRecords(TableMeta tableMeta, String selectSQL,
                 }
             }
             rs = ps.executeQuery();
-            return TableRecords.buildRecords(tableMeta, rs);
+            return TableRecords.buildRecords(tableMeta, rs, statementProxy);
         } finally {
             IOUtil.close(rs);
         }
@@ -447,13 +442,13 @@ protected TableRecords buildTableRecords(Map<String, List<Object>> pkValuesMap)
             for (int r = 0; r < rowSize; r++) {
                 for (int c = 0; c < pkColumnNameList.size(); c++) {
                     List<Object> pkColumnValueList = pkValuesMap.get(pkColumnNameList.get(c));
-                    int dataType = tableMeta.getColumnMeta(pkColumnNameList.get(c)).getDataType();
+                    int dataType = getTableMeta().getColumnMeta(pkColumnNameList.get(c)).getDataType();
                     ps.setObject(paramIndex, pkColumnValueList.get(r), dataType);
                     paramIndex++;
                 }
             }
             rs = ps.executeQuery();
-            return TableRecords.buildRecords(getTableMeta(), rs);
+            return TableRecords.buildRecords(getTableMeta(), rs, statementProxy);
         } finally {
             IOUtil.close(rs);
         }

rm-datasource/src/main/java/io/seata/rm/datasource/exec/MultiUpdateExecutor.java

@@ -143,7 +143,7 @@ protected TableRecords afterImage(TableRecords beforeImage) throws SQLException
         try (PreparedStatement pst = statementProxy.getConnection().prepareStatement(selectSQL);) {
             SqlGenerateUtils.setParamForPk(beforeImage.pkRows(), getTableMeta().getPrimaryKeyOnlyName(), pst);
             rs = pst.executeQuery();
-            return TableRecords.buildRecords(tmeta, rs);
+            return TableRecords.buildRecords(tmeta, rs, statementProxy);
         } finally {
             IOUtil.close(rs);
         }

rm-datasource/src/main/java/io/seata/rm/datasource/exec/UpdateExecutor.java

@@ -123,7 +123,7 @@ protected TableRecords afterImage(TableRecords beforeImage) throws SQLException
         try (PreparedStatement pst = statementProxy.getConnection().prepareStatement(selectSQL)) {
             SqlGenerateUtils.setParamForPk(beforeImage.pkRows(), getTableMeta().getPrimaryKeyOnlyName(), pst);
             rs = pst.executeQuery();
-            return TableRecords.buildRecords(tmeta, rs);
+            return TableRecords.buildRecords(tmeta, rs, statementProxy);
         } finally {
             IOUtil.close(rs);
         }

rm-datasource/src/main/java/io/seata/rm/datasource/exec/mysql/MySQLInsertOrUpdateExecutor.java

@@ -269,7 +269,7 @@ public TableRecords buildTableRecords2(TableMeta tableMeta, String selectSQL, Ar
             }
 
             rs = ps.executeQuery();
-            return TableRecords.buildRecords(tableMeta, rs);
+            return TableRecords.buildRecords(tableMeta, rs, statementProxy);
         } finally {
             IOUtil.close(rs);
         }

rm-datasource/src/main/java/io/seata/rm/datasource/sql/struct/TableMeta.java

@@ -17,10 +17,12 @@
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import io.seata.common.exception.NotSupportYetException;
@@ -41,6 +43,11 @@ public class TableMeta {
      */
     private final Map<String, ColumnMeta> allColumns = new LowerCaseLinkHashMap<>();
 
+    /**
+     * unrefreshable column name
+     */
+    private Set<String> unrefreshableColumns = new HashSet<>();
+
     /**
      * key: index name
      */
@@ -92,6 +99,25 @@ public Map<String, IndexMeta> getAllIndexes() {
         return allIndexes;
     }
 
+    /**
+     * Gets unrefreshable columns.
+     *
+     * @return the all indexes
+     */
+    public Set<String> getUnrefreshableColumns() {
+        return unrefreshableColumns;
+    }
+
+    /**
+     * Add unrefreshable columns.
+     *
+     * @return the all indexes
+     */
+    public void addUnrefreshableColumn(String column) {
+        this.unrefreshableColumns.add(column);
+    }
+
+
     /**
      * Gets auto increase column.
      *

rm-datasource/src/main/java/io/seata/rm/datasource/sql/struct/TableMetaCache.java

@@ -34,6 +34,16 @@ public interface TableMetaCache {
      */
     TableMeta getTableMeta(Connection connection, String tableName, String resourceId);
 
+    /**
+     * add unrefreshable column
+     *
+     * @param connection the connection
+     * @param tableName  the table name
+     * @param resourceId the resource id
+     * @param colName
+     */
+    void addUnrefreshableCol(Connection connection, String tableName, String resourceId, String colName);
+
     /**
      * Clear the table meta cache
      *

rm-datasource/src/main/java/io/seata/rm/datasource/sql/struct/TableRecords.java

@@ -18,6 +18,7 @@
 import java.sql.Array;
 import java.sql.Blob;
 import java.sql.Clob;
+import java.sql.Connection;
 import java.sql.NClob;
 import java.sql.Ref;
 import java.sql.ResultSet;
@@ -34,6 +35,9 @@
 import javax.sql.rowset.serial.SerialJavaObject;
 import javax.sql.rowset.serial.SerialRef;
 import io.seata.common.exception.ShouldNeverHappenException;
+import io.seata.rm.datasource.ConnectionProxy;
+import io.seata.rm.datasource.StatementProxy;
+import io.seata.rm.datasource.exception.RmTableMetaException;
 import io.seata.rm.datasource.sql.serial.SerialArray;
 
 /**
@@ -182,7 +186,7 @@ public static TableRecords empty(TableMeta tableMeta) {
      * @return the table records
      * @throws SQLException the sql exception
      */
-    public static TableRecords buildRecords(TableMeta tmeta, ResultSet resultSet) throws SQLException {
+    private static TableRecords buildRecords(TableMeta tmeta, ResultSet resultSet) throws SQLException {
         TableRecords records = new TableRecords(tmeta);
         ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
         Map<String, ColumnMeta> primaryKeyMap = tmeta.getPrimaryKeyMap();
@@ -192,7 +196,7 @@ public static TableRecords buildRecords(TableMeta tmeta, ResultSet resultSet) th
             List<Field> fields = new ArrayList<>(columnCount);
             for (int i = 1; i <= columnCount; i++) {
                 String colName = resultSetMetaData.getColumnName(i);
-                ColumnMeta col = tmeta.getColumnMeta(colName);
+                ColumnMeta col = checkAndGetColumnMeta(tmeta,colName);
                 int dataType = col.getDataType();
                 Field field = new Field();
                 field.setName(col.getColumnName());
@@ -253,6 +257,82 @@ public static TableRecords buildRecords(TableMeta tmeta, ResultSet resultSet) th
         return records;
     }
 
+    /**
+     * check if the column is null and return
+     * @param tmeta the table meta
+     * @param colName the column nmae
+     */
+    private static ColumnMeta checkAndGetColumnMeta(TableMeta tmeta , String colName) {
+        ColumnMeta col = tmeta.getColumnMeta(colName);
+        if (col == null) {
+            throw new RmTableMetaException(colName,tmeta);
+        }
+        return col;
+    }
+
+
+    /**
+     * Build records table records.
+     *
+     * @param tmeta     the tmeta
+     * @param resultSet the result set
+     * @param statementProxy the statement proxy
+     * @return the table records
+     * @throws SQLException the sql exception
+     */
+    public static TableRecords buildRecords(TableMeta tmeta, ResultSet resultSet, StatementProxy statementProxy)
+        throws SQLException {
+        try {
+            return buildRecords(tmeta, resultSet);
+        } catch (RmTableMetaException e) {
+            if (statementProxy == null) {
+                throw e;
+            }
+            refreshTableMeta(statementProxy, e.getTableMeta(), e.getColumnName());
+            // try to build again after refresh table meta success
+            return buildRecords(getCacheTableMeta(statementProxy, tmeta.getTableName()), resultSet);
+        }
+    }
+
+
+    private static void refreshTableMeta(StatementProxy statementProxy, TableMeta tmeta, String columnName)
+        throws SQLException {
+        if (columnEmptyAndRefreshable(statementProxy, tmeta, columnName)) {
+            synchronized (TableRecords.class) {
+                if (columnEmptyAndRefreshable(statementProxy, tmeta, columnName)) {
+                    ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
+                    try (Connection connection = statementProxy.getConnection()) {
+                        TableMetaCacheFactory.getTableMetaCache(connectionProxy.getDbType()).refresh(connection,
+                            connectionProxy.getDataSourceProxy().getResourceId());
+                    } catch (Exception exp) {
+                        throw exp;
+                    }
+                    // still empty after refreshed
+                    if (getCacheTableMeta(statementProxy, tmeta.getTableName()).getColumnMeta(columnName) == null) {
+                        TableMetaCacheFactory.getTableMetaCache(connectionProxy.getDbType()).addUnrefreshableCol(
+                            connectionProxy.getTargetConnection(), tmeta.getTableName(),
+                            connectionProxy.getDataSourceProxy().getResourceId(), columnName);
+                    }
+                }
+            }
+        }
+    }
+
+    private static boolean columnEmptyAndRefreshable(StatementProxy statementProxy, TableMeta tmeta,
+        String columnName) {
+        TableMeta cacheTableMeta = getCacheTableMeta(statementProxy, tmeta.getTableName());
+        return cacheTableMeta.getColumnMeta(columnName) == null
+            && !cacheTableMeta.getUnrefreshableColumns().contains(columnName);
+    }
+
+
+    private static TableMeta getCacheTableMeta(StatementProxy statementProxy, String tableName) {
+        ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
+        TableMeta tmeta = TableMetaCacheFactory.getTableMetaCache(connectionProxy.getDbType()).getTableMeta(
+            connectionProxy.getTargetConnection(), tableName, connectionProxy.getDataSourceProxy().getResourceId());
+        return tmeta;
+    }
+
     /**
      * since there is no parameterless constructor for Blob, Clob and NClob just like mysql,
      * it needs to be converted to Serial_ type

rm-datasource/src/main/java/io/seata/rm/datasource/sql/struct/cache/AbstractTableMetaCache.java

@@ -72,6 +72,19 @@ public TableMeta getTableMeta(final Connection connection, final String tableNam
         return tmeta;
     }
 
+    @Override
+    public void addUnrefreshableCol(final Connection connection, String tableName, String resourceId, String colName) {
+        if (StringUtils.isNullOrEmpty(tableName)) {
+            throw new IllegalArgumentException("TableMeta cannot be fetched without tableName");
+        }
+
+        TableMeta tmeta;
+        final String key = getCacheKey(connection, tableName, resourceId);
+        tmeta = TABLE_META_CACHE.getIfPresent(key);
+        tmeta.addUnrefreshableColumn(colName);
+        TABLE_META_CACHE.put(key, tmeta);
+    }
+
     @Override
     public void refresh(final Connection connection, String resourceId) {
         ConcurrentMap<String, TableMeta> tableMetaMap = TABLE_META_CACHE.asMap();

rm-datasource/src/main/java/io/seata/rm/datasource/undo/AbstractUndoExecutor.java

@@ -321,7 +321,7 @@ protected TableRecords queryCurrentRecords(Connection conn) throws SQLException
             }
 
             checkSet = statement.executeQuery();
-            currentRecords = TableRecords.buildRecords(tableMeta, checkSet);
+            currentRecords = TableRecords.buildRecords(tableMeta, checkSet, null);
         } finally {
             IOUtil.close(checkSet, statement);
         }