001package gudusoft.gsqlparser.dlineage.dataflow.sqlenv; 002 003import gudusoft.gsqlparser.EDbVendor; 004import gudusoft.gsqlparser.dlineage.dataflow.model.SqlInfo; 005import gudusoft.gsqlparser.dlineage.metadata.Coordinate; 006import gudusoft.gsqlparser.sqlenv.*; 007import gudusoft.gsqlparser.sqlenv.parser.TJSONSQLEnvParser; 008import gudusoft.gsqlparser.sqlenv.parser.TSQLEnvParser; 009import gudusoft.gsqlparser.util.Logger; 010import gudusoft.gsqlparser.util.LoggerFactory; 011import gudusoft.gsqlparser.util.SQLUtil; 012 013import java.io.File; 014import java.util.*; 015import java.util.concurrent.CopyOnWriteArrayList; 016import java.util.concurrent.CountDownLatch; 017import java.util.concurrent.Executors; 018import java.util.concurrent.ThreadPoolExecutor; 019 020public class SQLEnvParser implements TSQLEnvParser { 021 022 private static final Logger logger = LoggerFactory.getLogger(Coordinate.class); 023 024 private TSQLEnv metadataSQLEnv; 025 026 private String defaultServer; 027 private String defaultDatabase; 028 private String defaultSchema; 029 private List<SqlInfo> metadataInfos = new CopyOnWriteArrayList<SqlInfo>(); 030 031 public SQLEnvParser(TSQLEnv metadataSQLEnv, String defaultServer, String defaultDatabase, String defaultSchema) { 032 this.metadataSQLEnv = metadataSQLEnv; 033 this.defaultServer = defaultServer; 034 this.defaultDatabase = defaultDatabase; 035 this.defaultSchema = defaultSchema; 036 } 037 038 /** 039 * The identifier configuration the environments this parser builds are indexed with; null means 040 * each environment uses its own vendor's default. 041 */ 042 private IdentifierProfile identifierProfile; 043 044 /** 045 * Configure the identifier rules the parsed environments register their metadata under. 046 * 047 * <p>Set it before {@link #parseSQLEnv}: DDL and metadata are keyed at registration time, and 048 * the analysis that reads them back must ask under the same rules. 049 * 050 * @param identifierProfile the analysis's configuration, or null for vendor defaults 051 * @return this parser 052 */ 053 public SQLEnvParser withIdentifierProfile(IdentifierProfile identifierProfile) { 054 this.identifierProfile = identifierProfile; 055 return this; 056 } 057 058 public SQLEnvParser(String defaultServer, String defaultDatabase, String defaultSchema) { 059 this.metadataSQLEnv = null; 060 this.defaultServer = defaultServer; 061 this.defaultDatabase = defaultDatabase; 062 this.defaultSchema = defaultSchema; 063 } 064 065 public TSQLEnv[] parseSQLEnv(final EDbVendor vendor, SqlInfo[] sqlInfos) { 066 if (sqlInfos == null || sqlInfos.length == 0) { 067 return null; 068 } 069 final CountDownLatch latch = new CountDownLatch(sqlInfos.length); 070 // The environments of each file, kept in the order of the files: the threads finish in any order, and the 071 // merge below keeps what it meets first, so merging in the order the threads finish gave a different 072 // environment from one run to the next. 073 final TSQLEnv[][] fileSqlEnvs = new TSQLEnv[sqlInfos.length][]; 074 int thread = Runtime.getRuntime().availableProcessors() / 4 + 1; 075 if (thread < 4) thread = 4; 076 ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors 077 .newFixedThreadPool(thread < sqlInfos.length 078 ? thread 079 : sqlInfos.length); 080 for (int i = 0; i < sqlInfos.length; i++) { 081 final SqlInfo sqlInfo = sqlInfos[i]; 082 final int fileIndex = i; 083 Runnable task = new Runnable() { 084 @Override 085 public void run() { 086 try { 087 if (sqlInfo == null) { 088 return; 089 } 090 String sql = sqlInfo.getSql(); 091 if (SQLUtil.isEmpty(sql) && sqlInfo.getFileName() != null 092 && new File(sqlInfo.getFileName()).exists()) { 093 sql = SQLUtil.getFileContent(sqlInfo.getFileName()); 094 } 095 if (SQLUtil.isEmpty(sql) && sqlInfo.getFilePath() != null 096 && new File(sqlInfo.getFilePath()).exists()) { 097 sql = SQLUtil.getFileContent(sqlInfo.getFilePath()); 098 } 099 TJSONSQLEnvParser jsonSQLEnvParser = new TJSONSQLEnvParser(defaultServer, defaultDatabase, defaultSchema) 100 .withIdentifierProfile(identifierProfile); 101 // Pass the manifest's directory so a sharded manifest's relative 102 // catalog paths resolve against it, not the process working 103 // directory (a null baseDir could load a same-named catalog from 104 // CWD and merge a wrong environment). 105 String jsonBaseDir = sqlInfo.getFilePath() != null 106 ? new File(sqlInfo.getFilePath()).getParent() : null; 107 TSQLEnv[] sqlenv = jsonSQLEnvParser.parseSQLEnv(vendor, sql, jsonBaseDir); 108 if (sqlenv == null) { 109 sqlenv = parseSQLEnv(vendor, sql); 110 } 111 else { 112 metadataInfos.add(sqlInfo); 113 } 114 fileSqlEnvs[fileIndex] = sqlenv; 115 } catch (Exception e) { 116 logger.error(e.getMessage(), e); 117 } finally { 118 latch.countDown(); 119 } 120 } 121 }; 122 executor.submit(task); 123 } 124 try { 125 latch.await(); 126 } catch (Exception e) { 127 logger.error("await latch occurs an exception.", e); 128 } 129 executor.shutdown(); 130 // latch.await() makes every task's write to fileSqlEnvs visible here 131 List<TSQLEnv> sqlenvs = new ArrayList<TSQLEnv>(); 132 for (TSQLEnv[] fileSqlEnv : fileSqlEnvs) { 133 if (fileSqlEnv != null) { 134 sqlenvs.addAll(Arrays.asList(fileSqlEnv)); 135 } 136 } 137 Map<String, List<TSQLEnv>> sqlenvMap = new HashMap<String, List<TSQLEnv>>(); 138 for (TSQLEnv sqlenv : sqlenvs) { 139 String serverName = sqlenv.getDefaultServerName(); 140 if(serverName == null) { 141 serverName= TSQLEnv.DEFAULT_SERVER_NAME; 142 } 143 if(!sqlenvMap.containsKey(serverName)) { 144 sqlenvMap.put(serverName, new ArrayList<TSQLEnv>()); 145 } 146 sqlenvMap.get(serverName).add(sqlenv); 147 } 148 149 List<TSQLEnv> mergeSQLEnvs = new ArrayList<TSQLEnv>(); 150 for(String key: sqlenvMap.keySet()) { 151 List<TSQLEnv> sqlenvItem = sqlenvMap.get(key); 152 mergeSQLEnvs.add(mergeSQLEnv(sqlenvItem)); 153 } 154 return mergeSQLEnvs.toArray(new TSQLEnv[0]); 155 } 156 157 public static TSQLEnv mergeSQLEnv(List<TSQLEnv> sqlenvs) { 158 if (sqlenvs.size() == 0) 159 return null; 160 TSQLEnv sqlenv = sqlenvs.get(0); 161 for (int i = 1; i < sqlenvs.size(); i++) { 162 TSQLEnv temp = sqlenvs.get(i); 163 // Merging pours temp's objects into sqlenv's indexes. Those indexes are keyed by 164 // sqlenv's configuration, so objects registered under different rules cannot be 165 // re-keyed here: with tables insensitive on one side and sensitive on the other, 166 // Orders and orders would arrive as one entry or two depending on which environment 167 // happened to be first. Say so instead of silently choosing. 168 if (!sqlenv.getIdentifierProfile().hasSameEffectivePolicyAs(temp.getIdentifierProfile())) { 169 throw new IllegalStateException("SQLEnvParser.mergeSQLEnv: cannot merge metadata" 170 + " indexed under different identifier configurations - " 171 + sqlenv.getIdentifierProfile() + " vs " + temp.getIdentifierProfile() 172 + ". Load every metadata input for one analysis with the same profile."); 173 } 174 if(SQLUtil.isEmpty(sqlenv.getDefaultServerName()) || sqlenv.getDefaultServerName().equals(TSQLEnv.DEFAULT_SERVER_NAME)) { 175 if(!SQLUtil.isEmpty(temp.getDefaultServerName()) && !temp.getDefaultServerName().equals(TSQLEnv.DEFAULT_SERVER_NAME)) { 176 sqlenv.setDefaultServerName(temp.getDefaultServerName()); 177 } 178 } 179 if (temp.getCatalogList() != null) { 180 for (int j = 0; j < temp.getCatalogList().size(); j++) { 181 mergeCatalog(sqlenv, temp.getCatalogList().get(j)); 182 } 183 } 184 } 185// if (sqlenv.getCatalogList() != null && sqlenv.getCatalogList().size() == 1) { 186// sqlenv.setDefaultCatalogName(sqlenv.getCatalogList().get(0).getName()); 187// if (sqlenv.getCatalogList().get(0).getSchemaList() != null 188// && sqlenv.getCatalogList().get(0).getSchemaList().size() == 1) { 189// sqlenv.setDefaultSchemaName(sqlenv.getCatalogList().get(0).getSchemaList().get(0).getName()); 190// } 191// } 192 return sqlenv; 193 } 194 195 private static void mergeCatalog(TSQLEnv sqlenv, TSQLCatalog catalog) { 196 TSQLCatalog mergeCatalog = sqlenv.getSQLCatalog(lexicalName(sqlenv, catalog), true); 197 List<TSQLSchema> schemaList = catalog.getSchemaList(); 198 if (schemaList != null) { 199 for (int i = 0; i < schemaList.size(); i++) { 200 mergeSchema(mergeCatalog, schemaList.get(i)); 201 } 202 } 203 } 204 205 private static void mergeSchema(TSQLCatalog catalog, TSQLSchema schema) { 206 TSQLSchema mergeSchema = catalog.getSchema(lexicalName(catalog.getSqlEnv(), schema), true); 207 List<TSQLSchemaObject> schemaObjectList = schema.getSchemaObjectList(); 208 if (schemaObjectList != null) { 209 for (int i = 0; i < schemaObjectList.size(); i++) { 210 mergeSchemaObject(mergeSchema, schemaObjectList.get(i)); 211 } 212 } 213 } 214 215 private static void mergeSchemaObject(TSQLSchema schema, TSQLSchemaObject schemaObject) { 216 if (schemaObject instanceof TSQLTable) { 217 TSQLTable sqlTable = (TSQLTable) schemaObject; 218 TSQLTable mergeTable = (TSQLTable) findMergeObject(schema, schemaObject); 219 if (mergeTable == null) { 220 mergeTable = schema.createTable(lexicalName(schema.getSqlEnv(), schemaObject), schemaObject.getPriority()); 221 mergeTable.setView(sqlTable.isView()); 222 } else if (schemaObject.getPriority() > mergeTable.getPriority()) { 223 mergeTable.setPriority(schemaObject.getPriority()); 224 } 225 List<TSQLColumn> columnList = sqlTable.getColumnList(); 226 if (columnList != null) { 227 for (int i = 0; i < columnList.size(); i++) { 228 String lexical = lexicalName(schema.getSqlEnv(), columnList.get(i)); 229 // with its type: the fields of a STRUCT or ARRAY column are only in its type. The first known type 230 // wins: a column that an earlier file gave without a type, as a CREATE TABLE AS SELECT does, takes 231 // the type of a later definition, and a type already there is kept. 232 mergeTable.addColumn(lexical, columnList.get(i).getColumnDataType()); 233 mergeTable.fillColumnDataType(lexical, columnList.get(i).getColumnDataType()); 234 if (columnList.get(i).isUnresolvedStar()) { 235 // the star lineage wrote back for an unresolved width travels with the table 236 TSQLColumn merged = mergeTable.getColumn(lexical); 237 if (merged != null) { 238 merged.setUnresolvedStar(true); 239 } 240 } 241 } 242 } 243 } else if (schemaObject instanceof TSQLRoutine) { 244 TSQLRoutine sqlRoutine = (TSQLRoutine) schemaObject; 245 TSQLSchemaObject mergeRoutine = findMergeObject(schema, schemaObject); 246 if (mergeRoutine == null) { 247 mergeRoutine = schema.createSchemaObject(lexicalName(schema.getSqlEnv(), schemaObject), schemaObject.getDataObjectType()); 248 } 249 if (sqlRoutine.getDefinition() != null) { 250 ((TSQLRoutine) mergeRoutine).setDefinition(sqlRoutine.getDefinition()); 251 } 252 // values preset for its parameters (Mantis 4962): a later file's preset of a parameter replaces an earlier 253 // one's, and the replacement is logged 254 if (mergeRoutine instanceof TSQLRoutine) { 255 ((TSQLRoutine) mergeRoutine).mergeParameterPresets(sqlRoutine, "merged metadata"); 256 } 257 } else if (schemaObject instanceof TSQLSynonyms) { 258 TSQLSynonyms sqlSynonyms = (TSQLSynonyms) schemaObject; 259 TSQLSynonyms mergeSynonyms = (TSQLSynonyms) findMergeObject(schema, schemaObject); 260 if (mergeSynonyms == null) { 261 mergeSynonyms = schema.createSynonyms(lexicalName(schema.getSqlEnv(), schemaObject), 262 sqlSynonyms.getSourceDatabase(), 263 sqlSynonyms.getSourceSchema(), 264 sqlSynonyms.getSourceName()); 265 } else if (schemaObject.getPriority() > mergeSynonyms.getPriority()) { 266 mergeSynonyms.setPriority(schemaObject.getPriority()); 267 mergeSynonyms.setBaseTarget(sqlSynonyms.getSourceDatabase(), 268 sqlSynonyms.getSourceSchema(), 269 sqlSynonyms.getSourceName()); 270 } 271 } 272 } 273 274 private static TSQLSchemaObject findMergeObject(TSQLSchema schema, TSQLSchemaObject object) { 275 // All three segments are stored names. The provider accepts them directly and 276 // keys them by identity; a flattened getQualifiedName() loses this provenance. 277 return schema.getSqlEnv().getCatalogProvider().findObject( 278 schema.getCatalog().getName(), schema.getName(), object.getName(), object.getDataObjectType()); 279 } 280 281 private static String lexicalName(TSQLEnv destination, TSQLObject object) { 282 // Legacy get-or-create / addColumn APIs accept SQL spellings. Encode exactly 283 // once at that boundary, preserving case, embedded dots and literal delimiters. 284 return destination.toLexical(object.getDataObjectType(), object.getName(), IdentifierInputForm.CATALOG_STORED); 285 } 286 287 public List<SqlInfo> getMetadataInfos() { 288 return metadataInfos; 289 } 290 291 public void setMetadataInfos(List<SqlInfo> metadataInfos) { 292 this.metadataInfos = metadataInfos; 293 } 294 295 @Override 296 public TSQLEnv[] parseSQLEnv(EDbVendor vendor, String sql) { 297 TDDLSQLEnv ddlSQLEnv = new TDDLSQLEnv(defaultServer, defaultDatabase, defaultSchema, metadataSQLEnv, vendor, 298 sql, identifierProfile); 299 // StatementMetadata registers these at their declaration/call sites during analysis. 300 ddlSQLEnv.setIncludeLocalTemporaryTables(false); 301 ddlSQLEnv.initSQLEnv(); 302 if (ddlSQLEnv.isInit()) { 303 return new TSQLEnv[] { ddlSQLEnv }; 304 } else { 305 return null; 306 } 307 } 308}