Temp 4b

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/reader/io/jdbc/uniformsplitter/range/BoundarySplitterFactory.java

@@ -39,6 +39,8 @@ public class BoundarySplitterFactory {
   private static final BigInteger SECONDS_TO_NANOS =
       BigInteger.valueOf(Duration.ofSeconds(1).toNanos());
 
+  @VisibleForTesting protected static final int MAX_STRING_PARTITION_PAD_LENGTH = 300;
+
   private static final ImmutableMap<Class, BoundarySplitter<?>> splittermap =
       ImmutableMap.<Class, BoundarySplitter<?>>builder()
           .put(
@@ -388,7 +390,8 @@ private static byte[] padLeadingZeroBytes(byte[] array, int expectedLength) {
     return result;
   }
 
-  private static String splitStrings(
+  @VisibleForTesting
+  protected static String splitStrings(
       String start,
       String end,
       PartitionColumn partitionColumn,
@@ -413,15 +416,33 @@ private static String splitStrings(
     // during a run.
     // To avoid undefined behaviour in the padding logic, we take the max of the input strings and
     // the partition column width.
+    int commonPrefixLength = 0;
+    while (commonPrefixLength < start.length() && commonPrefixLength < end.length()) {
+      int cpStart = start.codePointAt(commonPrefixLength);
+      int cpEnd = end.codePointAt(commonPrefixLength);
+      if (cpStart != cpEnd) {
+        break;
+      }
+      commonPrefixLength += Character.charCount(cpStart);
+    }
+    String commonPrefix = start.substring(0, commonPrefixLength);
+    String suffixStart = start.substring(commonPrefixLength);
+    String suffixEnd = end.substring(commonPrefixLength);
+
     int lengthToPad =
         Math.max(
-            Math.max(start.length(), end.length()), partitionColumn.stringMaxLength().intValue());
+            Math.max(suffixStart.length(), suffixEnd.length()),
+            Math.min(
+                Math.max(0, partitionColumn.stringMaxLength().intValue() - commonPrefixLength),
+                MAX_STRING_PARTITION_PAD_LENGTH));
     BigInteger bigIntegerStart =
-        (BigInteger) typeMapper.mapStringToBigInteger(start, lengthToPad, partitionColumn, c);
+        (BigInteger) typeMapper.mapStringToBigInteger(suffixStart, lengthToPad, partitionColumn, c);
     BigInteger bigIntegerEnd =
-        (BigInteger) typeMapper.mapStringToBigInteger(end, lengthToPad, partitionColumn, c);
+        (BigInteger) typeMapper.mapStringToBigInteger(suffixEnd, lengthToPad, partitionColumn, c);
     BigInteger bigIntegerSplit = splitBigIntegers(bigIntegerStart, bigIntegerEnd);
-    return (String) typeMapper.unMapStringFromBigInteger(bigIntegerSplit, partitionColumn, c);
+    String suffixMid =
+        (String) typeMapper.unMapStringFromBigInteger(bigIntegerSplit, partitionColumn, c);
+    return commonPrefix + suffixMid;
   }
 
   @VisibleForTesting

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/reader/io/jdbc/uniformsplitter/stringmapper/CollationIndex.java

@@ -16,6 +16,7 @@
 package com.google.cloud.teleport.v2.reader.io.jdbc.uniformsplitter.stringmapper;
 
 import com.google.auto.value.AutoValue;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import java.io.Serializable;
@@ -43,15 +44,15 @@ public abstract class CollationIndex implements Serializable {
    * Map of character to it's index position based on collation order. Helps us map a string to big
    * integer.
    */
-  public abstract ImmutableMap<Character, Long> characterToIndex();
+  public abstract ImmutableMap<String, Long> characterToIndex();
 
   /**
    * Map if Index back to character based on collation order. Helps us unmap a big integer to
    * string. Note this maps the index back to a minimum set of characters. For example in
    * case-insensitive collations, 'a' and 'A' will have the same index in {@link
    * #characterToIndex()} and {@link #indexToCharacter()} will map the index to 'A'.
    */
-  public abstract ImmutableMap<Long, Character> indexToCharacter();
+  public abstract ImmutableMap<Long, String> indexToCharacter();
 
   public static CollationIndex.Builder builder() {
     return new AutoValue_CollationIndex.Builder();
@@ -61,11 +62,11 @@ public long getCharsetSize() {
     return indexToCharacter().size();
   }
 
-  public long getOrdinalPosition(Character c) {
+  public long getOrdinalPosition(String c) {
     return characterToIndex().get(c);
   }
 
-  public Character getCharacterFromPosition(Long position) {
+  public String getCharacterFromPosition(Long position) {
     return indexToCharacter().get(position);
   }
 
@@ -80,15 +81,19 @@ public abstract static class Builder {
 
     abstract CollationIndexType indexType();
 
-    private Map<Character, Long> charToIndexCache = new HashMap<>();
-    private Map<Long, Character> indexToCharacterCache = new HashMap<>();
-    private Map<Character, Long> indexToCharacterReverseCache = new HashMap<>();
+    private Map<String, Long> charToIndexCache = new HashMap<>();
+    private Map<Long, String> indexToCharacterCache = new HashMap<>();
+    private Map<String, Long> indexToCharacterReverseCache = new HashMap<>();
 
-    abstract Builder setIndexToCharacter(ImmutableMap<Long, Character> value);
+    abstract Builder setIndexToCharacter(ImmutableMap<Long, String> value);
 
-    abstract Builder setCharacterToIndex(ImmutableMap<Character, Long> value);
+    abstract Builder setCharacterToIndex(ImmutableMap<String, Long> value);
+
+    public Builder addCharacter(String charsetChar, String equivalentChar, Long index) {
+      Preconditions.checkNotNull(charsetChar, "charsetChar cannot be null");
+      Preconditions.checkNotNull(equivalentChar, "equivalentChar cannot be null");
+      Preconditions.checkNotNull(index, "index cannot be null");
 
-    public Builder addCharacter(Character charsetChar, Character equivalentChar, Long index) {
       logger.debug(
           "Registering character order for {}, index-type = {}, character = {}, equivalentCharacter = {}, index = {}, isBlank = {}",
           collationReference(),

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/reader/io/jdbc/uniformsplitter/stringmapper/CollationMapper.java

@@ -81,7 +81,7 @@ public abstract class CollationMapper implements Serializable {
    * using utf8mb4), 'b') = 'ab' COLLATE <collation>;} returns 1. TODO(vardhanvthigle): Check this
    * behavior for PG and other databases.
    */
-  public abstract ImmutableSet<Character> emptyCharacters();
+  public abstract ImmutableSet<String> emptyCharacters();
 
   /**
    * Space Characters. MySQL ignores trailing space characters in comparisons for PAD space
@@ -90,11 +90,11 @@ public abstract class CollationMapper implements Serializable {
    * (UNHEX(C2H0)) when the collation is Pad Space. These have same behavior to ascii space as far
    * as trailing or non-trailing comparison is concerned.
    */
-  public abstract ImmutableSet<Character> spaceCharacters();
+  public abstract ImmutableSet<String> spaceCharacters();
 
   @Memoized
   String allSpaceCharacters() {
-    return this.spaceCharacters().stream().map(String::valueOf).collect(Collectors.joining(""));
+    return this.spaceCharacters().stream().collect(Collectors.joining(""));
   }
 
   @Memoized
@@ -103,8 +103,7 @@ String emptyReplacePattern() {
       return "";
     }
     return "["
-        + Pattern.quote(
-            this.emptyCharacters().stream().map(String::valueOf).collect(Collectors.joining("")))
+        + Pattern.quote(this.emptyCharacters().stream().collect(Collectors.joining("")))
         + "]";
   }
 
@@ -131,6 +130,14 @@ public BigInteger mapString(@Nullable String element, int lengthToPad) {
     if (element == null) {
       return BigInteger.valueOf(-1);
     }
+    // 'ret' stores the mapped value using a variable-base encoding.
+    // The base (charset size) can change depending on whether it's the trailing position
+    // in a pad-space collation.
+    // Example: For string "abcd" with lengthToPad = 6, let non-trailing base = 100 and trailing
+    // base = 90.
+    // If ordinals are a=1, b=2, c=3, d=4, the mapping evaluates to:
+    // ret = ((((1 * 100 + 2) * 100) + 3) * 90 + 4) * (100 ^ 2)
+    // unMapString reverses this by extracting modulo the trailing base first.
     BigInteger ret = BigInteger.ZERO;
 
     // MySQL ignores empty character in string comparisons.
@@ -150,14 +157,21 @@ public BigInteger mapString(@Nullable String element, int lengthToPad) {
     }
 
     // Convert the string to BigInteger.
-    for (int index = 0; index < element.length(); index++) {
-      Character c = element.charAt(index);
+    java.util.List<String> codePoints =
+        element
+            .codePoints()
+            .mapToObj(cp -> new String(Character.toChars(cp)))
+            .collect(Collectors.toList());
+    for (int index = 0; index < codePoints.size(); index++) {
+      String c = codePoints.get(index);
       ret =
-          ret.multiply(BigInteger.valueOf(getCharsetSize(index == (element.length() - 1))))
-              .add(BigInteger.valueOf(getOrdinalPosition(c, index == (element.length() - 1))));
+          ret.multiply(BigInteger.valueOf(getCharsetSize(index == (codePoints.size() - 1))))
+              .add(BigInteger.valueOf(getOrdinalPosition(c, index == (codePoints.size() - 1))));
     }
-    for (int index = element.length(); index < lengthToPad; index++) {
-      ret = ret.multiply(BigInteger.valueOf(getCharsetSize(index == (element.length() - 1))));
+    if (lengthToPad > codePoints.size()) {
+      ret =
+          ret.multiply(
+              BigInteger.valueOf(getCharsetSize(false)).pow(lengthToPad - codePoints.size()));
     }
     return ret;
   }
@@ -188,16 +202,16 @@ public String unMapString(BigInteger element) {
     }
 
     // Base Case that the string just represents single character
-    if (element == BigInteger.ZERO) {
-      char c = getCharacterFromPosition(element.longValue(), true);
-      return String.valueOf(c);
+    if (element.equals(BigInteger.ZERO)) {
+      String c = getCharacterFromPosition(element.longValue(), true);
+      return c;
     }
 
-    while (element != BigInteger.ZERO) {
+    while (!element.equals(BigInteger.ZERO)) {
       long charsetSize = getCharsetSize(index == 0);
 
       BigInteger reminder = element.mod(BigInteger.valueOf(charsetSize));
-      char c = getCharacterFromPosition(reminder.longValue(), (index == 0));
+      String c = getCharacterFromPosition(reminder.longValue(), (index == 0));
       word.append(c);
 
       element = element.divide(BigInteger.valueOf(charsetSize));
@@ -275,13 +289,13 @@ private long getCharsetSize(boolean lastCharacter) {
         : this.allPositionsIndex().getCharsetSize();
   }
 
-  private long getOrdinalPosition(Character c, boolean lastCharacter) {
+  private long getOrdinalPosition(String c, boolean lastCharacter) {
     return (lastCharacter && collationReference().padSpace())
         ? this.trailingPositionsPadSpace().getOrdinalPosition(c)
         : this.allPositionsIndex().getOrdinalPosition(c);
   }
 
-  private Character getCharacterFromPosition(long ordinalPosition, boolean firstIteration) {
+  private String getCharacterFromPosition(long ordinalPosition, boolean firstIteration) {
     return (firstIteration && collationReference().padSpace())
         ? this.trailingPositionsPadSpace().getCharacterFromPosition(ordinalPosition)
         : this.allPositionsIndex().getCharacterFromPosition(ordinalPosition);
@@ -307,9 +321,9 @@ public abstract static class Builder {
 
     abstract CollationIndex.Builder trailingPositionsPadSpaceBuilder();
 
-    abstract ImmutableSet.Builder<Character> emptyCharactersBuilder();
+    abstract ImmutableSet.Builder<String> emptyCharactersBuilder();
 
-    abstract ImmutableSet.Builder<Character> spaceCharactersBuilder();
+    abstract ImmutableSet.Builder<String> spaceCharactersBuilder();
 
     public Builder addCharacter(CollationOrderRow collationOrderRow) {
 

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/reader/io/jdbc/uniformsplitter/stringmapper/CollationOrderRow.java

@@ -41,10 +41,10 @@ public abstract class CollationOrderRow {
   private static final Logger logger = LoggerFactory.getLogger(CollationOrderRow.class);
 
   /** Character in the character set. */
-  public abstract Character charsetChar();
+  public abstract String charsetChar();
 
   /** A character with lowest rank charset_char character is equal to as per the collation. */
-  public abstract Character equivalentChar();
+  public abstract String equivalentChar();
 
   /** 0 offset rank of this character as per the collation sort ordering at all positions. */
   public abstract Long codepointRank();
@@ -54,7 +54,7 @@ public abstract class CollationOrderRow {
    * trailing position, in case a PAD SPACE comparison is needed. Unless you are looking at space
    * like characters, this will be exactly same as equivalent_character.
    */
-  public abstract Character equivalentCharPadSpace();
+  public abstract String equivalentCharPadSpace();
 
   /**
    * A character with lowest rank charset_char character is equal to as per the collation at
@@ -111,20 +111,25 @@ public static CollationOrderRow fromRS(ResultSet rs) throws SQLException {
         isSpace);
 
     Preconditions.checkArgument(
-        charSetChar.length() <= 1, "Found a long character in collation output " + charSetChar);
+        charSetChar.codePointCount(0, charSetChar.length()) <= 1,
+        "Found a multi-codepoint character in collation output: " + charSetChar);
     Preconditions.checkArgument(
-        equivalentCharsetChar.length() <= 1,
-        "Found a long equivalent character in collation output " + equivalentCharsetChar);
-    Preconditions.checkArgument(
-        equivalentCharsetCharPadSpace.length() <= 1,
-        "Found a long equivalent character for pad space in collation output "
+        equivalentCharsetChar.codePointCount(0, equivalentCharsetChar.length()) <= 1,
+        "Found a multi-codepoint equivalent character in collation output: "
             + equivalentCharsetChar);
+    Preconditions.checkArgument(
+        equivalentCharsetCharPadSpace == null
+            || equivalentCharsetCharPadSpace.codePointCount(
+                    0, equivalentCharsetCharPadSpace.length())
+                <= 1,
+        "Found a multi-codepoint equivalent character for pad space in collation output: "
+            + equivalentCharsetCharPadSpace);
 
     return CollationOrderRow.builder()
-        .setCharsetChar(charSetChar.charAt(0))
-        .setEquivalentChar(equivalentCharsetChar.charAt(0))
+        .setCharsetChar(charSetChar)
+        .setEquivalentChar(equivalentCharsetChar)
         .setCodepointRank(codePointRank)
-        .setEquivalentCharPadSpace(equivalentCharsetCharPadSpace.charAt(0))
+        .setEquivalentCharPadSpace(equivalentCharsetCharPadSpace)
         .setCodepointRankPadSpace(codePointRankPadSpace)
         .setIsEmpty(isEmpty)
         .setIsSpace(isSpace)
@@ -134,13 +139,13 @@ public static CollationOrderRow fromRS(ResultSet rs) throws SQLException {
   @AutoValue.Builder
   public abstract static class Builder {
 
-    public abstract Builder setCharsetChar(Character value);
+    public abstract Builder setCharsetChar(String value);
 
-    public abstract Builder setEquivalentChar(Character value);
+    public abstract Builder setEquivalentChar(String value);
 
     public abstract Builder setCodepointRank(Long value);
 
-    public abstract Builder setEquivalentCharPadSpace(Character value);
+    public abstract Builder setEquivalentCharPadSpace(String value);
 
     public abstract Builder setCodepointRankPadSpace(Long value);