[CP] [OBCDC] Add enable_compatible_charset config

src/logservice/libobcdc/src/ob_log_config.h

@@ -251,6 +251,7 @@ public:
   T_DEF_INT_INFT(svr_stream_cached_count, OB_CLUSTER_PARAMETER, 16, 1, "cached svr stream object count");
   T_DEF_INT_INFT(fetch_stream_cached_count, OB_CLUSTER_PARAMETER, 16, 1, "cached fetch stream object count");
 
+  T_DEF_BOOL(enable_compatible_charset, OB_CLUSTER_PARAMETER, 0,  "0:disabled, 1:enabled");
   // region
   DEF_STR(region, OB_CLUSTER_PARAMETER, "default_region", "OB region");
 

src/logservice/libobcdc/src/ob_log_meta_manager.cpp

@@ -90,6 +90,25 @@ namespace oceanbase
 {
 namespace libobcdc
 {
+void ObLogMetaManager::set_column_encoding_(const common::ObObjType &col_type,
+     const common::ObCharsetType &cs_type,
+     IColMeta *meta)
+{
+  if (1 == TCONF.enable_compatible_charset) {
+    if (cs_type > ObCharsetType::CHARSET_BINARY) {
+      SET_ENCODING(meta, cs_type);
+    } else if (ob_is_string_or_lob_type(col_type) ||
+        ob_is_enum_or_set_type(col_type) ||
+        ob_is_json(col_type)) {
+      SET_ENCODING(meta, cs_type);
+    } else {
+      meta->setEncoding(DEFAULT_ENCODING);
+    }
+  } else {
+    SET_ENCODING(meta, cs_type);
+  }
+}
+
 ObLogMetaManager::ObLogMetaManager() : inited_(false),
                                        enable_output_hidden_primary_key_(false),
                                        obj2str_helper_(NULL),
@@ -1169,7 +1188,8 @@ int ObLogMetaManager::set_column_meta_(
       col_meta->setSigned(signed_flag);
       col_meta->setIsPK(column_schema.is_original_rowkey_column());
       col_meta->setNotNull(! column_schema.is_nullable());
-      SET_ENCODING(col_meta, column_schema.get_charset_type());
+
+      set_column_encoding_(col_type, column_schema.get_charset_type(), col_meta);
 
       if (column_schema.is_xmltype()) {
         SET_ENCODING(col_meta, CS_TYPE_UTF8MB4_GENERAL_CI);

src/logservice/libobcdc/src/ob_log_meta_manager.h

@@ -131,6 +131,12 @@ public:
 
 class ObLogMetaManager : public IObLogMetaManager
 {
+private:
+  static void set_column_encoding_(
+      const common::ObObjType &col_type,
+      const common::ObCharsetType &cs_type,
+      IColMeta *meta);
+
 public:
   ObLogMetaManager();
   virtual ~ObLogMetaManager();