fix bug

JNSimba · Nov 29, 2023 · 8886427 · 8886427
1 parent 734662e
commit 8886427
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 4 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/DorisTypeMapper.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/DorisTypeMapper.java
@@ -124,8 +124,9 @@ public String visit(CharType charType) {
 
         @Override
         public String visit(VarCharType varCharType) {
-            int length = varCharType.getLength();
-            return length * 4 > 65533 ? STRING : String.format("%s(%s)", VARCHAR, length * 4);
+            //Flink varchar length max value is int, it may overflow after multiplying by 4
+            long length = varCharType.getLength();
+            return length * 4 >= 65533 ? STRING : String.format("%s(%s)", VARCHAR, length * 4);
         }
 
         @Override

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/rest/RestService.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/rest/RestService.java
@@ -458,10 +458,10 @@ public static Schema getSchema(DorisOptions options, DorisReadOptions readOption
 
     public static boolean isUniqueKeyType(DorisOptions options, DorisReadOptions readOptions, Logger logger)
             throws DorisRuntimeException {
-        //Enable 2pc in multi-table scenario
+        //disable 2pc in multi-table scenario
         if(StringUtils.isBlank(options.getTableIdentifier())){
             logger.info("table model verification is skipped in multi-table scenarios.");
-            return false;
+            return true;
         }
         try {
             return UNIQUE_KEYS_TYPE.equals(getSchema(options, readOptions, logger).getKeysType());

diff --git a/...doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeManager.java b/...doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeManager.java
@@ -120,6 +120,7 @@ public boolean execute(String ddl, String database) throws IOException, IllegalA
         if(StringUtils.isNullOrWhitespaceOnly(ddl)){
             return false;
         }
+        LOG.info("Execute SQL: {}", ddl);
         Map<String, String> param = new HashMap<>();
         param.put("stmt", ddl);
         String requestUrl = String.format(SCHEMA_CHANGE_API,