[FLINK-16464][sql-client]result-mode tableau may shift when content contains Chinese String in SQL CLI

flink-table/flink-sql-client/pom.xml

@@ -109,6 +109,13 @@ under the License.
  <version>3.9.0</version>
  </dependency>
 
+ <!-- Tools to unify display format for different languages -->
+ <dependency>
+ <groupId>com.ibm.icu</groupId>
+ <artifactId>icu4j</artifactId>
+ <version>65.1</version>
+ </dependency>
+
  <!-- configuration -->
  <dependency>
  <groupId>org.apache.flink</groupId>
@@ -521,6 +528,7 @@ under the License.
  <artifactSet>
  <includes combine.children="append">
  <include>org.jline:*</include>
+ <include>com.ibm.icu:icu4j</include>
  </includes>
  </artifactSet>
  </configuration>

flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableauResultView.java

@@ -49,6 +49,8 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Stream;
 
+import static org.apache.flink.table.client.cli.CliUtils.getStringDisplayWidth;
+import static org.apache.flink.table.client.cli.CliUtils.isFullWidth;
 import static org.apache.flink.table.client.cli.CliUtils.rowToString;
 
 /**
@@ -275,11 +277,12 @@ private void printSingleRow(int[] colWidths, String[] cols) {
  int idx = 0;
  for (String col : cols) {
  sb.append(" ");
- if (col.length() <= colWidths[idx]) {
- sb.append(StringUtils.repeat(' ', colWidths[idx] - col.length()));
+ int displayWidth = getStringDisplayWidth(col);
+ if (displayWidth <= colWidths[idx]) {
+ sb.append(StringUtils.repeat(' ', colWidths[idx] - displayWidth));
  sb.append(col);
  } else {
- sb.append(col, 0, colWidths[idx] - COLUMN_TRUNCATED_FLAG.length());
+ sb.append(truncateString(col, colWidths[idx] - COLUMN_TRUNCATED_FLAG.length()));
  sb.append(COLUMN_TRUNCATED_FLAG);
  }
  sb.append(" |");
@@ -389,7 +392,7 @@ private int[] columnWidthsByContent(List<TableColumn> columns, List<String[]> ro
  // fill column width with real data
  for (String[] row : rows) {
  for (int i = 0; i < row.length; ++i) {
- colWidths[i] = Math.max(colWidths[i], row[i].length());
+ colWidths[i] = Math.max(colWidths[i], getStringDisplayWidth(row[i]));
  }
  }
 
@@ -401,4 +404,27 @@ private int[] columnWidthsByContent(List<TableColumn> columns, List<String[]> ro
  return colWidths;
  }
 
+ private String truncateString(String col, int targetWidth) {
+ int passedWidth = 0;
+ int i = 0;
+ for (; i < col.length(); i++) {
+ if (isFullWidth(Character.codePointAt(col, i))) {
+ passedWidth += 2;
+ } else {
+ passedWidth += 1;
+ }
+ if (passedWidth >= targetWidth) {
+ break;
+ }
+ }
+ String substring = col.substring(0, i);
+
+ // pad with ' ' before the column
+ int lackedWidth = targetWidth - getStringDisplayWidth(substring);
+ if (lackedWidth > 0){
+ substring = StringUtils.repeat(' ', lackedWidth) + substring;
+ }
+ return substring;
+ }
+
 }

flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliUtils.java

@@ -23,6 +23,8 @@
 import org.apache.flink.table.utils.EncodingUtils;
 import org.apache.flink.types.Row;
 
+import com.ibm.icu.lang.UCharacter;
+import com.ibm.icu.lang.UProperty;
 import org.jline.utils.AttributedString;
 import org.jline.utils.AttributedStringBuilder;
 import org.jline.utils.AttributedStyle;
@@ -111,4 +113,33 @@ public static String[] typesToString(DataType[] types) {
  }
  return typesAsString;
  }
+
+ public static int getStringDisplayWidth(String str) {
+ int numOfFullWidthCh = (int) str.codePoints().filter(codePoint -> isFullWidth(codePoint)).count();
+ return str.length() + numOfFullWidthCh;
+ }
+
+ /**
+ * Check codePoint is FullWidth or not according to Unicode Standard version 12.0.0.
+ * See https://unicode.org/reports/tr11/
+ */
+ public static boolean isFullWidth(int codePoint) {
+ int value = UCharacter.getIntPropertyValue(codePoint, UProperty.EAST_ASIAN_WIDTH);
+ switch (value) {
+ case UCharacter.EastAsianWidth.NEUTRAL:
+ return false;
+ case UCharacter.EastAsianWidth.AMBIGUOUS:
+ return false;
+ case UCharacter.EastAsianWidth.HALFWIDTH:
+ return false;
+ case UCharacter.EastAsianWidth.FULLWIDTH:
+ return true;
+ case UCharacter.EastAsianWidth.NARROW:
+ return false;
+ case UCharacter.EastAsianWidth.WIDE:
+ return true;
+ default:
+ throw new RuntimeException("unknown UProperty.EAST_ASIAN_WIDTH: " + value);
+ }
+ }
 }

flink-table/flink-sql-client/src/main/resources/META-INF/NOTICE

@@ -9,3 +9,4 @@ See bundled license files for details.
 
 - org.jline:jline-terminal:3.9.0
 - org.jline:jline-reader:3.9.0
+- com.ibm.icu:icu4j:65.1