Merge branch 'master' into secondary_part_rebase_master2

Author: maochongxin

README.md

@@ -18,7 +18,7 @@ Import the dependency for your maven project:
 <dependency>
     <groupId>com.oceanbase</groupId>
     <artifactId>obkv-table-client</artifactId>
-    <version>1.4.0</version>
+    <version>1.4.1</version>
 </dependency>
 ```
 

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>com.oceanbase</groupId>
     <artifactId>obkv-table-client</artifactId>
-    <version>1.4.1-SNAPSHOT</version>
+    <version>1.4.3-SNAPSHOT</version>
 
     <name>${project.groupId}:${project.artifactId}</name>
     <description>OceanBase JavaClient for TableApi</description>

src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java

@@ -18,6 +18,8 @@
 package com.alipay.oceanbase.rpc.exception;
 
 public class ObTableEntryRefreshException extends ObTableException {
+    
+    private boolean connectInactive = false;
 
     /*
      * Ob table entry refresh exception.
@@ -52,7 +54,15 @@ public ObTableEntryRefreshException(String message) {
     public ObTableEntryRefreshException(String message, Throwable cause) {
         super(message, cause);
     }
+    
+    public ObTableEntryRefreshException(String message, Throwable cause, boolean connectInactive) {
+        super(message, cause);
+        this.connectInactive = connectInactive;
+    }
 
+    public boolean isConnectInactive() {
+        return connectInactive;
+    }
     /*
      * Ob table entry refresh exception.
      */

src/main/java/com/alipay/oceanbase/rpc/exception/ObTableEntryRefreshException.java

@@ -60,6 +60,7 @@ public class LocationUtil {
                                                                                         .getLogger(LocationUtil.class);
     static {
         ParserConfig.getGlobalInstance().setSafeMode(true);
+        loadJdbcDriver();
     }
 
     private static final String OB_VERSION_SQL                                      = "SELECT /*+READ_CONSISTENCY(WEAK)*/ OB_VERSION() AS CLUSTER_VERSION;";
@@ -325,22 +326,24 @@ private static String formatObServerUrl(ObServerAddr obServerAddr, long connectT
      */
     private static Connection getMetaRefreshConnection(String url, ObUserAuth sysUA)
                                                                                     throws ObTableEntryRefreshException {
-        loadJdbcDriver();
 
         try {
             return DriverManager.getConnection(url, sysUA.getUserName(), sysUA.getPassword());
         } catch (Exception e) {
             RUNTIME.error(LCD.convert("01-00005"), e.getMessage(), e);
-            throw new ObTableEntryRefreshException("fail to connect meta server", e);
+            // Since the JDBC connection fails here, it is likely that the server has crashed or scaling down.   
+            // Therefore, we need to set the Inactive flag of the ObTableEntryRefreshException to true.   
+            // This allows the upper-layer retry mechanism to catch this exception and immediately refresh the metadata.
+            throw new ObTableEntryRefreshException("fail to connect meta server", e, true /* connect inactive */);
         }
     }
 
     private static void loadJdbcDriver() {
         try {
             Class.forName("com.mysql.cj.jdbc.Driver");
             return;
-        } catch (ClassNotFoundException e) {
-            RUNTIME.info("Class 'com.mysql.cj.jdbc.Driver' not found, "
+        } catch (ClassNotFoundException ignored) {
+            RUNTIME.debug("Class 'com.mysql.cj.jdbc.Driver' not found, "
                          + "try to load legacy driver class 'com.mysql.jdbc.Driver'");
         }
 
@@ -430,9 +433,15 @@ private static TableEntry callTableEntryRefresh(ObServerAddr obServerAddr, Table
             } else {
                 RUNTIME.error(LCD.convert("01-00007"), url, key, e);
             }
-            throw new ObTableEntryRefreshException(format(
-                "fail to refresh table entry from remote url=%s, key=%s, message=%s", url, key,
-                e.getMessage()), e);
+            if (e instanceof ObTableEntryRefreshException) {
+                throw new ObTableEntryRefreshException(format(
+                        "fail to refresh table entry from remote url=%s, key=%s, message=%s", url, key,
+                        e.getMessage()), e, ((ObTableEntryRefreshException) e).isConnectInactive());
+            } else {
+                throw new ObTableEntryRefreshException(format(
+                        "fail to refresh table entry from remote url=%s, key=%s, message=%s", url, key,
+                        e.getMessage()), e.getCause());
+            }
         } finally {
             try {
                 if (null != connection) {
@@ -571,6 +580,10 @@ public static String loadTableNameWithGroupName(final ServerRoster serverRoster,
             RUNTIME.error("callTableEntryNameWithPriority meet exception", e);
             serverRoster.downgradePriority(addr);
             throw e;
+        } catch (ObTableEntryRefreshException e) {
+            RUNTIME.error("callTableEntryNameWithPriority meet exception", e);
+            throw new ObTableEntryRefreshException(format(
+                "fail to get table name from remote url=%s, key=%s", url, key), e, e.isConnectInactive());
         } catch (Exception e) {
             throw new ObTableNotExistException(format(
                 "fail to get table name from remote url=%s, key=%s", url, key), e);
@@ -797,12 +810,19 @@ private static TableEntry getTableEntryFromRemote(Connection connection, TableEn
                     }
                 }
             }
+        } catch (SQLException e) {
+            // cannot execute sql, maybe some of the observers have been killed
+            RUNTIME.error(LCD.convert("01-00010"), key, e.getMessage());
+            throw new ObTableEntryRefreshException("fail to get partition location entry from remote", e, true);
         } catch (ObTableNotExistException e) {
             // avoid to refresh meta for ObTableNotExistException
             RUNTIME.error("getTableEntryFromRemote meet exception", e);
             throw e;
         } catch (Exception e) {
             RUNTIME.error(LCD.convert("01-00009"), key, e);
+            if (e instanceof ObTableEntryRefreshException) {
+                throw e;
+            }
             throw new ObTableEntryRefreshException(format(
                 "fail to get table entry from remote, key=%s", key), e);
         } finally {
@@ -924,6 +944,10 @@ public static TableEntry getTableEntryLocationFromRemote(Connection connection,
             rs = ps.executeQuery();
             getPartitionLocationFromResultSetByTablet(tableEntry, rs, partitionEntry, tabletId,
                 withLsId);
+        } catch (SQLException e) {
+            // cannot execute sql, maybe some of the observers have been killed
+            RUNTIME.error(LCD.convert("01-00010"), key, tableEntry, e.getMessage());
+            throw new ObTableEntryRefreshException("fail to get partition location entry from remote", e, true);
         } catch (Exception e) {
             RUNTIME.error(LCD.convert("01-00010"), key, tableEntry, e);
             throw new ObTablePartitionLocationRefreshException(format(
@@ -973,6 +997,9 @@ public static TableEntry getTableEntryLocationFromRemote(Connection connection,
                 }
                 rs = ps.executeQuery();
                 partitionEntry = getPartitionLocationFromResultSet(tableEntry, rs, partitionEntry);
+            } catch (SQLException e) {
+                RUNTIME.error(LCD.convert("01-00010"), key, partitionNum, tableEntry, e);
+                throw new ObTableEntryRefreshException("fail to get partition location entry from remote", e, true);
             } catch (Exception e) {
                 RUNTIME.error(LCD.convert("01-00010"), key, partitionNum, tableEntry, e);
                 throw new ObTablePartitionLocationRefreshException(format(
@@ -1019,8 +1046,13 @@ public static Long getTableIdFromRemote(ObServerAddr obServerAddr, ObUserAuth sy
                                                        + " table_id from remote");
             }
         } catch (Exception e) {
-            throw new ObTableEntryRefreshException("fail to get " + tableName
-                                                   + " table_id from remote", e);
+            if (e instanceof ObTableEntryRefreshException) {
+                throw new ObTableEntryRefreshException(format(
+                    "fail to get " + tableName + " table_id from remote", e), e, ((ObTableEntryRefreshException) e).isConnectInactive());
+            } else {
+                throw new ObTableEntryRefreshException(format(
+                    "fail to get " + tableName + " table_id from remote", e), e);
+            }
         } finally {
             try {
                 if (null != rs) {
@@ -1058,10 +1090,18 @@ public static ObIndexInfo getIndexInfoFromRemote(ObServerAddr obServerAddr, ObUs
             } else {
                 throw new ObTableEntryRefreshException("index is not exist");
             }
+        } catch (SQLException e) {
+            // cannot execute sql, maybe some of the observers have been killed
+            RUNTIME.error(LCD.convert("01-00010"), indexTableName, e.getMessage());
+            throw new ObTableEntryRefreshException("fail to get index info from remote", e, true);
         } catch (Exception e) {
-            throw new ObTableEntryRefreshException(format(
-                "fail to get index info from remote, indexTableName: %s, error message: %s",
-                indexTableName, e.getMessage()), e);
+            if (e instanceof ObTableEntryRefreshException) {
+                throw new ObTableEntryRefreshException(format(
+                        "fail to get index info from remote, indexTableName: %s, error message: %s", indexTableName, e.getMessage()), e, ((ObTableEntryRefreshException) e).isConnectInactive());
+            } else {
+                throw new ObTableEntryRefreshException(format(
+                        "fail to get index info from remote, indexTableName: %s, error message: %s", indexTableName, e.getMessage()), e);
+            }
         } finally {
             try {
                 if (null != rs) {
@@ -1079,7 +1119,8 @@ public static ObIndexInfo getIndexInfoFromRemote(ObServerAddr obServerAddr, ObUs
 
     private static void fetchFirstPart(Connection connection, TableEntry tableEntry,
                                        ObPartFuncType obPartFuncType)
-                                                                     throws ObTablePartitionInfoRefreshException {
+                                                                     throws ObTablePartitionInfoRefreshException,
+                                                                            SQLException {
         String tableName = "";
         TableEntryKey key = tableEntry.getTableEntryKey();
         if (key != null) {
@@ -1127,6 +1168,8 @@ private static void fetchFirstPart(Connection connection, TableEntry tableEntry,
                 tableEntry.getPartitionInfo().setPartTabletIdMap(
                     parseFirstPartKeyHash(rs, tableEntry));
             }
+        } catch (SQLException e) {
+            throw e;
         } catch (Exception e) {
             RUNTIME.error(LCD.convert("01-00011"), tableEntry, obPartFuncType, e);
 
@@ -1149,7 +1192,8 @@ private static void fetchFirstPart(Connection connection, TableEntry tableEntry,
 
     private static void fetchSubPart(Connection connection, TableEntry tableEntry,
                                      ObPartFuncType subPartFuncType)
-                                                                    throws ObTablePartitionInfoRefreshException {
+                                                                    throws ObTablePartitionInfoRefreshException,
+                                                                           SQLException {
         String tableName = "";
         TableEntryKey key = tableEntry.getTableEntryKey();
         if (key != null) {
@@ -1196,6 +1240,8 @@ private static void fetchSubPart(Connection connection, TableEntry tableEntry,
                 tableEntry.getPartitionInfo().setPartTabletIdMap(
                     parseSubPartKeyHash(rs, tableEntry));
             }
+        } catch (SQLException e) {
+            throw e;
         } catch (Exception e) {
             RUNTIME.error(LCD.convert("01-00012"), tableEntry, subPartFuncType, e);
             throw new ObTablePartitionInfoRefreshException(format(
@@ -1476,7 +1522,8 @@ private static ReplicaLocation buildReplicaLocation(ResultSet rs) throws SQLExce
     }
 
     private static void fetchPartitionInfo(Connection connection, TableEntry tableEntry)
-                                                                                        throws ObTablePartitionInfoRefreshException {
+                                                                                        throws ObTablePartitionInfoRefreshException,
+                                                                                               SQLException {
         PreparedStatement pstmt = null;
         ResultSet rs = null;
         ObPartitionInfo info = null;
@@ -1499,6 +1546,8 @@ private static void fetchPartitionInfo(Connection connection, TableEntry tableEn
                 logger.info("get part info from remote info:{}", JSON.toJSON(info));
             }
             tableEntry.setPartitionInfo(info);
+        } catch (SQLException e) {
+            throw e;
         } catch (Exception e) {
             RUNTIME.error(LCD.convert("01-00014"), tableEntry);
             RUNTIME.error("fail to get part info from remote");

src/main/java/com/alipay/oceanbase/rpc/location/LocationUtil.java

@@ -1664,12 +1664,130 @@ public ObObjMeta getDefaultObjMeta() {
         public Comparable parseToComparable(Object o, ObCollationType ct) {
             throw new FeatureNotSupportedException("ObBitType is not supported .");
         }
-    };
+    },
     /*
     ObEnumType(36)
     ObSetType(37)
-      ObMaxType                    // invalid type, or count of obj type
+    ...
     */
+    ObMySQLDateType(52) {
+        /*
+         * Encode.
+         */
+        @Override
+        public byte[] encode(Object obj) {
+            return Serialization.encodeVi32((int) ((Date) obj).getTime());
+        }
+
+        /*
+         * Decode.
+         */
+        @Override
+        public Object decode(ByteBuf buf, ObCollationType type) {
+            return new Date(Serialization.decodeVi32(buf) * 1000L);
+        }
+
+        /*
+         * Get encoded size.
+         */
+        @Override
+        public int getEncodedSize(Object obj) {
+            return Serialization.getNeedBytes((int) ((Date) obj).getTime());
+        }
+
+        /*
+         * Get default obj meta.
+         */
+        @Override
+        public ObObjMeta getDefaultObjMeta() {
+            return new ObObjMeta(this, ObCollationLevel.CS_LEVEL_NUMERIC,
+                    ObCollationType.CS_TYPE_BINARY, (byte) 10);
+        }
+
+        /*
+         * Parse to comparable.
+         */
+        @Override
+        public Date parseToComparable(Object o, ObCollationType ct)
+                throws IllegalArgumentException,
+                FeatureNotSupportedException {
+
+            if (o instanceof Date) {
+                return (Date) o;
+            }
+
+            if (o instanceof String) {
+                return TimeUtils.strToDate((String) o);
+            }
+
+            if (o instanceof ObVString) {
+                return TimeUtils.strToDate(((ObVString) o).getStringVal());
+            }
+
+            if (o instanceof Long) {
+                return new Date((Long) o);
+            }
+
+            throw new IllegalArgumentException("ObMySQLDateType can not parseToComparable argument:" + o);
+        }
+    },
+    ObMySQLDateTimeType(53) {
+        /*
+         * Encode.
+         */
+        @Override
+        public byte[] encode(Object obj) {
+            // Date do not have timezone, when we use getTime, system will recognize it as our system timezone and transform it into UTC Time, which will changed the time.
+            // We should add back the lose part.
+            long targetTs = ((Date) obj).getTime()
+                    + OffsetDateTime.now().getOffset().getTotalSeconds() * 1000L;
+            return Serialization.encodeVi64(targetTs * 1000L);
+        }
+
+        /*
+         * Decode.
+         */
+        @Override
+        public Object decode(ByteBuf buf, ObCollationType type) {
+            return new Date(Serialization.decodeVi64(buf) / 1000L
+                    - OffsetDateTime.now().getOffset().getTotalSeconds() * 1000L);
+        }
+
+        /*
+         * Get encoded size.
+         */
+        @Override
+        public int getEncodedSize(Object obj) {
+            return Serialization.getNeedBytes(((Date) obj).getTime() * 1000L);
+        }
+
+        /*
+         * Get default obj meta.
+         */
+        @Override
+        public ObObjMeta getDefaultObjMeta() {
+            // scale set into 6 means microSecond
+            return new ObObjMeta(this, ObCollationLevel.CS_LEVEL_NUMERIC,
+                    ObCollationType.CS_TYPE_BINARY, (byte) 6);
+        }
+
+        /*
+         * Parse to comparable.
+         */
+        @Override
+        public Date parseToComparable(Object o, ObCollationType ct)
+                                                                   throws IllegalArgumentException,
+                FeatureNotSupportedException {
+            if (o instanceof String) {
+                return TimeUtils.strToDate((String) o);
+            }
+            return (Date) o;
+        }
+    };
+    /*
+        ObMaxType                    // invalid type, or count of obj type
+    */
+
     private int                            value;
     private static Map<Integer, ObObjType> map = new HashMap<Integer, ObObjType>();