001/** 002 * Copyright (C) 2006-2026 Talend Inc. - www.talend.com 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 */ 016package org.talend.sdk.component.api.record; 017 018import java.nio.charset.CharsetEncoder; 019import java.nio.charset.StandardCharsets; 020import java.util.Base64; 021import java.util.Objects; 022import java.util.Optional; 023import java.util.function.BiConsumer; 024import java.util.function.Function; 025 026import org.talend.sdk.component.api.record.Schema.Entry; 027 028import lombok.AccessLevel; 029import lombok.NoArgsConstructor; 030 031@NoArgsConstructor(access = AccessLevel.PRIVATE) 032public class SchemaCompanionUtil { 033 034 /** 035 * Sanitize name to be avro compatible. 036 * 037 * @param name : original name. 038 * 039 * @return avro compatible name. 040 */ 041 public static String sanitizeName(final String name) { 042 if (Schema.SKIP_SANITIZE || name == null || name.isEmpty()) { 043 return name; 044 } 045 046 final CharsetEncoder ascii = StandardCharsets.US_ASCII.newEncoder(); 047 final StringBuilder sanitizedBuilder = new StringBuilder(); 048 final char firstLetter = sanitizeFirstLetter(name, ascii); 049 if (firstLetter != (char) -1) { 050 sanitizedBuilder.append(firstLetter); 051 } 052 053 for (int i = 1; i < name.length(); i++) { 054 char current = name.charAt(i); 055 if (ascii.canEncode(current)) { 056 sanitizedBuilder.append(Character.isLetterOrDigit(current) ? current : '_'); 057 } else { 058 if (Character.isLowerCase(current) || Character.isUpperCase(current)) { 059 sanitizedBuilder.append('_'); 060 } else { 061 final byte[] encoded = base64(name.substring(i, i + 1)); 062 final String enc = new String(encoded, StandardCharsets.UTF_8); 063 if (sanitizedBuilder.length() == 0 && Character.isDigit(enc.charAt(0))) { 064 sanitizedBuilder.append('_'); 065 } 066 067 for (int iter = 0; iter < enc.length(); iter++) { 068 final char encodedCurrentChar = enc.charAt(iter); 069 final char sanitizedLetter = Character.isLetterOrDigit(encodedCurrentChar) 070 ? encodedCurrentChar 071 : '_'; 072 sanitizedBuilder.append(sanitizedLetter); 073 } 074 } 075 } 076 077 } 078 return sanitizedBuilder.toString(); 079 } 080 081 private static byte[] base64(final String value) { 082 return Base64.getEncoder().encode(value.getBytes(StandardCharsets.UTF_8)); 083 } 084 085 private static char sanitizeFirstLetter(final String name, final CharsetEncoder ascii) { 086 char current = name.charAt(0); 087 final boolean skipFirstChar = !(ascii.canEncode(current) && validFirstLetter(current)) 088 && name.length() > 1 && !Character.isDigit(name.charAt(1)); 089 090 // indicates that first letter is not valid, so it has to be skipped. 091 // and because the next letter is valid (or can be sanitized) we can use it as first letter. 092 if (skipFirstChar) { 093 return (char) -1; 094 } 095 096 if (validFirstLetter(current) && ascii.canEncode(current)) { 097 return current; 098 } else { 099 return '_'; 100 } 101 } 102 103 private static boolean validFirstLetter(final char value) { 104 return Character.isLetter(value) || value == '_'; 105 } 106 107 /** 108 * May return a different entry with different name. 109 */ 110 public static Schema.Entry avoidCollision(final Schema.Entry newEntry, 111 final Function<String, Entry> entryGetter, 112 final BiConsumer<String, Entry> replaceFunction) { 113 if (Schema.SKIP_SANITIZE) { 114 return newEntry; 115 } 116 117 final Entry alreadyExistedEntry = findCollidedEntry(newEntry, entryGetter); 118 if (alreadyExistedEntry == null) { 119 // No collision, return new entry. 120 return newEntry; 121 } 122 123 final boolean matchedToChange = !isEmpty(alreadyExistedEntry.getRawName()); 124 if (matchedToChange) { 125 // the rename has to be applied on entry already inside schema, so replace. (dunno why) 126 // replace existed entry with a new name 127 final String newSanitizedName = newNotCollidedName(entryGetter, alreadyExistedEntry.getRawName()); 128 final Entry updatedExistedEntry = alreadyExistedEntry.toBuilder() 129 .withName(newSanitizedName) 130 .build(); 131 replaceFunction.accept(alreadyExistedEntry.getName(), updatedExistedEntry); 132 return newEntry; 133 } else if (isEmpty(newEntry.getRawName())) { 134 // try to add exactly same raw, skip the add here. 135 return null; 136 } else { 137 // raw name isn't empty, so we need to create a new entry with a new name (sanitized). 138 final String newSanitizedName = newNotCollidedName(entryGetter, newEntry.getRawName()); 139 return newEntry.toBuilder() 140 .withName(newSanitizedName) 141 .build(); 142 } 143 } 144 145 private static Entry findCollidedEntry(final Entry newEntry, final Function<String, Entry> entryGetter) { 146 return Optional.ofNullable(entryGetter.apply(newEntry.getName())) 147 .filter(retrievedEntry -> !Objects.equals(retrievedEntry, newEntry)) 148 .orElse(null); 149 } 150 151 private static String newNotCollidedName(final Function<String, Entry> entryGetter, final String rawName) { 152 final String baseName = sanitizeName(rawName); 153 int indexForAnticollision = 1; 154 String newName = baseName + "_" + indexForAnticollision; 155 while (entryGetter.apply(newName) != null) { 156 indexForAnticollision++; 157 newName = baseName + "_" + indexForAnticollision; 158 } 159 return newName; 160 } 161 162 private static boolean isEmpty(final String value) { 163 return value == null || value.isEmpty(); 164 } 165}