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

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

@@ -0,0 +1,48 @@
+/*
+ *  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;
+import java.sql.SQLException;
+
+/**
+ * The type RmTableMetaException exception.
+ *
+ * @author Bughue
+ */
+public class TableMetaException extends SQLException {
+    private String columnName;
+    private TableMeta tableMeta;
+    private boolean refreshable;
+
+    public TableMetaException(String columnName, TableMeta tableMeta,boolean refreshable) {
+        this.columnName = columnName;
+        this.tableMeta = tableMeta;
+        this.refreshable = refreshable;
+    }
+
+    public TableMeta getTableMeta() {
+        return tableMeta;
+    }
+
+    public String getColumnName() {
+        return columnName;
+    }
+
+    public boolean isRefreshable() {
+        return refreshable;
+    }
+}

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

@@ -84,7 +84,6 @@ public abstract class BaseTransactionalExecutor<T, S extends Statement> implemen
      */
     protected List<SQLRecognizer> sqlRecognizers;
 
-    private TableMeta tableMeta;
 
     /**
      * Instantiates a new Base transactional executor.
@@ -302,13 +301,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;
     }
 
     /**
@@ -470,7 +465,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);
         }
@@ -511,13 +506,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

@@ -108,7 +108,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/MySQLInsertOnDuplicateUpdateExecutor.java

@@ -289,7 +289,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/exec/mysql/MySQLUpdateJoinExecutor.java

@@ -159,7 +159,7 @@ protected TableRecords afterImage(TableRecords beforeImage) throws SQLException
             try (PreparedStatement pst = statementProxy.getConnection().prepareStatement(selectSQL)) {
                 setAfterImageSQLPlaceHolderParams(joinConditionParams,tableBeforeImage.pkRows(), getTableMeta(tableItems[i]).getPrimaryKeyOnlyName(), pst);
                 rs = pst.executeQuery();
-                TableRecords afterImage = TableRecords.buildRecords(getTableMeta(tableItems[i]), rs);
+                TableRecords afterImage = TableRecords.buildRecords(getTableMeta(tableItems[i]), rs, statementProxy);
                 afterImagesMap.put(tableItems[i], afterImage);
             } finally {
                 IOUtil.close(rs);

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

@@ -34,6 +34,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.TableMetaException;
 import io.seata.rm.datasource.sql.serial.SerialArray;
 import static io.seata.rm.datasource.exec.oracle.OracleJdbcType.TIMESTAMP_WITH_LOCAL_TIME_ZONE;
 import static io.seata.rm.datasource.exec.oracle.OracleJdbcType.TIMESTAMP_WITH_TIME_ZONE;
@@ -186,7 +189,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();
@@ -196,7 +199,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());
@@ -259,6 +262,74 @@ 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) throws SQLException {
+        ColumnMeta col = tmeta.getColumnMeta(colName);
+        if (col == null) {
+            throw new TableMetaException(colName, tmeta, true);
+        }
+        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 (TableMetaException e) {
+            if (statementProxy == null || !e.isRefreshable()) {
+                throw e;
+            }
+            refreshTableMeta(statementProxy.getConnectionProxy(), e.getTableMeta(), e.getColumnName());
+            // try to build again after refresh table meta successfully
+            return buildRecords(getCacheTableMeta(statementProxy.getConnectionProxy(), tmeta.getTableName()), resultSet);
+        }
+    }
+
+
+    private static void refreshTableMeta(ConnectionProxy connectionProxy, TableMeta tmeta, String columnName)
+        throws SQLException {
+        if (columnEmptyAndRefreshable(connectionProxy, tmeta, columnName)) {
+            synchronized (TableRecords.class) {
+                if (columnEmptyAndRefreshable(connectionProxy, tmeta, columnName)) {
+                    TableMetaCacheFactory.getTableMetaCache(connectionProxy.getDbType()).refresh(
+                        connectionProxy.getTargetConnection(), connectionProxy.getDataSourceProxy().getResourceId());
+
+                    // still empty after refreshed
+                    if (getCacheTableMeta(connectionProxy, tmeta.getTableName()).getColumnMeta(columnName) == null) {
+                        throw new TableMetaException(columnName, tmeta, false);
+                    }
+                }
+            }
+        }
+    }
+
+    private static boolean columnEmptyAndRefreshable(ConnectionProxy connectionProxy, TableMeta tmeta,
+        String columnName) {
+        TableMeta cacheTableMeta = getCacheTableMeta(connectionProxy, tmeta.getTableName());
+        return cacheTableMeta.getColumnMeta(columnName) == null;
+    }
+
+
+    private static TableMeta getCacheTableMeta(ConnectionProxy connectionProxy, String tableName) {
+        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/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);
         }

rm-datasource/src/test/java/io/seata/rm/datasource/sql/struct/TableRecordsTest.java

@@ -20,6 +20,7 @@
 import java.sql.Types;
 import java.util.List;
 
+import io.seata.rm.datasource.exception.TableMetaException;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -48,19 +49,36 @@ public class TableRecordsTest {
             new Object[] {"", "", "table_records_test", "description", Types.CLOB, "CLOB", 64, 0, 10, 0, "", "", 0, 0, 64, 2, "YES", "NO"},
         };
 
+    private static Object[][] columnMetasNewField =
+            new Object[][] {
+                    new Object[] {"", "", "table_records_test", "id", Types.INTEGER, "INTEGER", 64, 0, 10, 1, "", "", 0, 0, 64, 1, "NO", "YES"},
+                    new Object[] {"", "", "table_records_test", "name", Types.VARCHAR, "VARCHAR", 64, 0, 10, 0, "", "", 0, 0, 64, 2, "YES", "NO"},
+                    new Object[] {"", "", "table_records_test", "information", Types.BLOB, "BLOB", 64, 0, 10, 0, "", "", 0, 0, 64, 2, "YES", "NO"},
+                    new Object[] {"", "", "table_records_test", "description", Types.CLOB, "CLOB", 64, 0, 10, 0, "", "", 0, 0, 64, 2, "YES", "NO"},
+                    new Object[] {"", "", "table_records_test", "newf", Types.CLOB, "CLOB", 64, 0, 10, 0, "", "", 0, 0, 64, 2, "YES", "NO"},
+            };
+
     private static Object[][] indexMetas =
         new Object[][] {
             new Object[] {"PRIMARY", "id", false, "", 3, 1, "A", 34},
         };
 
     private static List<String> returnValueColumnLabels = Lists.newArrayList("id", "name", "information", "description");
 
+    private static List<String> returnValueColumnLabelsNewField = Lists.newArrayList("id", "name", "information", "description","newf");
+
     private static Object[][] returnValue =
         new Object[][] {
             new Object[] {1, "Tom", "hello", "world"},
             new Object[] {2, "Jack", "hello", "world"},
         };
 
+    private static Object[][] returnValueNewField =
+            new Object[][] {
+                    new Object[] {1, "Tom", "hello", "world","newf"},
+                    new Object[] {2, "Jack", "hello", "world","newf"},
+            };
+
     @BeforeEach
     public void initBeforeEach() {
     }
@@ -91,7 +109,7 @@ public void testPkRow() throws SQLException {
 
         ResultSet resultSet = mockDriver.executeQuery(mockStatement, "select * from table_records_test");
 
-        TableRecords tableRecords = TableRecords.buildRecords(tableMeta, resultSet);
+        TableRecords tableRecords = TableRecords.buildRecords(tableMeta, resultSet, null);
 
         Assertions.assertEquals(returnValue.length, tableRecords.pkRows().size());
     }
@@ -110,11 +128,30 @@ public void testBuildRecords() throws SQLException {
 
         ResultSet resultSet = mockDriver.executeQuery(mockStatement, "select * from table_records_test");
 
-        TableRecords tableRecords = TableRecords.buildRecords(tableMeta, resultSet);
+        TableRecords tableRecords = TableRecords.buildRecords(tableMeta, resultSet, null);
 
         Assertions.assertNotNull(tableRecords);
     }
 
+    @Test
+    public void testBuildRecordsNewFeild() throws SQLException {
+        MockDriver mockDriver = new MockDriver(returnValueColumnLabels, returnValue, columnMetas, indexMetas);
+        DruidDataSource dataSource = new DruidDataSource();
+        dataSource.setUrl("jdbc:mock:xxx");
+        dataSource.setDriver(mockDriver);
+        MockStatementBase mockStatement = new MockStatement(dataSource.getConnection().getConnection());
+        DataSourceProxy proxy = new DataSourceProxy(dataSource);
+
+        TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache(JdbcConstants.MYSQL).getTableMeta(proxy.getPlainConnection(),
+                "table_records_test", proxy.getResourceId());
+
+
+        //  模拟新字段增加
+        MockDriver mockDriverNewField = new MockDriver(returnValueColumnLabelsNewField, returnValueNewField, columnMetasNewField, indexMetas);
+        ResultSet resultSet = mockDriverNewField.executeQuery(mockStatement, "select * from table_records_test");
+        Assertions.assertThrows(TableMetaException.class, () -> TableRecords.buildRecords(tableMeta, resultSet, null));
+    }
+
     @Test
     public void testEmpty() {
         TableRecords.EmptyTableRecords emptyTableRecords = new TableRecords.EmptyTableRecords();

rm-datasource/src/test/java/io/seata/rm/datasource/undo/BaseH2Test.java

@@ -98,7 +98,7 @@ protected static TableRecords execQuery(TableMeta tableMeta, String sql) throws
         try {
             s = connection.createStatement();
             set = s.executeQuery(sql);
-            return TableRecords.buildRecords(tableMeta, set);
+            return TableRecords.buildRecords(tableMeta, set, null);
         } finally {
             IOUtil.close(set, s);
         }

test/src/test/java/io/seata/at/ATModeSupportDataBaseDataTypeTest.java

@@ -132,7 +132,7 @@ private void doTestOracleTypePhase(int sqlType, boolean globalCommit, String tab
         LOGGER.info("the helperRes is:[{}]", helperRes);
         TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache(io.seata.sqlparser.util.JdbcConstants.ORACLE)
             .getTableMeta(dataSourceProxy.getPlainConnection(), tableName, dataSourceProxy.getResourceId());
-        TableRecords beforeImage = TableRecords.buildRecords(tableMeta, helperRes);
+        TableRecords beforeImage = TableRecords.buildRecords(tableMeta, helperRes, null);
 
         // if not throw exception update record
         Assertions.assertDoesNotThrow(() -> testStat.execute(updateSql));
@@ -152,7 +152,7 @@ private void doTestOracleTypePhase(int sqlType, boolean globalCommit, String tab
             helperConn = helperDS.getConnection();
             helperStat = helperConn.createStatement();
             helperRes = helperStat.executeQuery("select * from " + tableName + " where id = " + TEST_RECORD_ID);
-            TableRecords currentImage = TableRecords.buildRecords(tableMeta, helperRes);
+            TableRecords currentImage = TableRecords.buildRecords(tableMeta, helperRes, null);
             LOGGER.info("the currentImage Rows is:[{}]", currentImage.getRows());
             Assertions.assertTrue(DataCompareUtils.isRecordsEquals(beforeImage, currentImage).getResult());
             helperRes.close();

test/src/test/java/io/seata/at/mysql/MysqlUpdateJoinTest.java

@@ -98,11 +98,11 @@ private void doTestPhase2(boolean globalCommit, String updateSql) throws Throwab
         table1HelperRes = helperStat.executeQuery("select * from t where id = " + testRecordId );
         TableMeta table1Meta = TableMetaCacheFactory.getTableMetaCache(JdbcConstants.MYSQL).getTableMeta(dataSourceProxy.getPlainConnection(),
                 "t", dataSourceProxy.getResourceId());
-        TableRecords table1BeforeImage = TableRecords.buildRecords(table1Meta, table1HelperRes);
+        TableRecords table1BeforeImage = TableRecords.buildRecords(table1Meta, table1HelperRes, null);
         table2HelperRes = helperStat.executeQuery("select * from t1 where id = " + testRecordId1);
         TableMeta table2Meta = TableMetaCacheFactory.getTableMetaCache(JdbcConstants.MYSQL).getTableMeta(dataSourceProxy.getPlainConnection(),
                 "t1", dataSourceProxy.getResourceId());
-        TableRecords table2BeforeImage = TableRecords.buildRecords(table2Meta, table2HelperRes);
+        TableRecords table2BeforeImage = TableRecords.buildRecords(table2Meta, table2HelperRes, null);
         // >>> update record should not throw exception
         Assertions.assertDoesNotThrow(() -> testStat.execute(updateSql));
         // >>> close the statement and connection
@@ -120,9 +120,9 @@ private void doTestPhase2(boolean globalCommit, String updateSql) throws Throwab
             helperConn = helperDS.getConnection();
             helperStat = helperConn.createStatement();
             table1HelperRes = helperStat.executeQuery("select * from t where id = " + testRecordId);
-            TableRecords table1CurrentImage = TableRecords.buildRecords(table1Meta, table1HelperRes);
+            TableRecords table1CurrentImage = TableRecords.buildRecords(table1Meta, table1HelperRes, null);
             table2HelperRes = helperStat.executeQuery("select * from t1 where id = " + testRecordId1);
-            TableRecords table2CurrentImage = TableRecords.buildRecords(table2Meta, table2HelperRes);
+            TableRecords table2CurrentImage = TableRecords.buildRecords(table2Meta, table2HelperRes, null);
             Assertions.assertTrue(DataCompareUtils.isRecordsEquals(table1BeforeImage, table1CurrentImage).getResult());
             Assertions.assertTrue(DataCompareUtils.isRecordsEquals(table2BeforeImage, table2CurrentImage).getResult());
             table1HelperRes.close();