001package gudusoft.gsqlparser.dlineage.util; 002 003import gudusoft.gsqlparser.dlineage.comment.CommentBindingBundle; 004import gudusoft.gsqlparser.dlineage.comment.CommentBindingProjection; 005import gudusoft.gsqlparser.EDbVendor; 006import gudusoft.gsqlparser.dlineage.DataFlowAnalyzer; 007import gudusoft.gsqlparser.dlineage.ParallelDataFlowAnalyzer; 008import gudusoft.gsqlparser.dlineage.dataflow.metadata.MetadataReader; 009import gudusoft.gsqlparser.dlineage.dataflow.metadata.sqldep.SQLDepMetadataAnalyzer; 010import gudusoft.gsqlparser.dlineage.dataflow.model.ModelBindingManager; 011import gudusoft.gsqlparser.dlineage.dataflow.model.Option; 012import gudusoft.gsqlparser.sqlenv.CanonKey; 013import gudusoft.gsqlparser.sqlenv.ESQLDataObjectType; 014import gudusoft.gsqlparser.sqlenv.IdentifierProfile; 015import gudusoft.gsqlparser.sqlenv.IdentifierService; 016import gudusoft.gsqlparser.sqlenv.QualifiedCanonKey; 017import gudusoft.gsqlparser.dlineage.dataflow.model.RelationshipType; 018import gudusoft.gsqlparser.dlineage.dataflow.model.json.Error; 019import gudusoft.gsqlparser.dlineage.dataflow.model.json.Process; 020import gudusoft.gsqlparser.dlineage.dataflow.model.json.*; 021import gudusoft.gsqlparser.dlineage.dataflow.model.xml.*; 022import gudusoft.gsqlparser.dlineage.metadata.Sqlflow; 023import gudusoft.gsqlparser.util.Logger; 024import gudusoft.gsqlparser.util.LoggerFactory; 025import gudusoft.gsqlparser.util.SQLUtil; 026 027import java.io.File; 028import java.util.*; 029import java.util.concurrent.atomic.AtomicLong; 030import java.util.stream.Collectors; 031 032public class DataflowUtility { 033 034 private static final Logger logger = LoggerFactory.getLogger(FunctionUtility.class); 035 036 public static dataflow mergeFunctionCallDataflow(dataflow dataflow, EDbVendor dbVendor) { 037 if (ModelBindingManager.getGlobalVendor() == null) { 038 ModelBindingManager.setGlobalVendor(dbVendor); 039 } 040 dataflow instance = cloneDataflow(dataflow); 041 Map<String,String> commentObjects = instance.getCommentBindings()==null ? null : new HashMap<String,String>(); 042 Map<String,String> commentRelations = instance.getCommentBindings()==null ? null : new HashMap<String,String>(); 043 List<procedure> procedures = new ArrayList<>(instance.getProcedures()); 044 if (instance.getPackages() != null) { 045 for (oraclePackage pkg : instance.getPackages()) { 046 if (pkg.getProcedures() != null) { 047 for (procedure procedure : pkg.getProcedures()) { 048 procedure.setOraclePackage(pkg); 049 procedures.add(procedure); 050 } 051 } 052 } 053 } 054 055 Map<String, procedure> procedureIdMap = procedures.stream().collect(Collectors.toMap( 056 t -> t.getId(), 057 t -> t, 058 (existingValue, newValue) -> newValue 059 )); 060 Map<String, table> functionIdMap = dataflow.getResultsets().stream().collect(Collectors.toMap( 061 t -> t.getId(), 062 t -> t, 063 (existingValue, newValue) -> newValue 064 )); 065 066 Map<String, table> functionMap = new HashMap<>(); 067 Map<String, procedure> procedureMap = new HashMap<>(); 068 Map<String, oraclePackage> oraclePackageMap = new HashMap<>(); 069 Map<String, Set<String>> oraclePackageProcedureMap = new HashMap<>(); 070 if (instance.getPackages() != null) { 071 for (oraclePackage pkg : instance.getPackages()) { 072 String qualifiedPackageName = DlineageUtil.getIdentifierOraclePackageNameWithArgNum(pkg); 073 if (!oraclePackageMap.containsKey(qualifiedPackageName)) { 074 oraclePackageMap.put(qualifiedPackageName, pkg); 075 } 076 if (commentObjects != null) commentObjects.put(pkg.getId(), oraclePackageMap.get(qualifiedPackageName).getId()); 077 for (procedure procedure : pkg.getProcedures()) { 078 String qualifiedProcedureName = qualifiedPackageName + "." + DlineageUtil.getIdentifierProcedureNameWithArgNum(procedure); 079 if (!oraclePackageProcedureMap.containsKey(qualifiedPackageName)) { 080 oraclePackageProcedureMap.put(qualifiedPackageName, new HashSet<>()); 081 } 082 oraclePackageProcedureMap.get(qualifiedPackageName).add(qualifiedProcedureName); 083 } 084 } 085 } 086 087 088 089 for (procedure procedure : procedures) { 090 String qualifiedProcedureName = DlineageUtil.getIdentifierProcedureNameWithArgNum(procedure); 091 if (procedure.getOraclePackage() != null) { 092 String qualifiedPackageName = DlineageUtil.getIdentifierOraclePackageNameWithArgNum(procedure.getOraclePackage()); 093 qualifiedProcedureName = qualifiedPackageName + "." + DlineageUtil.getIdentifierProcedureNameWithArgNum(procedure); 094 } 095 if (!procedureMap.containsKey(qualifiedProcedureName)) { 096 procedureMap.put(qualifiedProcedureName, procedure); 097 } 098 if (commentObjects != null) commentObjects.put(procedure.getId(), procedureMap.get(qualifiedProcedureName).getId()); 099 } 100 101 for (table function : instance.getResultsets()) { 102 String qualifiedFunctionName = DlineageUtil.getIdentifierFunctionName(function); 103 if (!functionMap.containsKey(qualifiedFunctionName)) { 104 functionMap.put(qualifiedFunctionName, function); 105 } 106 if (commentObjects != null) commentObjects.put(function.getId(), functionMap.get(qualifiedFunctionName).getId()); 107 } 108 109 Set<oraclePackage> oraclePackages = new LinkedHashSet<>(); 110 for (String key : oraclePackageMap.keySet()) { 111 oraclePackage standardPackage = oraclePackageMap.get(key); 112 oraclePackages.add(standardPackage); 113 Set<String> procedureNames = oraclePackageProcedureMap.get(key); 114 if (procedureNames != null) { 115 for (String procedureName : procedureNames) { 116 procedureName = key + "." + procedureName; 117 if (procedureMap.containsKey(procedureName)) { 118 procedure standardProcedure = procedureMap.get(procedureName); 119 if (!standardPackage.getProcedures().contains(standardProcedure)) { 120 standardPackage.getProcedures().add(standardProcedure); 121 } 122 procedureMap.remove(procedureName); 123 } 124 } 125 } 126 } 127 128 instance.setPackages(new ArrayList<>(oraclePackages)); 129 instance.setProcedures(new ArrayList<>(procedureMap.values())); 130 instance.setResultsets(new ArrayList<>(functionMap.values())); 131 132 Map<String, relationship> mergeRelations = new LinkedHashMap<String, relationship>(); 133 134 for (relationship relationship : instance.getRelationships()) { 135 if (relationship.getCaller() == null || relationship.getCallees() == null || relationship.getCallees().size() == 0) { 136 continue; 137 } 138 139 String targetId = relationship.getCaller().getId(); 140 if (procedureIdMap.containsKey(targetId)) { 141 procedure procedure = procedureIdMap.get(targetId); 142 String procedureName = DlineageUtil.getIdentifierProcedureNameWithArgNum(procedure); 143 if (procedure.getOraclePackage() != null) { 144 String qualifiedPackageName = DlineageUtil.getIdentifierOraclePackageNameWithArgNum(procedure.getOraclePackage()); 145 procedureName = qualifiedPackageName + "." + procedureName; 146 } 147 procedure standardProcedure = procedureMap.get(procedureName); 148 relationship.getCaller().setId(standardProcedure.getId()); 149 } else if (functionIdMap.containsKey(targetId)) { 150 table function = functionIdMap.get(targetId); 151 String functionName = DlineageUtil.getIdentifierFunctionName(function); 152 table standardFunction = functionMap.get(functionName); 153 relationship.getCaller().setId(standardFunction.getId()); 154 } 155 156 for (sourceColumn sourceColumn : relationship.getCallees()) { 157 String sourceId = sourceColumn.getId(); 158 if (procedureIdMap.containsKey(sourceId)) { 159 procedure procedure = procedureIdMap.get(sourceId); 160 String procedureName = DlineageUtil.getIdentifierProcedureNameWithArgNum(procedure); 161 if (procedure.getOraclePackage() != null) { 162 String qualifiedPackageName = DlineageUtil.getIdentifierOraclePackageNameWithArgNum(procedure.getOraclePackage()); 163 procedureName = qualifiedPackageName + "." + procedureName; 164 } 165 procedure standardProcedure = procedureMap.get(procedureName); 166 sourceColumn.setId(standardProcedure.getId()); 167 } else if (functionIdMap.containsKey(sourceId)) { 168 table function = functionIdMap.get(sourceId); 169 String functionName = DlineageUtil.getIdentifierFunctionName(function); 170 table standardFunction = functionMap.get(functionName); 171 sourceColumn.setId(standardFunction.getId()); 172 } 173 } 174 175 String key = relationship.toDedupKey(); 176 if (!mergeRelations.containsKey(key)) { 177 mergeRelations.put(key, relationship); 178 } 179 if (commentRelations != null) commentRelations.put(relationship.getId(), mergeRelations.get(key).getId()); 180 } 181 182 instance.setRelationships(new ArrayList<relationship>(mergeRelations.values())); 183 184 if (instance.getPackages() != null) { 185 for (oraclePackage pkg : instance.getPackages()) { 186 if (pkg.getProcedures() != null) { 187 for (procedure procedure : pkg.getProcedures()) { 188 procedure.setOraclePackage(null); 189 } 190 } 191 } 192 } 193 194 if (commentObjects != null) { 195 CommentBindingProjection.remap(instance, commentObjects, Collections.<String,String>emptyMap(), commentRelations); 196 CommentBindingProjection.retain(instance); 197 } 198 return instance; 199 } 200 201 public static dataflow convertTableLevelToFunctionCallDataflow(dataflow dataflow, boolean showBuiltIn, EDbVendor dbVendor) { 202 if (ModelBindingManager.getGlobalVendor() == null) { 203 ModelBindingManager.setGlobalVendor(dbVendor); 204 } 205 206 dataflow instance = cloneDataflow(dataflow); 207 208 if (instance.getRelationships() == null) { 209 return instance; 210 } 211 212 List<relationship> callRelationships = instance.getRelationships().stream() 213 .filter(t -> RelationshipType.call.name().equals(t.getType())) 214 .filter(t -> !showBuiltIn ? !Boolean.TRUE.equals(t.getBuiltIn()): true).collect(Collectors.toList()); 215 instance.setRelationships(callRelationships); 216 217 Set<String> ids = new HashSet<>(); 218 219 220 callRelationships.stream().forEach(t -> { 221 ids.add(t.getCaller().getId()); 222 t.getCallees().stream().forEach(t1 -> ids.add(t1.getId())); 223 }); 224 225 Iterator<table> iterator = instance.getTables().iterator(); 226 while (iterator.hasNext()) { 227 table t = iterator.next(); 228 if (!ids.contains(t.getId())) { 229 iterator.remove(); 230 } 231 } 232 233 iterator = instance.getResultsets().iterator(); 234 while (iterator.hasNext()) { 235 table t = iterator.next(); 236 if (!ids.contains(t.getId())) { 237 iterator.remove(); 238 } 239 } 240 241 iterator = instance.getViews().iterator(); 242 while (iterator.hasNext()) { 243 table t = iterator.next(); 244 if (!ids.contains(t.getId())) { 245 iterator.remove(); 246 } 247 } 248 249 iterator = instance.getStages().iterator(); 250 while (iterator.hasNext()) { 251 table t = iterator.next(); 252 if (!ids.contains(t.getId())) { 253 iterator.remove(); 254 } 255 } 256 257 iterator = instance.getStreams().iterator(); 258 while (iterator.hasNext()) { 259 table t = iterator.next(); 260 if (!ids.contains(t.getId())) { 261 iterator.remove(); 262 } 263 } 264 265 iterator = instance.getVariables().iterator(); 266 while (iterator.hasNext()) { 267 table t = iterator.next(); 268 if (!ids.contains(t.getId())) { 269 iterator.remove(); 270 } 271 } 272 273 iterator = instance.getPaths().iterator(); 274 while (iterator.hasNext()) { 275 table t = iterator.next(); 276 if (!ids.contains(t.getId())) { 277 iterator.remove(); 278 } 279 } 280 281 iterator = instance.getDatasources().iterator(); 282 while (iterator.hasNext()) { 283 table t = iterator.next(); 284 if (!ids.contains(t.getId())) { 285 iterator.remove(); 286 } 287 } 288 289 iterator = instance.getDatabases().iterator(); 290 while (iterator.hasNext()) { 291 table t = iterator.next(); 292 if (!ids.contains(t.getId())) { 293 iterator.remove(); 294 } 295 } 296 297 iterator = instance.getSchemas().iterator(); 298 while (iterator.hasNext()) { 299 table t = iterator.next(); 300 if (!ids.contains(t.getId())) { 301 iterator.remove(); 302 } 303 } 304 305 iterator = instance.getSequences().iterator(); 306 while (iterator.hasNext()) { 307 table t = iterator.next(); 308 if (!ids.contains(t.getId())) { 309 iterator.remove(); 310 } 311 } 312 313 return mergeFunctionCallDataflow(instance, dbVendor); 314 } 315 316 public static dataflow convertToTableLevelDataflow(dataflow dataflow) { 317 dataflow instance = cloneDataflow(dataflow); 318 319 if (instance.getRelationships() == null) { 320 return instance; 321 } 322 323 Map<String, LinkedHashSet<Pair<String, String>>> relationMap = new HashMap<String, LinkedHashSet<Pair<String, String>>>(); 324 Map<String, LinkedHashSet<Pair3<String, String, relationship>>> callRelationMap = new HashMap<String, LinkedHashSet<Pair3<String, String, relationship>>>(); 325 for (RelationshipType type : RelationshipType.values()) { 326 relationMap.put(type.name(), new LinkedHashSet<Pair<String, String>>()); 327 callRelationMap.put(type.name(), new LinkedHashSet<Pair3<String, String, relationship>>()); 328 } 329 for (relationship relationship : instance.getRelationships()) { 330 if (RelationshipType.call.name().equals(relationship.getType())) { 331 String targetId = relationship.getCaller().getId(); 332 for (sourceColumn sourceColumn : relationship.getCallees()) { 333 String sourceId = sourceColumn.getId(); 334 callRelationMap.get(relationship.getType()).add(new Pair3<>(targetId, sourceId, relationship)); 335 } 336 } else { 337 String targetId = relationship.getTarget().getParent_id(); 338 for (sourceColumn sourceColumn : relationship.getSources()) { 339 String sourceId = sourceColumn.getParent_id(); 340 relationMap.get(relationship.getType()).add(new Pair<>(targetId, sourceId)); 341 } 342 } 343 } 344 345 long maxId = 0; 346 Map<String, table> dbObjMap = getDataflowDbObjMap(instance); 347 348 List<procedure> procedures = new ArrayList<>(instance.getProcedures()); 349 if (instance.getPackages() != null) { 350 for (oraclePackage pkg : instance.getPackages()) { 351 procedures.addAll(pkg.getProcedures()); 352 } 353 } 354 355 Map<String, procedure> procedureMap = procedures.stream().collect(Collectors.toMap( 356 t -> t.getId(), 357 t -> t, 358 (existingValue, newValue) -> newValue 359 )); 360 361 for (table table : dbObjMap.values()) { 362 if (Long.valueOf(table.getId()) > maxId) { 363 maxId = Long.valueOf(table.getId()); 364 } 365 } 366 367 AtomicLong id = new AtomicLong(maxId + 10000000); 368 369 for (table table : dbObjMap.values()) { 370 column column = new column(); 371 column.setId(String.valueOf(id.incrementAndGet())); 372 column.setName(table.getType()); 373 table.setColumns(Arrays.asList(column)); 374 } 375 376 List<relationship> relations = new ArrayList<relationship>(); 377 for (RelationshipType type : RelationshipType.values()) { 378 LinkedHashSet<Pair<String, String>> relationSet = relationMap.get(type.name()); 379 LinkedHashSet<Pair3<String, String, relationship>> callRelationSet = callRelationMap.get(type.name()); 380 if (RelationshipType.call.equals(type)) { 381 for (Pair3<String, String, relationship> pair : callRelationSet) { 382 if ((dbObjMap.get(pair.first) == null && procedureMap.get(pair.first) == null) 383 || (dbObjMap.get(pair.second) == null && procedureMap.get(pair.second) == null)) { 384 continue; 385 } 386 387 relationship relationship = new relationship(); 388 relationship.setType(type.name()); 389 relationship.setId(String.valueOf(id.incrementAndGet())); 390 relationship.setCallStmt(pair.third.getCallStmt()); 391 relationship.setCallCoordinate(pair.third.getCallCoordinate()); 392 393 targetColumn targetColumn = new targetColumn(); 394 if (dbObjMap.containsKey(pair.first)) { 395 targetColumn.setId(dbObjMap.get(pair.first).getId()); 396 targetColumn.setName(dbObjMap.get(pair.first).getName()); 397 targetColumn.setCoordinate(dbObjMap.get(pair.first).getCoordinate()); 398 } else { 399 targetColumn.setId(procedureMap.get(pair.first).getId()); 400 targetColumn.setName(procedureMap.get(pair.first).getName()); 401 targetColumn.setCoordinate(procedureMap.get(pair.first).getCoordinate()); 402 } 403 404 sourceColumn sourceColumn = new sourceColumn(); 405 if (dbObjMap.containsKey(pair.second)) { 406 sourceColumn.setId(dbObjMap.get(pair.second).getId()); 407 sourceColumn.setName(dbObjMap.get(pair.second).getName()); 408 sourceColumn.setCoordinate(dbObjMap.get(pair.second).getCoordinate()); 409 } else { 410 sourceColumn.setId(procedureMap.get(pair.second).getId()); 411 sourceColumn.setName(procedureMap.get(pair.second).getName()); 412 sourceColumn.setCoordinate(procedureMap.get(pair.second).getCoordinate()); 413 } 414 415 relationship.setCaller(targetColumn); 416 relationship.setCallees(Arrays.asList(sourceColumn)); 417 relationship.setBuiltIn(pair.third.getBuiltIn()); 418 relations.add(relationship); 419 } 420 } else { 421 for (Pair<String, String> pair : relationSet) { 422 if (dbObjMap.get(pair.first) == null || dbObjMap.get(pair.second) == null) { 423 continue; 424 } 425 relationship relationship = new relationship(); 426 relationship.setType(type.name()); 427 relationship.setId(String.valueOf(id.incrementAndGet())); 428 targetColumn targetColumn = new targetColumn(); 429 targetColumn.setId(dbObjMap.get(pair.first).getColumns().get(0).getId()); 430 targetColumn.setColumn(dbObjMap.get(pair.first).getColumns().get(0).getName()); 431 targetColumn.setParent_id(pair.first); 432 targetColumn.setParent_name(dbObjMap.get(pair.first).getName()); 433 sourceColumn sourceColumn = new sourceColumn(); 434 sourceColumn.setId(dbObjMap.get(pair.second).getColumns().get(0).getId()); 435 sourceColumn.setColumn(dbObjMap.get(pair.second).getColumns().get(0).getName()); 436 sourceColumn.setParent_id(pair.second); 437 sourceColumn.setParent_name(dbObjMap.get(pair.second).getName()); 438 relationship.setTarget(targetColumn); 439 relationship.setSources(Arrays.asList(sourceColumn)); 440 relations.add(relationship); 441 } 442 } 443 } 444 instance.setRelationships(relations); 445 CommentBindingProjection.retain(instance); 446 return instance; 447 } 448 449 public static dataflow convertToSchemaLevelDataflow(dataflow dataflow, EDbVendor dbVendor) throws Exception { 450 return convertToSchemaLevelDataflow(dataflow, dbVendor, false); 451 } 452 453 public static dataflow convertToSchemaLevelDataflow(dataflow dataflow, EDbVendor dbVendor, boolean isSimple) throws Exception { 454 dataflow instance = cloneDataflow(dataflow); 455 456 if (!isSimple) { 457 DataFlowAnalyzer analyzer = new DataFlowAnalyzer("", dbVendor, true); 458 instance = analyzer.getSimpleDataflow(instance, true, Arrays.asList(new String[]{"fdd", "fdr"})); 459 } 460 461 List<table> allTables = new ArrayList<table>(); 462 allTables.addAll(instance.getTables()); 463 allTables.addAll(instance.getViews()); 464 465 ModelBindingManager.setGlobalVendor(dbVendor); 466 467 Map<String, String> tableIdSchameNameMap = allTables.stream().collect(Collectors.toMap(table -> table.getId(), table -> table.getFullSchemaName(), (existingValue, newValue) -> newValue)); 468 469 if (instance.getRelationships() == null) { 470 return instance; 471 } 472 473 Map<String, LinkedHashSet<Pair<String, String>>> relationMap = new HashMap<String, LinkedHashSet<Pair<String, String>>>(); 474 for (RelationshipType type : RelationshipType.values()) { 475 relationMap.put(type.name(), new LinkedHashSet<Pair<String, String>>()); 476 } 477 for (relationship relationship : instance.getRelationships()) { 478 String targetId = relationship.getTarget().getParent_id(); 479 for (sourceColumn sourceColumn : relationship.getSources()) { 480 String sourceId = sourceColumn.getParent_id(); 481 relationMap.get(relationship.getType()).add(new Pair<>(tableIdSchameNameMap.get(targetId), tableIdSchameNameMap.get(sourceId))); 482 } 483 } 484 485 Map<String, table> dbObjMap = getDataflowDbObjMap(instance); 486 LinkedHashSet<String> schemaNameSet = new LinkedHashSet<String>(); 487 schemaNameSet.addAll(tableIdSchameNameMap.values()); 488 List<table> schemaTables = new ArrayList<table>(); 489 490 AtomicLong id = new AtomicLong(0); 491 492 for (String schemaName : schemaNameSet) { 493 table table = new table(); 494 table.setId(String.valueOf(id.incrementAndGet())); 495 table.setName(schemaName); 496 table.setType("schema"); 497 schemaTables.add(table); 498 dbObjMap.put(schemaName, table); 499 } 500 501 for (table table : schemaTables) { 502 column column = new column(); 503 column.setId(String.valueOf(id.incrementAndGet())); 504 column.setName(table.getType()); 505 table.setColumns(Arrays.asList(column)); 506 } 507 508 List<relationship> relations = new ArrayList<relationship>(); 509 for (RelationshipType type : RelationshipType.values()) { 510 LinkedHashSet<Pair<String, String>> relationSet = relationMap.get(type.name()); 511 512 for (Pair<String, String> pair : relationSet) { 513 if (dbObjMap.get(pair.first) == null || dbObjMap.get(pair.second) == null) { 514 continue; 515 } 516 relationship relationship = new relationship(); 517 relationship.setType(type.name()); 518 relationship.setId(String.valueOf(id.incrementAndGet())); 519 targetColumn targetColumn = new targetColumn(); 520 targetColumn.setId(dbObjMap.get(pair.first).getColumns().get(0).getId()); 521 targetColumn.setColumn(dbObjMap.get(pair.first).getColumns().get(0).getName()); 522 targetColumn.setParent_id(dbObjMap.get(pair.first).getId()); 523 targetColumn.setParent_name(dbObjMap.get(pair.first).getName()); 524 sourceColumn sourceColumn = new sourceColumn(); 525 sourceColumn.setId(dbObjMap.get(pair.second).getColumns().get(0).getId()); 526 sourceColumn.setColumn(dbObjMap.get(pair.second).getColumns().get(0).getName()); 527 sourceColumn.setParent_id(dbObjMap.get(pair.second).getId()); 528 sourceColumn.setParent_name(dbObjMap.get(pair.second).getName()); 529 relationship.setTarget(targetColumn); 530 relationship.setSources(Arrays.asList(sourceColumn)); 531 relations.add(relationship); 532 } 533 } 534 535 dataflow schemaDataflow = new dataflow(); 536 schemaDataflow.setTables(schemaTables); 537 schemaDataflow.setRelationships(relations); 538 schemaDataflow.setCommentBindings(dataflow.getCommentBindings()); 539 CommentBindingProjection.retain(schemaDataflow); 540 return schemaDataflow; 541 } 542 543 private static final int PARALLEL_CLONE_THRESHOLD = 1000; 544 545 public static dataflow cloneDataflow(dataflow dataflow) { 546 dataflow dataflowCopy = new dataflow(); 547 dataflowCopy.setCommentBindings(dataflow.getCommentBindings()); 548 if (dataflow.getTables() != null) { 549 dataflowCopy.setTables(cloneTables(dataflow.getTables())); 550 } 551 if (dataflow.getViews() != null) { 552 dataflowCopy.setViews(cloneTables(dataflow.getViews())); 553 } 554 if (dataflow.getRelationships() != null) { 555 dataflowCopy.setRelationships(new ArrayList<>(dataflow.getRelationships())); 556 } 557 if (dataflow.getErrors() != null) { 558 dataflowCopy.setErrors(new ArrayList<>(dataflow.getErrors())); 559 } 560 if (dataflow.getPaths() != null) { 561 dataflowCopy.setPaths(cloneTables(dataflow.getPaths())); 562 } 563 if (dataflow.getPackages() != null) { 564 dataflowCopy.setPackages(new ArrayList<>(dataflow.getPackages())); 565 } 566 if (dataflow.getProcedures() != null) { 567 dataflowCopy.setProcedures(new ArrayList<>(dataflow.getProcedures())); 568 } 569 if (dataflow.getProcesses() != null) { 570 dataflowCopy.setProcesses(new ArrayList<>(dataflow.getProcesses())); 571 } 572 if (dataflow.getResultsets() != null) { 573 dataflowCopy.setResultsets(cloneTables(dataflow.getResultsets())); 574 } 575 if (dataflow.getVariables() != null) { 576 dataflowCopy.setVariables(cloneTables(dataflow.getVariables())); 577 } 578 if (dataflow.getStages() != null) { 579 dataflowCopy.setStages(cloneTables(dataflow.getStages())); 580 } 581 if (dataflow.getSequences() != null) { 582 dataflowCopy.setSequences(cloneTables(dataflow.getSequences())); 583 } 584 if (dataflow.getDatasources() != null) { 585 dataflowCopy.setDatasources(cloneTables(dataflow.getDatasources())); 586 } 587 if (dataflow.getDatabases() != null) { 588 dataflowCopy.setDatabases(cloneTables(dataflow.getDatabases())); 589 } 590 if (dataflow.getSchemas() != null) { 591 dataflowCopy.setSchemas(cloneTables(dataflow.getSchemas())); 592 } 593 if (dataflow.getStreams() != null) { 594 dataflowCopy.setStreams(cloneTables(dataflow.getStreams())); 595 } 596 dataflowCopy.setOrientation(dataflow.getOrientation()); 597 return dataflowCopy; 598 } 599 600 private static List<table> cloneTables(List<table> tableList) { 601 if (tableList == null) { 602 return null; 603 } 604 int size = tableList.size(); 605 if (size >= PARALLEL_CLONE_THRESHOLD) { 606 return tableList.parallelStream() 607 .map(DataflowUtility::cloneTable) 608 .collect(Collectors.toCollection(() -> new ArrayList<>(size))); 609 } else { 610 List<table> tables = new ArrayList<>(size); 611 for (int i = 0; i < size; i++) { 612 tables.add(cloneTable(tableList.get(i))); 613 } 614 return tables; 615 } 616 } 617 618 private static table cloneTable(table item) { 619 try { 620 table cloned = (table) item.clone(); 621 List<column> columns = item.getColumns(); 622 if (columns != null) { 623 cloned.setColumns(new ArrayList<>(columns)); 624 } 625 return cloned; 626 } catch (CloneNotSupportedException e) { 627 throw new RuntimeException("Clone table failed.", e); 628 } 629 } 630 631 public static Map<String, table> getDataflowDbObjMap(dataflow dataflow) { 632 List<table> tables = new ArrayList<table>(); 633 if (dataflow.getTables() != null) { 634 tables.addAll(dataflow.getTables()); 635 } 636 if (dataflow.getViews() != null) { 637 tables.addAll(dataflow.getViews()); 638 } 639 if (dataflow.getPaths() != null) { 640 tables.addAll(dataflow.getPaths()); 641 } 642 if (dataflow.getResultsets() != null) { 643 tables.addAll(dataflow.getResultsets()); 644 } 645 if (dataflow.getVariables() != null) { 646 tables.addAll(dataflow.getVariables()); 647 } 648 if (dataflow.getStages() != null) { 649 tables.addAll(dataflow.getStages()); 650 } 651 if (dataflow.getSequences() != null) { 652 tables.addAll(dataflow.getSequences()); 653 } 654 if (dataflow.getDatasources() != null) { 655 tables.addAll(dataflow.getDatasources()); 656 } 657 if (dataflow.getDatabases() != null) { 658 tables.addAll(dataflow.getDatabases()); 659 } 660 if (dataflow.getSchemas() != null) { 661 tables.addAll(dataflow.getSchemas()); 662 } 663 if (dataflow.getStreams() != null) { 664 tables.addAll(dataflow.getStreams()); 665 } 666 667 Map<String, table> dbObjMap = new HashMap<>(); 668 for (table table : tables) { 669 dbObjMap.put(table.getId(), table); 670 } 671 return dbObjMap; 672 } 673 public static Map<String, table> getDataflowDbObjNameMap(dataflow dataflow) { 674 List<table> tables = new ArrayList<table>(); 675 if (dataflow.getTables() != null) { 676 tables.addAll(dataflow.getTables()); 677 } 678 if (dataflow.getViews() != null) { 679 tables.addAll(dataflow.getViews()); 680 } 681 if (dataflow.getPaths() != null) { 682 tables.addAll(dataflow.getPaths()); 683 } 684 if (dataflow.getResultsets() != null) { 685 tables.addAll(dataflow.getResultsets()); 686 } 687 if (dataflow.getVariables() != null) { 688 tables.addAll(dataflow.getVariables()); 689 } 690 if (dataflow.getStages() != null) { 691 tables.addAll(dataflow.getStages()); 692 } 693 if (dataflow.getSequences() != null) { 694 tables.addAll(dataflow.getSequences()); 695 } 696 if (dataflow.getDatasources() != null) { 697 tables.addAll(dataflow.getDatasources()); 698 } 699 if (dataflow.getDatabases() != null) { 700 tables.addAll(dataflow.getDatabases()); 701 } 702 if (dataflow.getSchemas() != null) { 703 tables.addAll(dataflow.getSchemas()); 704 } 705 if (dataflow.getStreams() != null) { 706 tables.addAll(dataflow.getStreams()); 707 } 708 709 Map<String, table> dbObjMap = new HashMap<>(); 710 for(table table: tables){ 711 dbObjMap.put(table.getFullName(), table); 712 } 713 return dbObjMap; 714 } 715 public static dataflow mergeDataflowsWithDifferentStartId(Collection<dataflow> dataflows, EDbVendor vendor) { 716 return mergeDataflowsWithDifferentStartId(dataflows, vendor, null); 717 } 718 719 /** 720 * As {@link #mergeDataflowsWithDifferentStartId(Collection, EDbVendor)}, under an explicit 721 * identifier configuration. 722 * 723 * @param dataflows results whose ID spaces already do not overlap 724 * @param vendor dialect of the results 725 * @param identifierProfile the configuration the results were produced under; null means the 726 * vendor default 727 * @return the merged result 728 */ 729 public static dataflow mergeDataflowsWithDifferentStartId(Collection<dataflow> dataflows, 730 EDbVendor vendor, IdentifierProfile identifierProfile) { 731 IdentifierProfile effective = requireOneMergePolicy(dataflows, vendor, identifierProfile, 732 "DataflowUtility.mergeDataflowsWithDifferentStartId"); 733 IdentifierService previous = ModelBindingManager.getGlobalIdentifierServiceRaw(); 734 ModelBindingManager.setGlobalIdentifierService(IdentifierService.forProfile(effective)); 735 ModelBindingManager.setGlobalVendor(vendor); 736 try { 737 dataflow merged = mergeDataflowsByStartId(dataflows, DATAFLOW_ID_RANGE * dataflows.size()); 738 if (merged != null) { 739 merged.setIdentifierPolicy(effective); 740 } 741 return merged; 742 } finally { 743 ModelBindingManager.removeGlobalVendor(); 744 ModelBindingManager.setGlobalIdentifierService(previous); 745 } 746 } 747 748 /** 749 * 内部合并实现:调用方需保证所有输入 dataflow 的 ID 空间不冲突,并且已提前设置好 ModelBindingManager。 750 * <p> 751 * 该方法代替之前位于 {@code ParallelDataFlowAnalyzer} 的同名实现,外部调用者应使用 752 * {@link #mergeDataflowsWithDifferentStartId(Collection, EDbVendor)} 或 753 * {@link #mergeDataflows(Collection, EDbVendor)}。 754 */ 755 public static dataflow mergeDataflowsByStartId(Collection<dataflow> dataflows, long startId) { 756 dataflow mergeDataflow = new dataflow(); 757 List<CommentBindingBundle> commentBundles = null; 758 759 760 List<table> tableCopy = new ArrayList<table>(); 761 List<table> viewCopy = new ArrayList<table>(); 762 List<table> databaseCopy = new ArrayList<table>(); 763 List<table> schemaCopy = new ArrayList<table>(); 764 List<table> stageCopy = new ArrayList<table>(); 765 List<table> dataSourceCopy = new ArrayList<table>(); 766 List<table> streamCopy = new ArrayList<table>(); 767 List<table> fileCopy = new ArrayList<table>(); 768 List<table> variableCopy = new ArrayList<table>(); 769 List<table> resultSetCopy = new ArrayList<table>(); 770 List<table> sequenceCopy = new ArrayList<table>(); 771 772 List<process> processCopy = new ArrayList<process>(); 773 List<relationship> relationshipCopy = new ArrayList<relationship>(); 774 List<procedure> procedureCopy = new ArrayList<procedure>(); 775 List<oraclePackage> packageCopy = new ArrayList<oraclePackage>(); 776 List<error> errorCopy = new ArrayList<error>(); 777 778 List<table> tables = new ArrayList<table>(); 779 Set<table> physicalTables = Collections.newSetFromMap(new IdentityHashMap<table, Boolean>()); 780 // which merged dataflow each relationship comes from (sameEdgeOfOneInput) 781 Map<relationship, Integer> inputOf = new IdentityHashMap<relationship, Integer>(); 782 int dfIdx = 0; 783 for (dataflow dataflow : dataflows) { 784 if (dataflow != null && dataflow.getCommentBindings() != null) { 785 if (commentBundles == null) commentBundles = new ArrayList<CommentBindingBundle>(); 786 commentBundles.add(dataflow.getCommentBindings()); 787 } 788 789 if (dataflow == null) { 790 continue; 791 } 792 793 if (dataflow.getTables() != null) { 794 tableCopy.addAll(dataflow.getTables()); 795 } 796 mergeDataflow.setTables(tableCopy); 797 if (dataflow.getViews() != null) { 798 viewCopy.addAll(dataflow.getViews()); 799 } 800 mergeDataflow.setViews(viewCopy); 801 if (dataflow.getDatabases() != null) { 802 databaseCopy.addAll(dataflow.getDatabases()); 803 } 804 mergeDataflow.setDatabases(databaseCopy); 805 if (dataflow.getSchemas() != null) { 806 schemaCopy.addAll(dataflow.getSchemas()); 807 } 808 mergeDataflow.setSchemas(schemaCopy); 809 if (dataflow.getStages() != null) { 810 stageCopy.addAll(dataflow.getStages()); 811 } 812 mergeDataflow.setStages(stageCopy); 813 if (dataflow.getDatasources() != null) { 814 dataSourceCopy.addAll(dataflow.getDatasources()); 815 } 816 mergeDataflow.setDatasources(dataSourceCopy); 817 if (dataflow.getStreams() != null) { 818 streamCopy.addAll(dataflow.getStreams()); 819 } 820 mergeDataflow.setStreams(streamCopy); 821 if (dataflow.getPaths() != null) { 822 fileCopy.addAll(dataflow.getPaths()); 823 } 824 mergeDataflow.setPaths(fileCopy); 825 if (dataflow.getVariables() != null) { 826 variableCopy.addAll(dataflow.getVariables()); 827 } 828 mergeDataflow.setVariables(variableCopy); 829 if (dataflow.getResultsets() != null) { 830 resultSetCopy.addAll(dataflow.getResultsets()); 831 } 832 mergeDataflow.setResultsets(resultSetCopy); 833 if (dataflow.getSequences() != null) { 834 sequenceCopy.addAll(dataflow.getSequences()); 835 } 836 mergeDataflow.setSequences(sequenceCopy); 837 838 dfIdx++; 839 840 if (dataflow.getProcesses() != null) { 841 processCopy.addAll(dataflow.getProcesses()); 842 } 843 mergeDataflow.setProcesses(processCopy); 844 if (dataflow.getRelationships() != null) { 845 relationshipCopy.addAll(dataflow.getRelationships()); 846 for (relationship r : dataflow.getRelationships()) { 847 inputOf.put(r, dfIdx); 848 } 849 } 850 mergeDataflow.setRelationships(relationshipCopy); 851 if (dataflow.getProcedures() != null) { 852 procedureCopy.addAll(dataflow.getProcedures()); 853 } 854 mergeDataflow.setProcedures(procedureCopy); 855 if (dataflow.getPackages() != null) { 856 packageCopy.addAll(dataflow.getPackages()); 857 } 858 mergeDataflow.setPackages(packageCopy); 859 if (dataflow.getErrors() != null) { 860 errorCopy.addAll(dataflow.getErrors()); 861 } 862 if (errorCopy.size() > 10000) { 863 errorCopy = errorCopy.subList(0, 10000); 864 } 865 mergeDataflow.setErrors(errorCopy); 866 867 tables.addAll(dataflow.getTables()); 868 tables.addAll(dataflow.getViews()); 869 physicalTables.addAll(dataflow.getTables()); 870 physicalTables.addAll(dataflow.getViews()); 871 tables.addAll(dataflow.getDatabases()); 872 tables.addAll(dataflow.getSchemas()); 873 tables.addAll(dataflow.getStages()); 874 tables.addAll(dataflow.getDatasources()); 875 tables.addAll(dataflow.getStreams()); 876 tables.addAll(dataflow.getPaths()); 877 tables.addAll(dataflow.getResultsets()); 878 tables.addAll(dataflow.getVariables()); 879 if (dataflow.getSequences() != null) { 880 tables.addAll(dataflow.getSequences()); 881 } 882 } 883 884 Map<String, List<table>> tableMap = new HashMap<String, List<table>>(); 885 Map<String, String> tableTypeMap = new HashMap<String, String>(); 886 Map<String, String> tableIdMap = new HashMap<String, String>(); 887 888 Map<String, List<column>> columnMap = new HashMap<String, List<column>>(); 889 Map<String, Set<String>> tableColumnMap = new HashMap<String, Set<String>>(); 890 Map<String, String> columnIdMap = new HashMap<String, String>(); 891 Map<String, column> columnMergeIdMap = new HashMap<String, column>(); 892 // columns of one table renamed apart because their names collide across identities, by column id 893 Map<String, String> renamedColumns = new HashMap<String, String>(); 894 895 List<procedure> procedures = new ArrayList<>(mergeDataflow.getProcedures()); 896 if (mergeDataflow.getPackages() != null) { 897 for (oraclePackage pkg : mergeDataflow.getPackages()) { 898 procedures.addAll(pkg.getProcedures()); 899 } 900 } 901 902 Set<String> procedureIdSet = procedures.stream().map(t -> t.getId()).collect(Collectors.toSet()); 903 904 PhysicalTableMergeKeys mergeKeys = physicalTableMergeKeys(tables, physicalTables); 905 for (table table : tables) { 906 String tableFullName = getTableFullNameKey(table, physicalTables, mergeKeys); 907 908 if (!tableMap.containsKey(tableFullName)) { 909 tableMap.put(tableFullName, new ArrayList<table>()); 910 } 911 912 tableMap.get(tableFullName).add(table); 913 914 if (!tableTypeMap.containsKey(tableFullName)) { 915 tableTypeMap.put(tableFullName, table.getType()); 916 } else if ("view".equals(table.getSubType())) { 917 tableTypeMap.put(tableFullName, table.getType()); 918 } else if ("database".equals(table.getSubType())) { 919 tableTypeMap.put(tableFullName, table.getType()); 920 } else if ("schema".equals(table.getSubType())) { 921 tableTypeMap.put(tableFullName, table.getType()); 922 } else if ("stage".equals(table.getSubType())) { 923 tableTypeMap.put(tableFullName, table.getType()); 924 } else if ("datasource".equals(table.getSubType())) { 925 tableTypeMap.put(tableFullName, table.getType()); 926 } else if ("stream".equals(table.getSubType())) { 927 tableTypeMap.put(tableFullName, table.getType()); 928 } else if ("file".equals(table.getSubType())) { 929 tableTypeMap.put(tableFullName, table.getType()); 930 } else if ("sequence".equals(table.getSubType())) { 931 tableTypeMap.put(tableFullName, table.getType()); 932 } else if ("table".equals(tableTypeMap.get(tableFullName))) { 933 tableTypeMap.put(tableFullName, table.getType()); 934 } else if ("variable".equals(tableTypeMap.get(tableFullName))) { 935 tableTypeMap.put(tableFullName, table.getType()); 936 } 937 938 if (table.getColumns() != null) { 939 if (!tableColumnMap.containsKey(tableFullName)) { 940 tableColumnMap.put(tableFullName, new LinkedHashSet<String>()); 941 } 942 for (column column : table.getColumns()) { 943 String columnFullName = tableFullName + "." 944 + LegacyDlineageWireNamesV1.columnBucket(column.getName()); 945 946 if (!columnMap.containsKey(columnFullName)) { 947 columnMap.put(columnFullName, new ArrayList<column>()); 948 tableColumnMap.get(tableFullName).add(columnFullName); 949 } 950 951 columnMap.get(columnFullName).add(column); 952 } 953 } 954 } 955 956 Iterator<String> tableNameIter = tableMap.keySet().iterator(); 957 while (tableNameIter.hasNext()) { 958 String tableName = tableNameIter.next(); 959 List<table> tableList = tableMap.get(tableName); 960 table table; 961 if (tableList.size() > 1) { 962 GroupNames chosen = groupNames(tableList); 963 table standardTable = chosen.displayed; 964 //Function允许重名,不做合并处理 965 //检查列表中任意一个表是否为function,防止function表被合并丢失 966 boolean hasFunction = false; 967 for (table t : tableList) { 968 if (t.isFunction()) { 969 hasFunction = true; 970 break; 971 } 972 } 973 if (hasFunction) { 974 continue; 975 } 976 977 String type = tableTypeMap.get(tableName); 978 table = new table(); 979 table.setId(String.valueOf(++startId)); 980 table names = chosen.names; 981 carryGroupIdentity(table, tableList, chosen); 982 table.setServer(names.getServer()); 983 table.setDatabase(names.getDatabase()); 984 table.setSchema(names.getSchema()); 985 table.setName(names.getName()); 986 table.setDisplayName(names.getDisplayName()); 987 table.setParent(standardTable.getParent()); 988 table.setColumns(new ArrayList<column>()); 989 String subType = null; 990 for(table item: tableList){ 991 if (item.getSubType() != null) { 992 subType = item.getSubType(); 993 break; 994 } 995 } 996 if (subType != null) { 997 table.setSubType(subType); 998 } else { 999 table.setSubType(standardTable.getSubType()); 1000 } 1001 mergeEndpointClassification(table, tableList); 1002 Set<String> processIds = new LinkedHashSet<String>(); 1003 for (int k = 0; k < tableList.size(); k++) { 1004 if (tableList.get(k).getProcessIds() != null) { 1005 processIds.addAll(tableList.get(k).getProcessIds()); 1006 } 1007 } 1008 if (!processIds.isEmpty()) { 1009 table.setProcessIds(new ArrayList<String>(processIds)); 1010 } 1011 table.setType(type); 1012 for (table item : tableList) { 1013 if (!SQLUtil.isEmpty(table.getCoordinate()) && !SQLUtil.isEmpty(item.getCoordinate())) { 1014 if (table.getCoordinate().indexOf(item.getCoordinate()) == -1) { 1015 table.appendCoordinate(item.getCoordinate()); 1016 } 1017 } else if (!SQLUtil.isEmpty(item.getCoordinate())) { 1018 table.setCoordinate(item.getCoordinate()); 1019 } 1020 1021 if (!SQLUtil.isEmpty(table.getAlias()) && !SQLUtil.isEmpty(item.getAlias())) { 1022 table.setAlias(table.getAlias() + "," + item.getAlias()); 1023 } else if (!SQLUtil.isEmpty(item.getAlias())) { 1024 table.setAlias(item.getAlias()); 1025 } 1026 1027 tableIdMap.put(item.getId(), table.getId()); 1028 1029 if (item.isView()) { 1030 mergeDataflow.getViews().remove(item); 1031 } else if (item.isDatabaseType()) { 1032 mergeDataflow.getDatabases().remove(item); 1033 } else if (item.isSchemaType()) { 1034 mergeDataflow.getSchemas().remove(item); 1035 } else if (item.isStage()) { 1036 mergeDataflow.getStages().remove(item); 1037 } else if (item.isDataSource()) { 1038 mergeDataflow.getDatasources().remove(item); 1039 } else if (item.isStream()) { 1040 mergeDataflow.getStreams().remove(item); 1041 } else if (item.isFile()) { 1042 mergeDataflow.getPaths().remove(item); 1043 } else if (item.isVariable()) { 1044 mergeDataflow.getVariables().remove(item); 1045 } else if (item.isTable()) { 1046 mergeDataflow.getTables().remove(item); 1047 } else if (item.isResultSet()) { 1048 mergeDataflow.getResultsets().remove(item); 1049 } else if (item.isSequence()) { 1050 mergeDataflow.getSequences().remove(item); 1051 } 1052 } 1053 1054 if (table.isView()) { 1055 mergeDataflow.getViews().add(table); 1056 } else if (table.isDatabaseType()) { 1057 mergeDataflow.getDatabases().add(table); 1058 } else if (table.isSchemaType()) { 1059 mergeDataflow.getSchemas().add(table); 1060 } else if (table.isStage()) { 1061 mergeDataflow.getStages().add(table); 1062 } else if (table.isDataSource()) { 1063 mergeDataflow.getDatasources().add(table); 1064 } else if (table.isStream()) { 1065 mergeDataflow.getStreams().add(table); 1066 } else if (table.isFile()) { 1067 mergeDataflow.getPaths().add(table); 1068 } else if (table.isVariable()) { 1069 mergeDataflow.getVariables().add(table); 1070 } else if (table.isResultSet()) { 1071 mergeDataflow.getResultsets().add(table); 1072 } else if (table.isSequence()) { 1073 mergeDataflow.getSequences().add(table); 1074 } else { 1075 mergeDataflow.getTables().add(table); 1076 } 1077 } else { 1078 table = tableList.get(0); 1079 } 1080 1081 List<column> mergeColumns = new ArrayList<column>(); 1082 if (tableList.size() == 1 && table.isResultSet()) { 1083 // A result set one analysis produced keeps its columns as that analysis published them (a worker's 1084 // result sets are named by its ids, so another result never holds them). Merging them again by 1085 // folded name dropped columns it kept apart: a.ID beside b.ID, or two select items of one alias 1086 // (the second published as ID(1)). 1087 if (table.getColumns() == null) { 1088 continue; 1089 } 1090 for (column item : table.getColumns()) { 1091 mergeColumns.add(item); 1092 columnIdMap.put(item.getId(), item.getId()); 1093 columnMergeIdMap.put(item.getId(), item); 1094 } 1095 table.setColumns(mergeColumns); 1096 continue; 1097 } 1098 List<List<column>> columnsByFoldedName = new ArrayList<List<column>>(); 1099 for (String columnName : tableColumnMap.get(tableName)) { 1100 columnsByFoldedName.add(columnMap.get(columnName)); 1101 } 1102 for (Object entry : columnsOfTable(columnsByFoldedName)) { 1103 if (entry instanceof column) { 1104 column function = (column) entry; 1105 mergeColumns.add(function); 1106 columnIdMap.put(function.getId(), function.getId()); 1107 columnMergeIdMap.put(function.getId(), function); 1108 continue; 1109 } 1110 @SuppressWarnings("unchecked") 1111 List<column> columnList = (List<column>) entry; 1112 column firstColumn = columnList.iterator().next(); 1113 if (columnList.size() > 1) { 1114 column mergeColumn = new column(); 1115 mergeColumn.setId(String.valueOf(++startId)); 1116 mergeColumn.setName(firstColumn.getName()); 1117 mergeColumn.setStar(firstColumn.getStar()); 1118 mergeColumn.setDisplayName(firstColumn.getDisplayName()); 1119 mergeColumn.setSource(firstColumn.getSource()); 1120 mergeColumn.setQualifiedTable(firstColumn.getQualifiedTable()); 1121 mergeColumn.setIdentifierKey(identityOf(columnList)); 1122 mergeColumn.setSqlSpelling(sqlSpellingOf(columnList)); 1123 mergeColumns.add(mergeColumn); 1124 for (column item : columnList) { 1125 mergeColumn.appendCoordinate(item.getCoordinate()); 1126 columnIdMap.put(item.getId(), mergeColumn.getId()); 1127 takeAttributes(mergeColumn, item); 1128 } 1129 columnMergeIdMap.put(mergeColumn.getId(), mergeColumn); 1130 } else { 1131 mergeColumns.add(firstColumn); 1132 columnIdMap.put(firstColumn.getId(), firstColumn.getId()); 1133 columnMergeIdMap.put(firstColumn.getId(), firstColumn); 1134 } 1135 } 1136 table.setColumns(mergeColumns); 1137 renamedColumns.putAll(nameCollidingIdentities(mergeColumns, ModelBindingManager.getGlobalVendor())); 1138 } 1139 1140 // the result set of each routine, by the routine's identity (routineKey) 1141 Map<Object, String> procedureNameToResultSetId = new HashMap<>(); 1142 for (dataflow df : dataflows) { 1143 if (df == null || df.getProcedures() == null) { 1144 continue; 1145 } 1146 1147 Map<String, String> rsIdToProcId = new HashMap<>(); 1148 if (df.getResultsets() != null) { 1149 for (table t : df.getResultsets()) { 1150 if (!t.isFunction() && t.getProcedureId() != null) { 1151 rsIdToProcId.put(t.getId(), t.getProcedureId()); 1152 } 1153 } 1154 } 1155 if (df.getTables() != null) { 1156 for (table t : df.getTables()) { 1157 if (t.getProcedureId() != null) { 1158 rsIdToProcId.put(t.getId(), t.getProcedureId()); 1159 } 1160 } 1161 } 1162 if (df.getVariables() != null) { 1163 for (table t : df.getVariables()) { 1164 if (t.getProcedureId() != null) { 1165 rsIdToProcId.put(t.getId(), t.getProcedureId()); 1166 } 1167 } 1168 } 1169 if (rsIdToProcId.isEmpty()) { 1170 continue; 1171 } 1172 1173 Map<String, Object> procIdToName = new HashMap<>(); 1174 for (procedure proc : df.getProcedures()) { 1175 procIdToName.put(proc.getId(), routineKeyOf(proc)); 1176 } 1177 1178 for (Map.Entry<String, String> entry : rsIdToProcId.entrySet()) { 1179 Object procName = procIdToName.get(entry.getValue()); 1180 if (procName != null) { 1181 String mergedRsId = tableIdMap.get(entry.getKey()); 1182 String resultSetId = mergedRsId != null ? mergedRsId : entry.getKey(); 1183 procedureNameToResultSetId.put(procName, resultSetId); 1184 } 1185 } 1186 } 1187 1188 List<table> allTables = dataflow.getAllTables(mergeDataflow); 1189 Map<String, table> allTableIdMap = new HashMap<>(allTables.size() * 4 / 3 + 1); 1190 for (table t : allTables) { 1191 allTableIdMap.put(t.getId(), t); 1192 } 1193 1194 Map<String, String> functionIdToResultSetId = new HashMap<>(); 1195 Map<String, column> funcColIdToResultSetCol = new HashMap<>(); 1196 for (table t : allTableIdMap.values()) { 1197 if (t.isFunction()) { 1198 String resultSetId = procedureNameToResultSetId.get(calledRoutineKey(t)); 1199 if (resultSetId != null) { 1200 functionIdToResultSetId.put(t.getId(), resultSetId); 1201 table resultSet = allTableIdMap.get(resultSetId); 1202 if (resultSet != null && t.getColumns() != null && resultSet.getColumns() != null) { 1203 List<column> funcCols = t.getColumns(); 1204 List<column> rsCols = resultSet.getColumns(); 1205 int minSize = Math.min(funcCols.size(), rsCols.size()); 1206 for (int i = 0; i < minSize; i++) { 1207 funcColIdToResultSetCol.put(funcCols.get(i).getId(), rsCols.get(i)); 1208 } 1209 } 1210 } 1211 } 1212 } 1213 1214 if (commentBundles != null) mergeDataflow.setCommentBindings(CommentBindingProjection.merge(commentBundles)); 1215 Map<String,String> commentRelationIds = commentBundles == null ? null : new HashMap<String,String>(); 1216 if (mergeDataflow.getRelationships() != null) { 1217 Map<String, relationship> mergeRelations = new LinkedHashMap<String, relationship>(); 1218 for (int i = 0; i < mergeDataflow.getRelationships().size(); i++) { 1219 relationship relation = mergeDataflow.getRelationships().get(i); 1220 if (RelationshipType.call.name().equals(relation.getType())) { 1221 targetColumn target = relation.getCaller(); 1222 List<sourceColumn> sources = relation.getCallees(); 1223 if (target == null) { 1224 continue; 1225 } 1226 String mappedTargetId = tableIdMap.get(target.getId()); 1227 if (mappedTargetId != null) { 1228 target.setId(mappedTargetId); 1229 } 1230 1231 Set<sourceColumn> sourceSet = new LinkedHashSet<sourceColumn>(); 1232 if (sources != null) { 1233 for (sourceColumn source : sources) { 1234 String mappedSourceId = tableIdMap.get(source.getId()); 1235 if (mappedSourceId != null) { 1236 source.setId(mappedSourceId); 1237 } 1238 } 1239 sourceSet.addAll(sources); 1240 relation.setCallees(new ArrayList<sourceColumn>(sourceSet)); 1241 } 1242 1243 String key = relation.toDedupKey(); 1244 mergeRelations.putIfAbsent(key, relation); 1245 if (commentRelationIds != null) commentRelationIds.put(relation.getId(), mergeRelations.get(key).getId()); 1246 } else { 1247 targetColumn target = relation.getTarget(); 1248 if (target == null) { 1249 continue; 1250 } 1251 String mappedTargetParentId = tableIdMap.get(target.getParent_id()); 1252 if (mappedTargetParentId != null) { 1253 target.setParent_id(mappedTargetParentId); 1254 } 1255 1256 String mappedTargetColId = columnIdMap.get(target.getId()); 1257 if (mappedTargetColId != null) { 1258 target.setId(mappedTargetColId); 1259 column mergedCol = columnMergeIdMap.get(mappedTargetColId); 1260 if (mergedCol != null) { 1261 target.setCoordinate(mergedCol.getCoordinate()); 1262 } 1263 } 1264 1265 List<sourceColumn> sources = relation.getSources(); 1266 Set<sourceColumn> sourceSet = new LinkedHashSet<sourceColumn>(); 1267 // Same contract as DataFlowAnalyzer.mergeTables: dedup runs 1268 // under the merged column's stamped coordinate, but a source's 1269 // own usable reference-site coordinate is restored on the 1270 // surviving relation. 1271 Map<sourceColumn, String> referenceCoordinates = new IdentityHashMap<sourceColumn, String>(); 1272 if (sources != null) { 1273 for (sourceColumn source : sources) { 1274 String mappedParentId = tableIdMap.get(source.getParent_id()); 1275 if (mappedParentId != null) { 1276 source.setParent_id(mappedParentId); 1277 } 1278 String mappedSourceId = tableIdMap.get(source.getSource_id()); 1279 if (mappedSourceId != null) { 1280 source.setSource_id(mappedSourceId); 1281 } 1282 String originalParentId = source.getParent_id(); 1283 if (!functionIdToResultSetId.isEmpty()) { 1284 String resultSetId = functionIdToResultSetId.get(originalParentId); 1285 if (resultSetId != null) { 1286 source.setParent_id(resultSetId); 1287 column matched = funcColIdToResultSetCol.get(source.getId()); 1288 if (matched != null) { 1289 source.setId(matched.getId()); 1290 source.setColumn(matched.getName()); 1291 } 1292 table resultSetTable = allTableIdMap.get(source.getParent_id()); 1293 if (resultSetTable != null) { 1294 source.setParent_name(resultSetTable.getName()); 1295 } 1296 } 1297 } 1298 String mappedColId = columnIdMap.get(source.getId()); 1299 if (mappedColId != null) { 1300 source.setId(mappedColId); 1301 column mergedCol = columnMergeIdMap.get(mappedColId); 1302 if (mergedCol != null) { 1303 String ownCoordinate = source.getCoordinate(); 1304 source.setCoordinate(mergedCol.getCoordinate()); 1305 if (DlineageUtil.hasUsableCoordinate(ownCoordinate)) { 1306 referenceCoordinates.put(source, ownCoordinate); 1307 } 1308 } 1309 } 1310 } 1311 1312 sourceSet.addAll(sources); 1313 relation.setSources(new ArrayList<sourceColumn>(sourceSet)); 1314 } 1315 1316 String key = relation.toDedupKey(); 1317 relationship first = mergeRelations.get(key); 1318 relationship kept; 1319 if (first == null) { 1320 mergeRelations.put(key, relation); 1321 for (Map.Entry<sourceColumn, String> entry : referenceCoordinates.entrySet()) { 1322 entry.getKey().setCoordinate(entry.getValue()); 1323 } 1324 kept = relation; 1325 } else if (sameEdgeOfOneInput(first, relation, inputOf)) { 1326 // one dataflow published both, from their own reference sites: its analysis keeps them apart, 1327 // so they stay two relationships unless they repeat each other exactly 1328 for (Map.Entry<sourceColumn, String> entry : referenceCoordinates.entrySet()) { 1329 entry.getKey().setCoordinate(entry.getValue()); 1330 } 1331 String exact = exactKey(relation); 1332 if (exact.equals(exactKey(first))) { 1333 kept = first; 1334 } else { 1335 relationship prior = mergeRelations.putIfAbsent(key + '\u0000' + exact, relation); 1336 kept = prior == null ? relation : prior; 1337 } 1338 } else { 1339 kept = first; 1340 } 1341 if (commentRelationIds != null) commentRelationIds.put(relation.getId(), kept.getId()); 1342 } 1343 } 1344 1345 mergeDataflow.setRelationships(new ArrayList<relationship>(mergeRelations.values())); 1346 } 1347 1348 renameEndpoints(mergeDataflow.getRelationships(), renamedColumns); 1349 startId = addStarColumnBridgingRelations(mergeDataflow, startId); 1350 1351 if (commentRelationIds != null) { 1352 CommentBindingProjection.remap(mergeDataflow, tableIdMap, columnIdMap, commentRelationIds); 1353 CommentBindingProjection.retain(mergeDataflow); 1354 } 1355 labelEndpoints(mergeDataflow); 1356 1357 tableMap.clear(); 1358 tableTypeMap.clear(); 1359 tableIdMap.clear(); 1360 columnMap.clear(); 1361 tableColumnMap.clear(); 1362 columnIdMap.clear(); 1363 columnMergeIdMap.clear(); 1364 tables.clear(); 1365 1366 return mergeDataflow; 1367 } 1368 1369 /** 1370 * 合并后同一张表可能同时拥有 * 列(来自只做 SELECT * 的来源)和具名列(来自显式引用列名的来源), 1371 * 但缺少 * -> 具名列 的 fdd 关系,会导致 上游.* -> 表.* 与 表.具名列 -> 下游.具名列 之间链路断开。 1372 * 本方法为这类表补齐 * -> 每个具名列 的 fdd 桥接(已存在则跳过;忽略 system / function 列)。 1373 */ 1374 private static long addStarColumnBridgingRelations(dataflow df, long startId) { 1375 if (df == null) { 1376 return startId; 1377 } 1378 Map<String, Set<String>> bridgeIndex = new HashMap<>(); 1379 if (df.getRelationships() != null) { 1380 for (relationship rel : df.getRelationships()) { 1381 if (!RelationshipType.fdd.name().equals(rel.getType())) { 1382 continue; 1383 } 1384 if (rel.getTarget() == null || rel.getSources() == null) { 1385 continue; 1386 } 1387 String parentId = rel.getTarget().getParent_id(); 1388 if (parentId == null) { 1389 continue; 1390 } 1391 for (sourceColumn sc : rel.getSources()) { 1392 if (parentId.equals(sc.getParent_id())) { 1393 bridgeIndex.computeIfAbsent(parentId, k -> new HashSet<>()) 1394 .add(rel.getTarget().getId() + "<-" + sc.getId()); 1395 } 1396 } 1397 } 1398 } 1399 1400 List<relationship> newRelations = new ArrayList<>(); 1401 for (table t : dataflow.getAllTables(df)) { 1402 if (t.getColumns() == null || t.getColumns().size() < 2) { 1403 continue; 1404 } 1405 column starColumn = null; 1406 List<column> regularColumns = new ArrayList<>(); 1407 for (column c : t.getColumns()) { 1408 if ("system".equals(c.getSource())) { 1409 continue; 1410 } 1411 if (Boolean.TRUE.toString().equals(c.getIsFunction())) { 1412 continue; 1413 } 1414 if (c.standsForStar()) { 1415 if (starColumn == null) { 1416 starColumn = c; 1417 } 1418 } else { 1419 regularColumns.add(c); 1420 } 1421 } 1422 if (starColumn == null || regularColumns.isEmpty()) { 1423 continue; 1424 } 1425 1426 Set<String> existingBridges = bridgeIndex.getOrDefault(t.getId(), Collections.emptySet()); 1427 1428 String starId = starColumn.getId(); 1429 for (column target : regularColumns) { 1430 if (existingBridges.contains(target.getId() + "<-" + starId)) { 1431 continue; 1432 } 1433 relationship rel = new relationship(); 1434 rel.setId(String.valueOf(++startId)); 1435 rel.setType(RelationshipType.fdd.name()); 1436 rel.setEffectType("expand_star"); 1437 1438 targetColumn tc = new targetColumn(); 1439 tc.setId(target.getId()); 1440 tc.setColumn(target.getName()); 1441 tc.setParent_id(t.getId()); 1442 tc.setParent_name(t.getName()); 1443 rel.setTarget(tc); 1444 1445 sourceColumn sc = new sourceColumn(); 1446 sc.setId(starId); 1447 sc.setColumn(starColumn.getName()); 1448 sc.setParent_id(t.getId()); 1449 sc.setParent_name(t.getName()); 1450 rel.setSources(Arrays.asList(sc)); 1451 1452 newRelations.add(rel); 1453 } 1454 } 1455 if (!newRelations.isEmpty()) { 1456 List<relationship> all = df.getRelationships() != null 1457 ? new ArrayList<>(df.getRelationships()) 1458 : new ArrayList<>(); 1459 all.addAll(newRelations); 1460 df.setRelationships(all); 1461 } 1462 return startId; 1463 } 1464 1465 /** 1466 * 单个 dataflow 的 ID 跨度(与 ParallelDataFlowAnalyzer 中的常量保持一致)。 1467 */ 1468 private static final long DATAFLOW_ID_RANGE = 5000000L; 1469 1470 /** 1471 * 合并一组 startId 都为 0(即 ID 空间相互冲突)的 dataflow。 1472 * <p> 1473 * 标准合并 {@link #mergeDataflowsWithDifferentStartId(Collection, EDbVendor)} 的前提是 1474 * 每个 dataflow 在生成时已通过 {@code optionCopy.setStartId(5000000L * i)} 占用了 1475 * 不重叠的 ID 段;当外部调用方拿到的 dataflow 都是用默认 startId=0 生成时,直接合并 1476 * 会因为 ID 冲突而错乱。本方法会先按下标把每个 dataflow 的全部 ID 偏移 1477 * {@link #DATAFLOW_ID_RANGE} * index,再委托给标准合并逻辑。 1478 * <p> 1479 * 注意:此方法会原地修改入参 dataflow 的 ID(与标准合并行为一致)。 1480 */ 1481 public static dataflow mergeDataflows(Collection<dataflow> dataflows, EDbVendor vendor) { 1482 return mergeDataflows(dataflows, vendor, null); 1483 } 1484 1485 /** 1486 * Merge under an explicit identifier configuration — the configuration the analyses that 1487 * produced {@code dataflows} ran under. 1488 * 1489 * <p>The merge compares table and column identities, so it has to apply the same rules the 1490 * results were produced with. The {@code (Collection, EDbVendor)} overload states the vendor 1491 * DEFAULT; pass the profile here when the analyses were configured. 1492 * 1493 * @param dataflows results to merge, in source order 1494 * @param vendor dialect of the results 1495 * @param identifierProfile the configuration the results were produced under; null means the 1496 * vendor default 1497 * @return the merged result, or null for an empty input 1498 * @throws IllegalStateException if an input carries provenance of a different configuration 1499 */ 1500 public static dataflow mergeDataflows(Collection<dataflow> dataflows, EDbVendor vendor, 1501 IdentifierProfile identifierProfile) { 1502 if (dataflows == null || dataflows.isEmpty()) { 1503 return null; 1504 } 1505 IdentifierProfile effective = requireOneMergePolicy(dataflows, vendor, identifierProfile, 1506 "DataflowUtility.mergeDataflows"); 1507 ModelBindingManager.setGlobalIdentifierService(IdentifierService.forProfile(effective)); 1508 ModelBindingManager.setGlobalVendor(vendor); 1509 boolean optionSet = false; 1510 if (ModelBindingManager.getGlobalOption() == null) { 1511 Option option = new Option(); 1512 option.setVendor(vendor); 1513 ModelBindingManager.setGlobalOption(option); 1514 optionSet = true; 1515 } 1516 try { 1517 // ID 按下标 * DATAFLOW_ID_RANGE 偏移,避免 ID 冲突; 1518 // coordinate 第三维 fileIdx 按前面所有 dataflow 累积的源文件数偏移, 1519 // 使得合并后每个来源的 fileIdx 保持连续且不重叠。 1520 long cumulativeFileIdx = 0L; 1521 int i = 0; 1522 for (dataflow df : dataflows) { 1523 if (df != null) { 1524 int maxFileIdx = findMaxFileIdx(df); 1525 offsetDataflowIds(df, DATAFLOW_ID_RANGE * i, cumulativeFileIdx); 1526 if (maxFileIdx >= 0) { 1527 cumulativeFileIdx += (maxFileIdx + 1L); 1528 } 1529 } 1530 i++; 1531 } 1532 return mergeDataflowsWithDifferentStartId(dataflows, vendor, effective); 1533 } finally { 1534 ModelBindingManager.removeGlobalVendor(); 1535 ModelBindingManager.removeGlobalIdentifierService(); 1536 if (optionSet) { 1537 ModelBindingManager.removeGlobalOption(); 1538 } 1539 } 1540 } 1541 1542 /** 1543 * The one configuration a merge may run under, or a diagnosable failure. 1544 * 1545 * <p>Results produced under different rules cannot be merged: with tables insensitive on one 1546 * side and sensitive on the other, {@code Orders} and {@code orders} become one table or two 1547 * depending on which input happens to be merged first. Each input carries its producing 1548 * analysis's profile while it is in memory ({@link dataflow#getIdentifierPolicy()}), which is 1549 * what makes a mismatch reportable rather than silent. An input with NO provenance — anything 1550 * loaded back from XML — cannot be checked; it is accepted under the stated policy, and the 1551 * caller on that path is responsible for stating the right one. 1552 */ 1553 private static IdentifierProfile requireOneMergePolicy(Collection<dataflow> dataflows, 1554 EDbVendor vendor, IdentifierProfile stated, String caller) { 1555 IdentifierProfile effective = stated != null 1556 ? stated 1557 : IdentifierProfile.forVendor(vendor, IdentifierProfile.VendorFlags.defaults()); 1558 if (stated != null && stated.getVendor() != vendor) { 1559 throw new IllegalStateException(caller + ": identifier profile is for vendor " 1560 + stated.getVendor() + " but the merge vendor is " + vendor); 1561 } 1562 for (dataflow df : dataflows) { 1563 if (df == null) { 1564 continue; 1565 } 1566 IdentifierProfile produced = df.getIdentifierPolicy(); 1567 if (produced != null && !produced.hasSameEffectivePolicyAs(effective)) { 1568 throw new IllegalStateException(caller + ": cannot merge results produced under" 1569 + " different identifier configurations - one input was produced with " 1570 + produced + " but this merge runs under " + effective 1571 + ". Analyze every input of one merge with the same profile, or pass that" 1572 + " profile to this merge."); 1573 } 1574 } 1575 return effective; 1576 } 1577 1578 /** 1579 * 从一组临时 XML 文件迭代加载并合并 dataflow,节约内存。 1580 * <p> 1581 * 与 {@link #iterativeMergeDataflowsFromFilesByStartId(List, long)} 不同的是,本方法假定文件中的 1582 * dataflow 是用默认 startId=0 生成的(ID 空间互相冲突),因此会按下标先做 ID 偏移再合并。 1583 * <p> 1584 * 注意:本方法按入参 {@code tempFiles} 顺序迭代合并,不做任何排序。 1585 * coordinate 第三维 fileIdx 的偏移也是严格按入参顺序累积的, 1586 * 调用方传入的顺序应与原始源文件的语义顺序保持一致。 1587 */ 1588 public static dataflow iterativeMergeDataflowsFromFiles(List<File> tempFiles, EDbVendor vendor) { 1589 if (tempFiles == null || tempFiles.isEmpty()) { 1590 return null; 1591 } 1592 ModelBindingManager.setGlobalVendor(vendor); 1593 boolean optionSet = false; 1594 if (ModelBindingManager.getGlobalOption() == null) { 1595 Option option = new Option(); 1596 option.setVendor(vendor); 1597 ModelBindingManager.setGlobalOption(option); 1598 optionSet = true; 1599 } 1600 try { 1601 return iterativeMergeFromFiles(tempFiles, DATAFLOW_ID_RANGE * tempFiles.size(), true, null); 1602 } finally { 1603 ModelBindingManager.removeGlobalVendor(); 1604 if (optionSet) { 1605 ModelBindingManager.removeGlobalOption(); 1606 } 1607 } 1608 } 1609 1610 /** 1611 * 从一组临时 XML 文件迭代加载并合并 dataflow,调用方需保证文件中的 dataflow ID 空间互不冲突, 1612 * 且已设置好 ModelBindingManager。 1613 * <p> 1614 * 主要供 {@code ParallelDataFlowAnalyzer} 等内部生成方使用:每个临时文件中的 dataflow 已通过 1615 * {@code optionCopy.setStartId(DATAFLOW_ID_RANGE * i)} 占用了不重叠的 ID 段。 1616 * <p> 1617 * 注意:本方法按入参 {@code tempFiles} 顺序迭代合并,不做任何排序。 1618 * coordinate 第三维 fileIdx 的偏移严格按入参顺序累积。 1619 */ 1620 public static dataflow iterativeMergeDataflowsFromFilesByStartId(List<File> tempFiles, long startId) { 1621 return iterativeMergeFromFiles(tempFiles, startId, false, null); 1622 } 1623 1624 /** 1625 * The same merge, calling {@code afterLoad} with each file's dataflow as soon as it is read, before 1626 * any id is remapped: a spilled model's same-run state that XML does not carry (its tables' carried 1627 * identities, see {@link #carriedIdentities}) is put back there. 1628 */ 1629 public static dataflow iterativeMergeDataflowsFromFilesByStartId(List<File> tempFiles, long startId, 1630 java.util.function.BiConsumer<File, dataflow> afterLoad) { 1631 return iterativeMergeFromFiles(tempFiles, startId, false, afterLoad); 1632 } 1633 1634 /** 1635 * What the columns of one output table become when the elements that publish it are merged: in order, a function 1636 * column as it is, or the columns that are one column. The columns come grouped under their folded names, in the 1637 * order the names first appear. Under one folded name a star, a system column and a named column are never one 1638 * column, and named columns are one column unless they carry different identities ({@link #identityGroups}); 1639 * columns of one identity and one qualifier are one column whatever their names. 1640 * 1641 * @param columnsByFoldedName the columns of the table under each folded name 1642 * @return each element is a function {@link column} or a {@code List<column>} that is one column 1643 */ 1644 public static List<Object> columnsOfTable(Collection<List<column>> columnsByFoldedName) { 1645 List<Object> items = new ArrayList<Object>(); 1646 Map<List<Object>, List<column>> groupOfIdentity = new HashMap<List<Object>, List<column>>(); 1647 for (List<column> columnsOfName : columnsByFoldedName) { 1648 List<column> others = new ArrayList<column>(); 1649 for (column t : columnsOfName) { 1650 if (Boolean.TRUE.toString().equals(t.getIsFunction())) { 1651 items.add(t); 1652 } else { 1653 others.add(t); 1654 } 1655 } 1656 for (List<column> group : identityGroups(others)) { 1657 column keyed = keyedMemberOf(group); 1658 List<Object> key = keyed == null ? null : identityKey(keyed); 1659 List<column> earlier = key == null ? null : groupOfIdentity.get(key); 1660 if (earlier != null) { 1661 earlier.addAll(group); 1662 } else { 1663 items.add(group); 1664 if (key != null) { 1665 groupOfIdentity.put(key, group); 1666 } 1667 } 1668 } 1669 } 1670 return items; 1671 } 1672 1673 /** 1674 * Names that tell the columns of one table apart. Where columns of different identities have names that collide 1675 * under the collision key of {@link NormalNameCollisions} (which ignores case), each of them that knows its 1676 * spelling as SQL ({@code column.getSqlSpelling()}) is renamed as scanned collisions are published: the column 1677 * whose identity the name denotes keeps it, another gets its stored or delimited name (PostgreSQL 1678 * {@code "Amount"} beside {@code AMOUNT}, Oracle {@code "Amount"} beside {@code AMOUNT}). 1679 * 1680 * @return the new name of each renamed column, by column id; the caller renames the relationships naming them 1681 */ 1682 public static Map<String, String> nameCollidingIdentities(List<column> columns, EDbVendor vendor) { 1683 Map<String, String> renamed = new HashMap<String, String>(); 1684 if (columns == null || vendor == null) { 1685 return renamed; 1686 } 1687 IdentifierService service = ModelBindingManager.getGlobalIdentifierService(); 1688 if (service == null || service.getProfile().getVendor() != vendor) { 1689 service = IdentifierService.defaultServiceFor(vendor); 1690 } 1691 Map<String, List<column>> byKey = new LinkedHashMap<String, List<column>>(); 1692 for (column c : columns) { 1693 if (!NAMED_ROLE.equals(roleOf(c)) || c.getIdentifierKey() == null || c.getName() == null 1694 || Boolean.TRUE.toString().equals(c.getIsFunction())) { 1695 continue; 1696 } 1697 String key; 1698 try { 1699 key = NormalNameCollisions.contestedKey(service, ESQLDataObjectType.dotColumn, c.getName()); 1700 } catch (RuntimeException notAName) { 1701 continue; 1702 } 1703 List<column> same = byKey.get(key); 1704 if (same == null) { 1705 same = new ArrayList<column>(); 1706 byKey.put(key, same); 1707 } 1708 same.add(c); 1709 } 1710 for (Map.Entry<String, List<column>> collision : byKey.entrySet()) { 1711 Set<CanonKey> identities = new HashSet<CanonKey>(); 1712 for (column c : collision.getValue()) { 1713 identities.add(c.getIdentifierKey()); 1714 } 1715 if (identities.size() < 2) { 1716 continue; 1717 } 1718 for (column c : collision.getValue()) { 1719 if (c.getSqlSpelling() == null) { 1720 continue; 1721 } 1722 String name; 1723 try { 1724 name = NormalNameCollisions.normalName(Collections.singleton(collision.getKey()), service, 1725 ESQLDataObjectType.dotColumn, c.getSqlSpelling(), c.getName()); 1726 } catch (RuntimeException notAName) { 1727 continue; 1728 } 1729 // non-identifier-compare: whether the published text changes, not whether two names are one 1730 if (name != null && !name.equals(c.getName())) { 1731 c.setName(name); 1732 renamed.put(c.getId(), name); 1733 } 1734 } 1735 } 1736 return renamed; 1737 } 1738 1739 /** The spelling as SQL one of the columns that are one column knows, or null. */ 1740 public static String sqlSpellingOf(List<column> group) { 1741 for (column item : group) { 1742 if (item.getSqlSpelling() != null) { 1743 return item.getSqlSpelling(); 1744 } 1745 } 1746 return null; 1747 } 1748 1749 /** Renames the endpoints of the relationships that name a renamed column (by column id). */ 1750 /** 1751 * Whether two relationships that meet under one merge key come from the same merged dataflow. A merge makes one 1752 * relationship of what several dataflows say about the same columns; it does not undo what one dataflow's own 1753 * analysis kept apart, such as one edge published from two reference sites. 1754 */ 1755 private static boolean sameEdgeOfOneInput(relationship first, relationship other, Map<relationship, Integer> inputOf) { 1756 Integer a = inputOf.get(first); 1757 return a != null && a.equals(inputOf.get(other)); 1758 } 1759 1760 /** 1761 * The identity a routine and a call of it meet under in a merge, from a name as written: keyed as a function name 1762 * by the analysis's identifier rules ({@link DlineageUtil#lexicalNameKey}). The analysis takes it when it 1763 * publishes a routine or a call ({@code procedure.getRoutineKey()}, {@code table.getRoutineKey()}), before 1764 * normalized output renames them; a merge reads a published name only where nothing carries it (a document read 1765 * back from XML). 1766 */ 1767 public static Object routineKey(String name) { 1768 return DlineageUtil.lexicalNameKey(ModelBindingManager.getGlobalVendor(), name, ESQLDataObjectType.dotFunction); 1769 } 1770 1771 /** 1772 * The {@link #routineKey} of the routine a call names: the call's name qualified by the default database and schema 1773 * as written ({@link DlineageUtil#getTableFullNameAsWritten}), the way the merge has always qualified it. 1774 */ 1775 public static Object calledRoutineKey(String callName) { 1776 return routineKey(DlineageUtil.getTableFullNameAsWritten(callName)); 1777 } 1778 1779 /** The identity of {@code proc}: the one it carries, else its published name read as written. */ 1780 private static Object routineKeyOf(procedure proc) { 1781 return proc.getRoutineKey() != null ? proc.getRoutineKey() : routineKey(proc.getName()); 1782 } 1783 1784 /** The identity of the routine a function table calls: the one it carries, else its published name read as written. */ 1785 private static Object calledRoutineKey(table functionTable) { 1786 return functionTable.getRoutineKey() != null ? functionTable.getRoutineKey() 1787 : calledRoutineKey(functionTable.getName()); 1788 } 1789 1790 /** 1791 * What makes {@code r} the relationship it is, in every attribute but its own id: {@link relationship#toDedupKey} 1792 * and the ids of its endpoints, which that key leaves out so that a merge can join columns it remaps. Under the 1793 * dedup key alone, two relationships of different columns that share a label, a parent and a coordinate meet as 1794 * one: {@code tgt.* <- RS-1.*} from {@code s1.*} and from {@code s2.*} of one select list, once coordinates are 1795 * ignored. Only a relationship equal to another under this key repeats it. 1796 */ 1797 static String exactKey(relationship r) { 1798 StringBuilder key = new StringBuilder(r.toDedupKey()).append("ids="); 1799 if (r.getTarget() != null) { 1800 key.append(r.getTarget().getId()); 1801 } 1802 key.append('/'); 1803 if (r.getCaller() != null) { 1804 key.append(r.getCaller().getId()); 1805 } 1806 appendSortedIds(key, r.getSources()); 1807 appendSortedIds(key, r.getCallees()); 1808 return key.toString(); 1809 } 1810 1811 /** The ids of {@code endpoints}, sorted as the dedup key sorts the endpoints themselves. */ 1812 private static void appendSortedIds(StringBuilder key, List<sourceColumn> endpoints) { 1813 key.append('['); 1814 if (endpoints != null) { 1815 List<String> ids = new ArrayList<String>(); 1816 for (sourceColumn endpoint : endpoints) { 1817 ids.add(String.valueOf(endpoint.getId())); 1818 } 1819 Collections.sort(ids); 1820 for (String id : ids) { 1821 key.append(id).append(','); 1822 } 1823 } 1824 key.append(']'); 1825 } 1826 1827 /** 1828 * The published name of every column of {@code flow}, by the column's parent and its own id: what a relationship 1829 * endpoint that names the column by those ids is labelled ({@link #labelEndpoints}). 1830 */ 1831 public static Map<String, String> publishedColumnNames(dataflow flow) { 1832 Map<String, String> names = new HashMap<String, String>(); 1833 for (table t : dataflow.getAllTables(flow)) { 1834 if (t.getColumns() == null) { 1835 continue; 1836 } 1837 for (column c : t.getColumns()) { 1838 if (c.getId() != null && c.getName() != null) { 1839 names.put(endpointKey(t.getId(), c.getId()), c.getName()); 1840 } 1841 } 1842 } 1843 return names; 1844 } 1845 1846 private static String endpointKey(String parentId, String columnId) { 1847 return parentId + '\u0000' + columnId; 1848 } 1849 1850 /** 1851 * Labels each endpoint of {@code r} with the published name of the column its ids name ({@link 1852 * #publishedColumnNames}), so the label of an endpoint and the name of the column it points to never disagree: 1853 * a merge that makes one column of several keeps one name, and the endpoints that named the others by their own 1854 * spellings (a star link's folded display, the spelling of another reference) now carry it too. An endpoint whose 1855 * ids name no published column keeps its label; routine calls are not column endpoints. 1856 */ 1857 public static void labelEndpoints(relationship r, Map<String, String> publishedColumnNames) { 1858 if (r == null || RelationshipType.call.name().equals(r.getType())) { 1859 return; 1860 } 1861 if (r.getTarget() != null) { 1862 String name = publishedColumnNames.get(endpointKey(r.getTarget().getParent_id(), r.getTarget().getId())); 1863 if (name != null) { 1864 r.getTarget().setColumn(name); 1865 } 1866 } 1867 if (r.getSources() != null) { 1868 for (sourceColumn source : r.getSources()) { 1869 String name = publishedColumnNames.get(endpointKey(source.getParent_id(), source.getId())); 1870 if (name != null) { 1871 source.setColumn(name); 1872 } 1873 } 1874 } 1875 } 1876 1877 /** 1878 * {@link #labelEndpoints} for every relationship of {@code flow}, once its merge is done; then a relationship that 1879 * now repeats another in every attribute but its id ({@link #exactKey}: coordinates and the endpoints' ids 1880 * included, so no evidence site and no edge is lost) is dropped, and the comment bindings that named it follow the 1881 * one kept. 1882 */ 1883 public static void labelEndpoints(dataflow flow) { 1884 if (flow == null || flow.getRelationships() == null) { 1885 return; 1886 } 1887 Map<String, String> names = publishedColumnNames(flow); 1888 for (relationship r : flow.getRelationships()) { 1889 labelEndpoints(r, names); 1890 } 1891 Map<String, relationship> kept = new LinkedHashMap<String, relationship>(); 1892 Map<String, String> keptFor = new HashMap<String, String>(); 1893 for (relationship r : flow.getRelationships()) { 1894 String key = exactKey(r); 1895 relationship first = kept.get(key); 1896 if (first == null) { 1897 kept.put(key, r); 1898 } else if (r.getId() != null && first.getId() != null) { 1899 keptFor.put(r.getId(), first.getId()); 1900 } 1901 } 1902 if (kept.size() == flow.getRelationships().size()) { 1903 return; 1904 } 1905 flow.setRelationships(new ArrayList<relationship>(kept.values())); 1906 if (!keptFor.isEmpty()) { 1907 CommentBindingProjection.remap(flow, Collections.<String, String>emptyMap(), 1908 Collections.<String, String>emptyMap(), keptFor); 1909 } 1910 } 1911 1912 public static void renameEndpoints(List<relationship> relationships, Map<String, String> renamed) { 1913 if (relationships == null || renamed == null || renamed.isEmpty()) { 1914 return; 1915 } 1916 for (relationship r : relationships) { 1917 if (RelationshipType.call.name().equals(r.getType())) { 1918 continue; 1919 } 1920 if (r.getTarget() != null && renamed.containsKey(r.getTarget().getId())) { 1921 r.getTarget().setColumn(renamed.get(r.getTarget().getId())); 1922 } 1923 if (r.getSources() != null) { 1924 for (sourceColumn source : r.getSources()) { 1925 if (renamed.containsKey(source.getId())) { 1926 source.setColumn(renamed.get(source.getId())); 1927 } 1928 } 1929 } 1930 } 1931 } 1932 1933 /** What a column is, for a merge: a star and a system column are never one column with a named one. */ 1934 private static final String STAR_ROLE = "star"; 1935 private static final String SYSTEM_ROLE = "system"; 1936 private static final String NAMED_ROLE = "named"; 1937 /** Named columns without an identity, beside named columns of several identities. */ 1938 private static final Object NAMED_WITHOUT_IDENTITY = new Object(); 1939 1940 private static String roleOf(column item) { 1941 if (item.standsForStar()) { 1942 return STAR_ROLE; 1943 } 1944 return "system".equals(item.getSource()) ? SYSTEM_ROLE : NAMED_ROLE; 1945 } 1946 1947 /** 1948 * The columns of one table gathered under one folded name, as the columns they stand for, in the order they 1949 * first appear. A star ({@code *}), a system column ({@code RelationRows}) and a named column are never one 1950 * column: SQL Server {@code [*]} is a column, not the star. Named columns of one identity and one qualifier are 1951 * one column; the fold puts PostgreSQL {@code "Amount"} and {@code amount} together, their identities keep them 1952 * apart, and {@code a.id} stays apart from {@code b.id}. A named column without an identity is not evidence of 1953 * one: while star links are grouped by identity (the default, which gives every star link of an identifier its 1954 * identity, so that one without is a literal, an expression or a label such as a name inferred from a routine 1955 * parameter) it is only ever one column with the others that carry none, as the output merge of one analysis keys 1956 * them. While they are grouped by display, the unkeyed star links stand for the columns their names fold to, so 1957 * the named columns under the name are one column unless they carry several identities, and those without one 1958 * form one more. 1959 */ 1960 private static List<List<column>> identityGroups(List<column> columns) { 1961 Set<List<Object>> identities = new HashSet<List<Object>>(); 1962 for (column item : columns) { 1963 if (NAMED_ROLE.equals(roleOf(item)) && item.getIdentifierKey() != null) { 1964 identities.add(identityKey(item)); 1965 } 1966 } 1967 boolean split = identities.size() > 1 || groupsByIdentity(); 1968 Map<Object, List<column>> groups = new LinkedHashMap<Object, List<column>>(); 1969 for (column item : columns) { 1970 Object key = roleOf(item); 1971 if (NAMED_ROLE.equals(key) && split) { 1972 key = item.getIdentifierKey() != null ? identityKey(item) : NAMED_WITHOUT_IDENTITY; 1973 } 1974 List<column> group = groups.get(key); 1975 if (group == null) { 1976 group = new ArrayList<column>(); 1977 groups.put(key, group); 1978 } 1979 group.add(item); 1980 } 1981 return new ArrayList<List<column>>(groups.values()); 1982 } 1983 1984 /** Whether star links are grouped by identity in this JVM (the analyzer's model column, not the JSON one). */ 1985 private static boolean groupsByIdentity() { 1986 return gudusoft.gsqlparser.dlineage.dataflow.model.Column.starLinksGroupedByIdentity(); 1987 } 1988 1989 /** A named column's identity with its qualifier: {@code a.id} and {@code b.id} of one result set are two. */ 1990 private static List<Object> identityKey(column item) { 1991 return Arrays.<Object>asList(item.getQualifiedTable(), item.getIdentifierKey()); 1992 } 1993 1994 /** The first named member of a group that carries an identity, or null. */ 1995 private static column keyedMemberOf(List<column> group) { 1996 for (column item : group) { 1997 if (NAMED_ROLE.equals(roleOf(item)) && item.getIdentifierKey() != null) { 1998 return item; 1999 } 2000 } 2001 return null; 2002 } 2003 2004 /** 2005 * The identity a group of columns stands for: the one its named columns carry, or null for a star or system 2006 * group and for named columns that carry none. 2007 */ 2008 public static CanonKey identityOf(List<column> group) { 2009 for (column item : group) { 2010 if (NAMED_ROLE.equals(roleOf(item)) && item.getIdentifierKey() != null) { 2011 return item.getIdentifierKey(); 2012 } 2013 } 2014 return null; 2015 } 2016 2017 /** 2018 * What a merged column takes from a column it joins: each attribute the joined column states (the data type and 2019 * keys since issue I6DB5S, and the nullability, identity, computed and default attributes of a catalog column). 2020 */ 2021 private static void takeAttributes(column merged, column item) { 2022 if (item.getDataType() != null) { 2023 merged.setDataType(item.getDataType()); 2024 } 2025 if (item.isForeignKey() != null) { 2026 merged.setForeignKey(item.isForeignKey()); 2027 } 2028 if (item.isUnqiueKey() != null) { 2029 merged.setUnqiueKey(item.isUnqiueKey()); 2030 } 2031 if (item.isIndexKey() != null) { 2032 merged.setIndexKey(item.isIndexKey()); 2033 } 2034 if (item.isPrimaryKey() != null) { 2035 merged.setPrimaryKey(item.isPrimaryKey()); 2036 } 2037 if (item.isNullable() != null) { 2038 merged.setNullable(item.isNullable()); 2039 } 2040 if (item.isIdentity() != null) { 2041 merged.setIdentity(item.isIdentity()); 2042 } 2043 if (item.isComputed() != null) { 2044 merged.setComputed(item.isComputed()); 2045 } 2046 if (item.isPersisted() != null) { 2047 merged.setPersisted(item.isPersisted()); 2048 } 2049 if (item.getComputedDefinition() != null) { 2050 merged.setComputedDefinition(item.getComputedDefinition()); 2051 } 2052 if (item.getDefaultValue() != null) { 2053 merged.setDefaultValue(item.getDefaultValue()); 2054 } 2055 if (item.getDefaultName() != null) { 2056 merged.setDefaultName(item.getDefaultName()); 2057 } 2058 } 2059 2060 /** 2061 * What a spilled model's physical tables and views carried that XML does not, by table id: the 2062 * identities their models minted, which tables publish a reader's vouched SQL spelling 2063 * ({@code table.isPublishedNameSqlSpelling()}), the minted names a merged table carries 2064 * ({@code table.getMintedNames()}), which merged tables span legacy folded names 2065 * ({@code table.isAcrossLegacyNames()}), and the exact names a merged table's carried tables published 2066 * ({@code table.getCarriedPublishedNames()}). Independent facts, kept apart, so a restored table is keyed, 2067 * and named, as its producer left it. 2068 */ 2069 public static final class SpilledIdentities { 2070 private final Map<String, QualifiedCanonKey> minted = new HashMap<String, QualifiedCanonKey>(); 2071 private final Set<String> sqlSpellings = new HashSet<String>(); 2072 private final Map<String, table> mintedNames = new HashMap<String, table>(); 2073 private final Set<String> acrossLegacyNames = new HashSet<String>(); 2074 private final Map<String, Map<String, String>> carriedPublishedNames = new HashMap<String, Map<String, String>>(); 2075 /** The identity of each column that carries one, by column id: an XML round trip drops it. */ 2076 private final Map<String, CanonKey> columnKeys = new HashMap<String, CanonKey>(); 2077 /** The spelling as SQL of each column that knows it, by column id: an XML round trip drops it too. */ 2078 private final Map<String, String> columnSpellings = new HashMap<String, String>(); 2079 /** The routine identity of each routine and function table that carries one, by element id. */ 2080 private final Map<String, Object> routineKeys = new HashMap<String, Object>(); 2081 } 2082 2083 /** Captures the {@link SpilledIdentities} of a dataflow about to be serialized. */ 2084 public static SpilledIdentities carriedIdentities(dataflow df) { 2085 SpilledIdentities identities = new SpilledIdentities(); 2086 if (df == null) { 2087 return identities; 2088 } 2089 for (List<table> list : Arrays.asList(df.getTables(), df.getViews())) { 2090 if (list == null) { 2091 continue; 2092 } 2093 for (table t : list) { 2094 if (t.getId() == null) { 2095 continue; 2096 } 2097 if (t.getMergeIdentity() != null) { 2098 identities.minted.put(t.getId(), t.getMergeIdentity()); 2099 } 2100 if (t.isPublishedNameSqlSpelling()) { 2101 identities.sqlSpellings.add(t.getId()); 2102 } 2103 if (t.getMintedNames() != null) { 2104 identities.mintedNames.put(t.getId(), t.getMintedNames()); 2105 } 2106 if (t.isAcrossLegacyNames()) { 2107 identities.acrossLegacyNames.add(t.getId()); 2108 } 2109 if (t.getCarriedPublishedNames() != null) { 2110 identities.carriedPublishedNames.put(t.getId(), t.getCarriedPublishedNames()); 2111 } 2112 } 2113 } 2114 for (procedure proc : allProcedures(df)) { 2115 if (proc.getId() != null && proc.getRoutineKey() != null) { 2116 identities.routineKeys.put(proc.getId(), proc.getRoutineKey()); 2117 } 2118 } 2119 for (table t : dataflow.getAllTables(df)) { 2120 if (t.getId() != null && t.getRoutineKey() != null) { 2121 identities.routineKeys.put(t.getId(), t.getRoutineKey()); 2122 } 2123 if (t.getColumns() == null) { 2124 continue; 2125 } 2126 for (column c : t.getColumns()) { 2127 if (c.getId() != null && c.getIdentifierKey() != null) { 2128 identities.columnKeys.put(c.getId(), c.getIdentifierKey()); 2129 } 2130 if (c.getId() != null && c.getSqlSpelling() != null) { 2131 identities.columnSpellings.put(c.getId(), c.getSqlSpelling()); 2132 } 2133 } 2134 } 2135 return identities; 2136 } 2137 2138 /** 2139 * How a merge names the table it makes of a group of tables: which member it takes its properties from 2140 * ({@link #displayed}), which table's names it publishes ({@link #names}), and whether the group spans 2141 * legacy folded names, which only identities join ({@link #acrossLegacyNames}). 2142 * 2143 * <p>The member is the first of the group, as always, unless the group spans legacy folded names (a table 2144 * merged earlier may already span them); then it is the first table that offers names a model minted with 2145 * its identity (the vendor's bare names, rather than a metadata reader's delimited spelling), its own or 2146 * ones merged into it earlier, and those are the names published. A staged or incremental merge so names a 2147 * table as one merge of all the inputs does.</p> 2148 */ 2149 public static final class GroupNames { 2150 public final table displayed; 2151 public final table names; 2152 public final boolean acrossLegacyNames; 2153 2154 private GroupNames(table displayed, table names, boolean acrossLegacyNames) { 2155 this.displayed = displayed; 2156 this.names = names; 2157 this.acrossLegacyNames = acrossLegacyNames; 2158 } 2159 } 2160 2161 /** How a merge names the table it makes of {@code group}, in the group's order; see {@link GroupNames}. */ 2162 public static GroupNames groupNames(List<table> group) { 2163 if (spansLegacyNames(group)) { 2164 for (table t : group) { 2165 table minted = mintedNamesOf(t); 2166 if (minted != null) { 2167 return new GroupNames(t, minted, true); 2168 } 2169 } 2170 return new GroupNames(group.get(0), group.get(0), true); 2171 } 2172 return new GroupNames(group.get(0), group.get(0), false); 2173 } 2174 2175 /** 2176 * Whether the tables merged into one span legacy folded names: a table merged earlier spans them, or 2177 * the tables' published names fold apart. 2178 */ 2179 private static boolean spansLegacyNames(List<table> merged) { 2180 for (table t : merged) { 2181 if (t.isAcrossLegacyNames()) { 2182 return true; 2183 } 2184 } 2185 String firstLegacy = legacyTableKey(DlineageUtil.getQualifiedTableName(merged.get(0))); 2186 for (table t : merged) { 2187 String legacy = legacyTableKey(DlineageUtil.getQualifiedTableName(t)); 2188 if (firstLegacy == null ? legacy != null : !firstLegacy.equals(legacy)) { // non-identifier-compare: legacy bucket keys 2189 return true; 2190 } 2191 } 2192 return false; 2193 } 2194 2195 /** 2196 * Whether a table publishes names a model minted with its identity: it carries an identity, its names 2197 * are not a reader's spelling, and it carries no other table's minted names (a merged table does when its 2198 * own names came from a table without them). Which names a table publishes is a fact of its own, 2199 * independent of the identity a merge gave it. 2200 */ 2201 private static boolean hasMintedNames(table t) { 2202 return t.getMintedNames() == null && t.getMergeIdentity() != null && !t.isPublishedNameSqlSpelling(); 2203 } 2204 2205 /** The minted names a table offers a merge: its own, or those merged into it earlier; null if none. */ 2206 private static table mintedNamesOf(table t) { 2207 return hasMintedNames(t) ? t : t.getMintedNames(); 2208 } 2209 2210 /** The first minted names the tables offer, as a table holding only those names and their identity. */ 2211 private static table firstMintedNames(List<table> tables) { 2212 for (table t : tables) { 2213 table minted = mintedNamesOf(t); 2214 if (minted != null) { 2215 return minted == t ? mintedNamesCopy(t) : minted; 2216 } 2217 } 2218 return null; 2219 } 2220 2221 private static table mintedNamesCopy(table t) { 2222 table names = new table(); 2223 names.setServer(t.getServer()); 2224 names.setDatabase(t.getDatabase()); 2225 names.setSchema(t.getSchema()); 2226 names.setName(t.getName()); 2227 names.setDisplayName(t.getDisplayName()); 2228 names.setMergeIdentity(t.getMergeIdentity()); 2229 return names; 2230 } 2231 2232 /** The identity a model minted for any of the tables merged into one; they share one merge key. */ 2233 public static QualifiedCanonKey groupIdentity(List<table> merged) { 2234 for (table t : merged) { 2235 if (t.getMergeIdentity() != null) { 2236 return t.getMergeIdentity(); 2237 } 2238 } 2239 return null; 2240 } 2241 2242 /** 2243 * Gives a merged table the identity of the tables it replaces (the one a model minted), the marker of the 2244 * table its names come from (whether those names are a reader's vouched SQL spelling), when those names 2245 * are not ones a model minted the first minted names the group offers, and whether the group spans 2246 * legacy folded names. The facts stay independent, so a later merge that includes the table chooses 2247 * names as one merge of all the inputs would. 2248 */ 2249 public static void carryGroupIdentity(table merged, List<table> group, GroupNames chosen) { 2250 merged.setMergeIdentity(groupIdentity(group)); 2251 merged.setPublishedNameSqlSpelling(chosen.names.isPublishedNameSqlSpelling()); 2252 merged.setMintedNames(hasMintedNames(chosen.names) ? null : firstMintedNames(group)); 2253 merged.setAcrossLegacyNames(chosen.acrossLegacyNames); 2254 Map<String, String> published = new LinkedHashMap<String, String>(); 2255 for (table t : group) { 2256 addCarriedPublishedNames(published, t); 2257 } 2258 merged.setCarriedPublishedNames(carriedPublishedNamesBeside(published, chosen.names)); 2259 } 2260 2261 /** 2262 * Adds the exact names a table offers a merge as evidence (see {@link PhysicalTableMergeKeys#addTable}): those 2263 * merged into it, and its own when it carries an identity. A name added without its legacy key is keyed later. 2264 */ 2265 private static void addCarriedPublishedNames(Map<String, String> into, table t) { 2266 if (t.getCarriedPublishedNames() != null) { 2267 for (Map.Entry<String, String> name : t.getCarriedPublishedNames().entrySet()) { 2268 if (!into.containsKey(name.getKey())) { 2269 into.put(name.getKey(), name.getValue()); 2270 } 2271 } 2272 } 2273 if (t.getMergeIdentity() != null || t.isPublishedNameSqlSpelling()) { 2274 String own = DlineageUtil.getQualifiedTableName(t); 2275 if (own != null && !into.containsKey(own)) { 2276 into.put(own, null); 2277 } 2278 } 2279 } 2280 2281 /** 2282 * The evidence names a merged table keeps: those other than the name it publishes itself, which a merge 2283 * registers anyway, with their legacy keys; null when there are none. 2284 */ 2285 private static Map<String, String> carriedPublishedNamesBeside(Map<String, String> published, table namesFrom) { 2286 published.remove(DlineageUtil.getQualifiedTableName(namesFrom)); 2287 if (published.isEmpty()) { 2288 return null; 2289 } 2290 for (Map.Entry<String, String> name : published.entrySet()) { 2291 if (name.getValue() == null) { 2292 name.setValue(legacyTableKey(name.getKey())); 2293 } 2294 } 2295 return published; 2296 } 2297 2298 /** Puts back what {@link #carriedIdentities} captured, matching tables by id. */ 2299 /** The routines of {@code df}: its procedures and those of its packages. */ 2300 private static List<procedure> allProcedures(dataflow df) { 2301 List<procedure> procedures = new ArrayList<procedure>(); 2302 if (df.getProcedures() != null) { 2303 procedures.addAll(df.getProcedures()); 2304 } 2305 if (df.getPackages() != null) { 2306 for (oraclePackage pkg : df.getPackages()) { 2307 if (pkg.getProcedures() != null) { 2308 procedures.addAll(pkg.getProcedures()); 2309 } 2310 } 2311 } 2312 return procedures; 2313 } 2314 2315 public static void restoreCarriedIdentities(dataflow df, SpilledIdentities identities) { 2316 if (df == null || identities == null) { 2317 return; 2318 } 2319 for (List<table> list : Arrays.asList(df.getTables(), df.getViews())) { 2320 if (list == null) { 2321 continue; 2322 } 2323 for (table t : list) { 2324 if (t.getId() == null) { 2325 continue; 2326 } 2327 QualifiedCanonKey identity = identities.minted.get(t.getId()); 2328 if (identity != null) { 2329 t.setMergeIdentity(identity); 2330 } 2331 if (identities.sqlSpellings.contains(t.getId())) { 2332 t.setPublishedNameSqlSpelling(true); 2333 } 2334 table mintedNames = identities.mintedNames.get(t.getId()); 2335 if (mintedNames != null) { 2336 t.setMintedNames(mintedNames); 2337 } 2338 if (identities.acrossLegacyNames.contains(t.getId())) { 2339 t.setAcrossLegacyNames(true); 2340 } 2341 Map<String, String> published = identities.carriedPublishedNames.get(t.getId()); 2342 if (published != null) { 2343 t.setCarriedPublishedNames(published); 2344 } 2345 } 2346 } 2347 if (!identities.routineKeys.isEmpty()) { 2348 for (procedure proc : allProcedures(df)) { 2349 Object key = proc.getId() == null ? null : identities.routineKeys.get(proc.getId()); 2350 if (key != null) { 2351 proc.setRoutineKey(key); 2352 } 2353 } 2354 for (table t : dataflow.getAllTables(df)) { 2355 Object key = t.getId() == null ? null : identities.routineKeys.get(t.getId()); 2356 if (key != null) { 2357 t.setRoutineKey(key); 2358 } 2359 } 2360 } 2361 if (identities.columnKeys.isEmpty() && identities.columnSpellings.isEmpty()) { 2362 return; 2363 } 2364 for (table t : dataflow.getAllTables(df)) { 2365 if (t.getColumns() == null) { 2366 continue; 2367 } 2368 for (column c : t.getColumns()) { 2369 CanonKey key = c.getId() == null ? null : identities.columnKeys.get(c.getId()); 2370 if (key != null) { 2371 c.setIdentifierKey(key); 2372 } 2373 String spelling = c.getId() == null ? null : identities.columnSpellings.get(c.getId()); 2374 if (spelling != null) { 2375 c.setSqlSpelling(spelling); 2376 } 2377 } 2378 } 2379 } 2380 2381 private static dataflow iterativeMergeFromFiles(List<File> tempFiles, long startId, boolean offsetIds, 2382 java.util.function.BiConsumer<File, dataflow> afterLoad) { 2383 if (tempFiles == null || tempFiles.isEmpty()) { 2384 logger.warn("iterativeMergeDataflowsFromFiles: tempFiles is null or empty"); 2385 return null; 2386 } 2387 2388 logger.info("iterativeMergeDataflowsFromFiles: total dataflows: " + tempFiles.size()); 2389 2390 long startTime = System.currentTimeMillis(); 2391 int totalIterations = tempFiles.size() - 1; 2392 logger.info("iterativeMergeDataflowsFromFiles: start merging, total iterations: " + totalIterations); 2393 2394 long loadStartTime = System.currentTimeMillis(); 2395 dataflow mergedDataflow = XML2Model.loadXML(dataflow.class, tempFiles.get(0)); 2396 if (afterLoad != null && mergedDataflow != null) { 2397 afterLoad.accept(tempFiles.get(0), mergedDataflow); 2398 } 2399 long cumulativeFileIdx = 0L; 2400 if (mergedDataflow != null) { 2401 long idOffset = offsetIds ? 0L : 0L; 2402 offsetDataflowIds(mergedDataflow, idOffset, 0L); 2403 int maxFileIdx = findMaxFileIdx(mergedDataflow); 2404 if (maxFileIdx >= 0) { 2405 cumulativeFileIdx = maxFileIdx + 1L; 2406 } 2407 } 2408 long loadEndTime = System.currentTimeMillis(); 2409 int initialRelationCount = mergedDataflow != null && mergedDataflow.getRelationships() != null ? mergedDataflow.getRelationships().size() : 0; 2410 logger.info("iterativeMergeDataflowsFromFiles: loaded initial dataflow, relation count: " + initialRelationCount + ", time: " + (loadEndTime - loadStartTime) + "ms"); 2411 2412 Set<Long> mergedDedupKeys = new HashSet<>(Math.max(initialRelationCount, 16)); 2413 if (mergedDataflow != null && mergedDataflow.getRelationships() != null) { 2414 for (relationship rel : mergedDataflow.getRelationships()) { 2415 mergedDedupKeys.add(rel.toDedupHash()); 2416 } 2417 } 2418 2419 List<CommentBindingBundle> commentBundles = null; 2420 if (mergedDataflow != null && mergedDataflow.getCommentBindings() != null) { 2421 commentBundles = new ArrayList<CommentBindingBundle>(); 2422 commentBundles.add(mergedDataflow.getCommentBindings()); 2423 } 2424 long currentMaxId = startId; 2425 // the names each column of an identity was published under, by column id (joinColumnsWithoutIdentity) 2426 Map<String, Set<String>> identityColumnNames = new HashMap<String, Set<String>>(); 2427 for (int i = 1; i < tempFiles.size(); i++) { 2428 long iterationStartTime = System.currentTimeMillis(); 2429 int currentIteration = i; 2430 int remainingIterations = totalIterations - currentIteration + 1; 2431 2432 logger.info("iterativeMergeDataflowsFromFiles: iteration " + currentIteration + "/" + totalIterations + ", remaining: " + remainingIterations); 2433 2434 long loadCurrentStartTime = System.currentTimeMillis(); 2435 dataflow currentDataflow = XML2Model.loadXML(dataflow.class, tempFiles.get(i)); 2436 if (afterLoad != null && currentDataflow != null) { 2437 afterLoad.accept(tempFiles.get(i), currentDataflow); 2438 } 2439 if (currentDataflow != null) { 2440 int maxFileIdx = findMaxFileIdx(currentDataflow); 2441 long idOffset = offsetIds ? (DATAFLOW_ID_RANGE * i) : 0L; 2442 offsetDataflowIds(currentDataflow, idOffset, cumulativeFileIdx); 2443 if (maxFileIdx >= 0) { 2444 cumulativeFileIdx += (maxFileIdx + 1L); 2445 } 2446 } 2447 long loadCurrentEndTime = System.currentTimeMillis(); 2448 int currentRelationCount = currentDataflow != null && currentDataflow.getRelationships() != null ? currentDataflow.getRelationships().size() : 0; 2449 logger.info("iterativeMergeDataflowsFromFiles: loaded dataflow[" + i + "], relation count: " + currentRelationCount + ", time: " + (loadCurrentEndTime - loadCurrentStartTime) + "ms"); 2450 2451 long mergeStartTime = System.currentTimeMillis(); 2452 //autogenerate 首个 dataflow 加载失败(mergedDataflow==null)时,以首个可用 dataflow 作为合并基准并初始化去重索引 2453 if (mergedDataflow == null && currentDataflow != null) { 2454 mergedDataflow = currentDataflow; 2455 if (mergedDataflow.getRelationships() != null) { 2456 for (relationship rel : mergedDataflow.getRelationships()) { 2457 mergedDedupKeys.add(rel.toDedupHash()); 2458 } 2459 } 2460 } else { 2461 currentMaxId = incrementalMergeDataflow(mergedDataflow, currentDataflow, currentMaxId, mergedDedupKeys, 2462 identityColumnNames); 2463 } 2464 long mergeEndTime = System.currentTimeMillis(); 2465 int mergedRelationCount = mergedDataflow != null && mergedDataflow.getRelationships() != null ? mergedDataflow.getRelationships().size() : 0; 2466 2467 long iterationEndTime = System.currentTimeMillis(); 2468 long iterationTime = iterationEndTime - iterationStartTime; 2469 long mergeTime = mergeEndTime - mergeStartTime; 2470 logger.info("iterativeMergeDataflowsFromFiles: iteration " + currentIteration + " completed, merged relation count: " + mergedRelationCount + ", merge time: " + mergeTime + "ms, total iteration time: " + iterationTime + "ms"); 2471 2472 if (currentDataflow != null && currentDataflow.getCommentBindings() != null) { 2473 if (commentBundles == null) commentBundles = new ArrayList<CommentBindingBundle>(); 2474 commentBundles.add(currentDataflow.getCommentBindings()); 2475 } 2476 2477 currentDataflow = null; 2478 } 2479 2480 long endTime = System.currentTimeMillis(); 2481 long totalTime = endTime - startTime; 2482 int finalRelationCount = mergedDataflow != null && mergedDataflow.getRelationships() != null ? mergedDataflow.getRelationships().size() : 0; 2483 logger.info("iterativeMergeDataflowsFromFiles: all iterations completed, final relation count: " + finalRelationCount + ", total time: " + totalTime + "ms (" + (totalTime / 1000.0) + "s)"); 2484 2485 if (mergedDataflow != null && commentBundles != null) { 2486 mergedDataflow.setCommentBindings(CommentBindingProjection.merge(commentBundles)); 2487 CommentBindingProjection.retain(mergedDataflow); 2488 } 2489 if (mergedDataflow != null) { 2490 joinColumnsWithoutIdentity(mergedDataflow, identityColumnNames); 2491 labelEndpoints(mergedDataflow); 2492 } 2493 2494 return mergedDataflow; 2495 } 2496 2497 private static Set<String> namesOf(Map<String, Set<String>> identityColumnNames, column identityColumn) { 2498 Set<String> names = identityColumnNames.get(identityColumn.getId()); 2499 if (names == null) { 2500 names = new HashSet<String>(); 2501 names.add(identityColumn.getName()); 2502 identityColumnNames.put(identityColumn.getId(), names); 2503 } 2504 return names; 2505 } 2506 2507 /** 2508 * The last step of the file-by-file merge while star links are grouped by display ({@code 2509 * -Dgsp.dlineage.starLinks=legacy}): a named column without an identity joins the column of the one identity 2510 * published under its name, as the merge of results in memory joins it ({@link #identityGroups}); under a name 2511 * that several identities were published under it stays a column of its own, since nothing tells which one it is. 2512 * The joined column keeps the place and name of whichever came first, the relationships and comment bindings that 2513 * named the other follow it, and relationships that became equal are kept once. Grouped by identity, the default, 2514 * a column without an identity is never taken for a column of one, and this step does nothing. 2515 */ 2516 private static void joinColumnsWithoutIdentity(dataflow merged, Map<String, Set<String>> identityColumnNames) { 2517 Map<String, column> survivorOf = new HashMap<String, column>(); 2518 for (table t : dataflow.getAllTables(merged)) { 2519 // a result set is one analysis's, never merged with another file's: it keeps its columns as published; 2520 // grouped by identity, a column without one is never taken for a column of an identity 2521 if (t.getColumns() == null || t.isFunction() || t.isResultSet() || groupsByIdentity()) { 2522 continue; 2523 } 2524 // the column of each identity (none when the table holds two of it), and the identities of each name 2525 Map<CanonKey, column> columnOfIdentity = new HashMap<CanonKey, column>(); 2526 Set<CanonKey> twice = new HashSet<CanonKey>(); 2527 Map<String, Set<CanonKey>> identitiesOfName = new HashMap<String, Set<CanonKey>>(); 2528 for (column c : t.getColumns()) { 2529 if (Boolean.TRUE.toString().equals(c.getIsFunction()) || !NAMED_ROLE.equals(roleOf(c)) 2530 || c.getIdentifierKey() == null) { 2531 continue; 2532 } 2533 if (columnOfIdentity.put(c.getIdentifierKey(), c) != null) { 2534 twice.add(c.getIdentifierKey()); 2535 } 2536 Set<String> names = identityColumnNames.get(c.getId()); 2537 for (String name : names != null ? names : Collections.singleton(c.getName())) { 2538 Set<CanonKey> identities = identitiesOfName.get(name); 2539 if (identities == null) { 2540 identities = new HashSet<CanonKey>(); 2541 identitiesOfName.put(name, identities); 2542 } 2543 identities.add(c.getIdentifierKey()); 2544 } 2545 } 2546 if (columnOfIdentity.isEmpty()) { 2547 continue; 2548 } 2549 // each column of one identity with the columns without an identity that join it, in table order 2550 Map<CanonKey, List<column>> groups = new LinkedHashMap<CanonKey, List<column>>(); 2551 for (column c : t.getColumns()) { 2552 if (Boolean.TRUE.toString().equals(c.getIsFunction()) || !NAMED_ROLE.equals(roleOf(c))) { 2553 continue; 2554 } 2555 CanonKey identity = c.getIdentifierKey(); 2556 if (identity == null) { 2557 Set<CanonKey> identities = identitiesOfName.get(c.getName()); 2558 identity = identities != null && identities.size() == 1 ? identities.iterator().next() : null; 2559 } 2560 if (identity == null || twice.contains(identity)) { 2561 continue; 2562 } 2563 List<column> group = groups.get(identity); 2564 if (group == null) { 2565 group = new ArrayList<column>(); 2566 groups.put(identity, group); 2567 } 2568 group.add(c); 2569 } 2570 for (Map.Entry<CanonKey, List<column>> group : groups.entrySet()) { 2571 List<column> members = group.getValue(); 2572 if (members.size() < 2) { 2573 continue; 2574 } 2575 column survivor = members.get(0); 2576 survivor.setIdentifierKey(group.getKey()); 2577 for (column other : members.subList(1, members.size())) { 2578 survivor.appendCoordinate(other.getCoordinate()); 2579 takeAttributes(survivor, other); 2580 if (survivor.getSqlSpelling() == null) { 2581 survivor.setSqlSpelling(other.getSqlSpelling()); 2582 } 2583 t.getColumns().remove(other); 2584 survivorOf.put(other.getId(), survivor); 2585 } 2586 } 2587 } 2588 // columns of one table whose names collide across identities get names that tell them apart 2589 Map<String, String> renamedColumns = new HashMap<String, String>(); 2590 for (table t : dataflow.getAllTables(merged)) { 2591 if (t.getColumns() != null && !t.isFunction() && !t.isResultSet()) { 2592 renamedColumns.putAll(nameCollidingIdentities(t.getColumns(), ModelBindingManager.getGlobalVendor())); 2593 } 2594 } 2595 if (survivorOf.isEmpty()) { 2596 renameEndpoints(merged.getRelationships(), renamedColumns); 2597 return; 2598 } 2599 Map<String, String> columnIds = new HashMap<String, String>(); 2600 for (Map.Entry<String, column> e : survivorOf.entrySet()) { 2601 columnIds.put(e.getKey(), e.getValue().getId()); 2602 } 2603 Map<String, String> relationIds = new HashMap<String, String>(); 2604 if (merged.getRelationships() != null) { 2605 // the relationships that named a joined column now name the column it joined; one that became equal to 2606 // another is kept once, and no other relationship is touched 2607 Set<relationship> renamed = Collections.newSetFromMap(new IdentityHashMap<relationship, Boolean>()); 2608 for (relationship r : merged.getRelationships()) { 2609 if (RelationshipType.call.name().equals(r.getType())) { 2610 continue; 2611 } 2612 if (r.getTarget() != null && survivorOf.containsKey(r.getTarget().getId())) { 2613 column survivor = survivorOf.get(r.getTarget().getId()); 2614 r.getTarget().setId(survivor.getId()); 2615 r.getTarget().setCoordinate(survivor.getCoordinate()); 2616 renamed.add(r); 2617 } 2618 if (r.getSources() != null) { 2619 for (sourceColumn source : r.getSources()) { 2620 if (survivorOf.containsKey(source.getId())) { 2621 source.setId(survivorOf.get(source.getId()).getId()); 2622 renamed.add(r); 2623 } 2624 } 2625 } 2626 } 2627 Map<Long, String> kept = new HashMap<Long, String>(); 2628 for (relationship r : merged.getRelationships()) { 2629 if (!renamed.contains(r) && !kept.containsKey(r.toDedupHash())) { 2630 kept.put(r.toDedupHash(), r.getId()); 2631 } 2632 } 2633 List<relationship> relationships = new ArrayList<relationship>(); 2634 for (relationship r : merged.getRelationships()) { 2635 if (renamed.contains(r)) { 2636 long hash = r.toDedupHash(); 2637 String earlier = kept.get(hash); 2638 if (earlier != null) { 2639 if (r.getId() != null) { 2640 relationIds.put(r.getId(), earlier); 2641 } 2642 continue; 2643 } 2644 kept.put(hash, r.getId()); 2645 } 2646 relationships.add(r); 2647 } 2648 merged.setRelationships(relationships); 2649 } 2650 if (merged.getCommentBindings() != null) { 2651 CommentBindingProjection.remap(merged, Collections.<String, String>emptyMap(), columnIds, relationIds); 2652 CommentBindingProjection.retain(merged); 2653 } 2654 renameEndpoints(merged.getRelationships(), renamedColumns); 2655 } 2656 2657 /** 2658 * 增量合并:将 {@code currentDataflow} 合并到 {@code mergedDataflow} 中,原地修改 mergedDataflow。 2659 * <p> 2660 * 与 {@link #mergeDataflowsByStartId} 不同,本方法保持 mergedDataflow 中已有表/列的 ID 不变, 2661 * 只将 currentDataflow 中同名表/列的属性追加到已有表/列上,已有关系不做任何处理。 2662 * 仅对 currentDataflow 的关系做 ID 重映射和去重后追加到 mergedDataflow。 2663 * 2664 * @param mergedDataflow 已合并的 dataflow(原地修改) 2665 * @param currentDataflow 新加载的 dataflow 2666 * @param currentMaxId 当前最大 ID,用于生成新 ID 2667 * @param mergedDedupKeys 已有关系的去重 hash 集合(原地修改) 2668 * @return 更新后的最大 ID 2669 */ 2670 private static long incrementalMergeDataflow(dataflow mergedDataflow, dataflow currentDataflow, 2671 long currentMaxId, Set<Long> mergedDedupKeys, 2672 Map<String, Set<String>> identityColumnNames) { 2673 //autogenerate mergedDataflow 为 null 时不可解引用,避免首个 dataflow 加载失败引发 NPE 2674 if (currentDataflow == null || mergedDataflow == null) { 2675 return currentMaxId; 2676 } 2677 2678 // 1. 构建 mergedDataflow 已有表的名称索引 2679 Map<String, table> existingTableMap = new LinkedHashMap<>(); 2680 Set<table> physicalTables = physicalTablesOf(mergedDataflow, currentDataflow); 2681 List<table> bothSides = new ArrayList<table>(dataflow.getAllTables(mergedDataflow)); 2682 bothSides.addAll(dataflow.getAllTables(currentDataflow)); 2683 PhysicalTableMergeKeys mergeKeys = physicalTableMergeKeys(bothSides, physicalTables); 2684 for (table t : dataflow.getAllTables(mergedDataflow)) { 2685 String tableFullName = getTableFullNameKey(t, physicalTables, mergeKeys); 2686 if (!existingTableMap.containsKey(tableFullName)) { 2687 existingTableMap.put(tableFullName, t); 2688 } 2689 } 2690 2691 // 2. 处理 currentDataflow 的表和列,构建 ID 映射(只映射新ID → 已有ID) 2692 Map<String, String> tableIdMap = new HashMap<>(); 2693 Map<String, String> columnIdMap = new HashMap<>(); 2694 Map<String, column> columnMergeIdMap = new HashMap<>(); 2695 Set<String> newOrModifiedTableIds = new HashSet<>(); 2696 2697 for (table t : dataflow.getAllTables(currentDataflow)) { 2698 String tableFullName = getTableFullNameKey(t, physicalTables, mergeKeys); 2699 table existingTable = existingTableMap.get(tableFullName); 2700 2701 if (existingTable != null && !t.isFunction()) { 2702 // 同名表:合并到已有表,已有表 ID 保持不变 2703 tableIdMap.put(t.getId(), existingTable.getId()); 2704 mergeTableProperties(existingTable, t); 2705 mergeColumnsIncremental(existingTable, t, columnIdMap, columnMergeIdMap, identityColumnNames); 2706 newOrModifiedTableIds.add(existingTable.getId()); 2707 } else { 2708 // 新表:直接追加到 mergedDataflow 2709 addTableToMergedDataflow(mergedDataflow, t); 2710 existingTableMap.put(tableFullName, t); 2711 tableIdMap.put(t.getId(), t.getId()); 2712 newOrModifiedTableIds.add(t.getId()); 2713 if (t.getColumns() != null) { 2714 for (column c : t.getColumns()) { 2715 columnIdMap.put(c.getId(), c.getId()); 2716 columnMergeIdMap.put(c.getId(), c); 2717 } 2718 } 2719 } 2720 } 2721 2722 // 3. 追加其他实体 2723 appendOtherEntities(mergedDataflow, currentDataflow); 2724 2725 // 4. 构建 function→resultSet 映射(基于合并后的完整数据) 2726 Map<String, String> functionIdToResultSetId = new HashMap<>(); 2727 Map<String, column> funcColIdToResultSetCol = new HashMap<>(); 2728 Map<String, table> allTableIdMap = new HashMap<>(); 2729 buildFunctionResultSetMapping(mergedDataflow, functionIdToResultSetId, 2730 funcColIdToResultSetCol, allTableIdMap); 2731 2732 // 5. 只处理 currentDataflow 的关系 2733 if (currentDataflow.getRelationships() != null) { 2734 List<relationship> newRelations = new ArrayList<>(); 2735 // the dedup keys this dataflow added, stamped and as published (sameEdgeOfOneInput) 2736 Set<Long> ownKeys = new HashSet<>(); 2737 Set<Long> ownPublishedKeys = new HashSet<>(); 2738 for (relationship rel : currentDataflow.getRelationships()) { 2739 remapAndDedupRelation(rel, tableIdMap, columnIdMap, columnMergeIdMap, 2740 functionIdToResultSetId, funcColIdToResultSetCol, allTableIdMap, 2741 mergedDedupKeys, ownKeys, ownPublishedKeys, newRelations); 2742 } 2743 if (!newRelations.isEmpty()) { 2744 if (mergedDataflow.getRelationships() == null) { 2745 mergedDataflow.setRelationships(new ArrayList<relationship>()); 2746 } 2747 mergedDataflow.getRelationships().addAll(newRelations); 2748 } 2749 } 2750 2751 if (currentDataflow.getCommentBindings() != null) { 2752 Map<Long,String> survivingRelations = new HashMap<Long,String>(); 2753 for (relationship r : mergedDataflow.getRelationships()) survivingRelations.put(r.toDedupHash(), r.getId()); 2754 Map<String,String> relationIds = new HashMap<String,String>(); 2755 for (relationship r : currentDataflow.getRelationships()) { 2756 String canonical = survivingRelations.get(r.toDedupHash()); 2757 if (canonical != null) relationIds.put(r.getId(), canonical); 2758 } 2759 CommentBindingProjection.remap(currentDataflow, tableIdMap, columnIdMap, relationIds); 2760 // The outer streaming merge collects these scalar snapshots and namespaces once. 2761 // Repeatedly copying the growing bundle would make comment merging quadratic. 2762 } 2763 2764 // 6. 为新表或修改过的表补齐 * 列桥接关系 2765 if (!newOrModifiedTableIds.isEmpty()) { 2766 currentMaxId = addStarColumnBridgingRelationsForTables(mergedDataflow, currentMaxId, 2767 newOrModifiedTableIds); 2768 } 2769 2770 return currentMaxId; 2771 } 2772 2773 /** The legacy folded name a table's merge bucket is keyed by. */ 2774 private static String legacyTableKey(String qualifiedTableName) { 2775 return LegacyDlineageWireNamesV1.tableBucket(qualifiedTableName); 2776 } 2777 2778 /** The first pass of {@link PhysicalTableMergeKeys} over the physical tables and views among {@code tables}. */ 2779 static PhysicalTableMergeKeys physicalTableMergeKeys(Collection<table> tables, Set<table> physicalTables) { 2780 PhysicalTableMergeKeys mergeKeys = new PhysicalTableMergeKeys(DlineageUtil.configuresIdentifierRules()); 2781 for (table table : tables) { 2782 if (physicalTables.contains(table)) { 2783 String qualifiedTableName = DlineageUtil.getQualifiedTableName(table); 2784 String legacyKey = legacyTableKey(qualifiedTableName); 2785 mergeKeys.addTable(table, legacyKey, qualifiedTableName); 2786 } 2787 } 2788 return mergeKeys; 2789 } 2790 2791 /** 2792 * 获取表的全名 key,用于同名表匹配。 2793 */ 2794 private static String getTableFullNameKey(table table, Set<table> physicalTables, 2795 PhysicalTableMergeKeys mergeKeys) { 2796 String qualifiedTableName = DlineageUtil.getQualifiedTableName(table); 2797 String tableFullName = legacyTableKey(qualifiedTableName); 2798 if (physicalTables.contains(table)) { 2799 // Two tables the analysis's rules keep apart must not be matched by their folded name. 2800 tableFullName = mergeKeys.keyOf(tableFullName, 2801 PhysicalTableMergeKeys.carriedIdentity(table, tableFullName, qualifiedTableName), qualifiedTableName); 2802 } 2803 if ("variable".endsWith(table.getType()) && !SQLUtil.isEmpty(table.getParent())) { 2804 tableFullName = table.getParent() + "." + tableFullName; 2805 } 2806 if (table.isFile() && (table.getDataSourceName() != null || table.getEndpointKind() != null)) { 2807 // A file a BULK INSERT names relative to an external data source, or a runtime template, is not the file 2808 // of its name in another unit (Mantis 4962); the same key as DataFlowAnalyzer's merge pass. 2809 tableFullName = tableFullName + "\u0000path\u0000" + table.getUri() + "\u0000" + table.getDataSourceName() 2810 + "\u0000" + table.getEndpointKind(); 2811 } 2812 return tableFullName; 2813 } 2814 2815 /** The physical tables and views of the given dataflows, by identity. */ 2816 private static Set<table> physicalTablesOf(dataflow... dataflows) { 2817 Set<table> physicalTables = Collections.newSetFromMap(new IdentityHashMap<table, Boolean>()); 2818 for (dataflow flow : dataflows) { 2819 if (flow.getTables() != null) { 2820 physicalTables.addAll(flow.getTables()); 2821 } 2822 if (flow.getViews() != null) { 2823 physicalTables.addAll(flow.getViews()); 2824 } 2825 } 2826 return physicalTables; 2827 } 2828 2829 /** 2830 * 将 newTable 的属性合并到 existingTable 上(已有表 ID 保持不变)。 2831 */ 2832 private static void mergeTableProperties(table existingTable, table newTable) { 2833 GroupNames chosen = groupNames(Arrays.asList(existingTable, newTable)); 2834 table names = chosen.names; 2835 Map<String, String> published = new LinkedHashMap<String, String>(); 2836 addCarriedPublishedNames(published, existingTable); 2837 addCarriedPublishedNames(published, newTable); 2838 if (names != existingTable) { 2839 // The name choice a batch merge makes (groupNames): a join across legacy folded names publishes 2840 // the first names a minting model published. 2841 existingTable.setServer(names.getServer()); 2842 existingTable.setDatabase(names.getDatabase()); 2843 existingTable.setSchema(names.getSchema()); 2844 existingTable.setName(names.getName()); 2845 existingTable.setDisplayName(names.getDisplayName()); 2846 existingTable.setPublishedNameSqlSpelling(names.isPublishedNameSqlSpelling()); 2847 existingTable.setMintedNames(null); 2848 } else if (!hasMintedNames(existingTable) && existingTable.getMintedNames() == null) { 2849 existingTable.setMintedNames(firstMintedNames(Collections.singletonList(newTable))); 2850 } 2851 existingTable.setAcrossLegacyNames(chosen.acrossLegacyNames); 2852 existingTable.setCarriedPublishedNames(carriedPublishedNamesBeside(published, existingTable)); 2853 if (existingTable.getMergeIdentity() == null) { 2854 existingTable.setMergeIdentity(newTable.getMergeIdentity()); 2855 } 2856 if (!SQLUtil.isEmpty(newTable.getCoordinate())) { 2857 if (SQLUtil.isEmpty(existingTable.getCoordinate())) { 2858 existingTable.setCoordinate(newTable.getCoordinate()); 2859 } else if (existingTable.getCoordinate().indexOf(newTable.getCoordinate()) == -1) { 2860 existingTable.appendCoordinate(newTable.getCoordinate()); 2861 } 2862 } 2863 if (!SQLUtil.isEmpty(newTable.getAlias())) { 2864 if (SQLUtil.isEmpty(existingTable.getAlias())) { 2865 existingTable.setAlias(newTable.getAlias()); 2866 } else if (!existingTable.getAlias().contains(newTable.getAlias())) { 2867 existingTable.setAlias(existingTable.getAlias() + "," + newTable.getAlias()); 2868 } 2869 } 2870 if (newTable.getProcessIds() != null) { 2871 if (existingTable.getProcessIds() == null) { 2872 existingTable.setProcessIds(new ArrayList<>(newTable.getProcessIds())); 2873 } else { 2874 Set<String> merged = new LinkedHashSet<>(existingTable.getProcessIds()); 2875 merged.addAll(newTable.getProcessIds()); 2876 existingTable.setProcessIds(new ArrayList<>(merged)); 2877 } 2878 } 2879 if (newTable.getProcedureId() != null && existingTable.getProcedureId() == null) { 2880 existingTable.setProcedureId(newTable.getProcedureId()); 2881 } 2882 mergeEndpointClassificationPair(existingTable, newTable); 2883 } 2884 2885 /** Canonical outermost-first order of {@code templatedParts} labels. */ 2886 private static final String[] TEMPLATED_PART_ORDER = { "server", "db", "schema", "name" }; 2887 2888 private static final String DYNAMIC_TEMPLATE_KIND = "DYNAMIC_TEMPLATE"; 2889 2890 /** 2891 * Carries the authoritative endpoint classification through a same-name merge instead of 2892 * silently dropping it (docs/tmp/dynamic-template-endpoint-provenance.md, merge finding). 2893 * The name-derived facts (kind, normalizedName) and the create-effect facts 2894 * (createdInSql, endpointIntroduction) survive as "any occurrence states it". 2895 * 2896 * <p>Template provenance is stricter, and "templated" means {@code templatedParts} is 2897 * present — NOT {@code endpointKind == DYNAMIC_TEMPLATE}, because a templated literal-temp 2898 * ({@code [#t_' + @X + ']}) keeps its temp kind while carrying parts. Parts survive only 2899 * when EVERY occurrence is templated; one plain occurrence means a STATIC statement also 2900 * references this name, so the merged node keeps the PLAIN occurrence's kind (making the 2901 * outcome independent of merge order and merge API) and drops the template claim — the 2902 * shared-node contract. 2903 */ 2904 private static void mergeEndpointClassification(table merged, List<table> occurrences) { 2905 boolean created = false; 2906 String introduction = null; 2907 String normalized = null; 2908 String templatedKind = null; 2909 String plainKind = null; 2910 boolean anyTemplate = false; 2911 boolean allTemplate = true; 2912 Set<String> parts = new LinkedHashSet<String>(); 2913 for (table occurrence : occurrences) { 2914 // non-identifier-compare: fixed enum-name/flag wire values, not DB object names 2915 if ("true".equals(occurrence.getCreatedInSql())) { 2916 created = true; 2917 } 2918 if (introduction == null) { 2919 introduction = occurrence.getEndpointIntroduction(); 2920 } 2921 if (normalized == null) { 2922 normalized = occurrence.getNormalizedName(); 2923 } 2924 if (occurrence.getTemplatedParts() != null) { 2925 anyTemplate = true; 2926 addTemplatedParts(parts, occurrence.getTemplatedParts()); 2927 if (templatedKind == null) { 2928 templatedKind = occurrence.getEndpointKind(); 2929 } 2930 } else { 2931 allTemplate = false; 2932 if (plainKind == null) { 2933 plainKind = occurrence.getEndpointKind(); 2934 } 2935 } 2936 } 2937 if (created) { 2938 merged.setCreatedInSql("true"); 2939 } 2940 if (introduction != null) { 2941 merged.setEndpointIntroduction(introduction); 2942 } 2943 if (normalized != null) { 2944 merged.setNormalizedName(normalized); 2945 } 2946 if (anyTemplate && allTemplate) { 2947 merged.setTemplatedParts(joinTemplatedParts(parts)); 2948 if (templatedKind != null) { 2949 merged.setEndpointKind(templatedKind); 2950 } 2951 } else if (plainKind != null) { 2952 // No template, or mixed static/templated: the plain occurrence's kind stands. 2953 merged.setEndpointKind(plainKind); 2954 } 2955 } 2956 2957 /** Pairwise form of {@link #mergeEndpointClassification} for the incremental merge. */ 2958 private static void mergeEndpointClassificationPair(table existingTable, table newTable) { 2959 // non-identifier-compare: fixed enum-name/flag wire values, not DB object names 2960 if ("true".equals(newTable.getCreatedInSql()) && existingTable.getCreatedInSql() == null) { 2961 existingTable.setCreatedInSql("true"); 2962 } 2963 if (existingTable.getEndpointIntroduction() == null) { 2964 existingTable.setEndpointIntroduction(newTable.getEndpointIntroduction()); 2965 } 2966 if (existingTable.getNormalizedName() == null) { 2967 existingTable.setNormalizedName(newTable.getNormalizedName()); 2968 } 2969 boolean existingTemplate = existingTable.getTemplatedParts() != null; 2970 boolean incomingTemplate = newTable.getTemplatedParts() != null; 2971 if (existingTemplate && incomingTemplate) { 2972 Set<String> parts = new LinkedHashSet<String>(); 2973 addTemplatedParts(parts, existingTable.getTemplatedParts()); 2974 addTemplatedParts(parts, newTable.getTemplatedParts()); 2975 existingTable.setTemplatedParts(joinTemplatedParts(parts)); 2976 if (existingTable.getEndpointKind() == null) { 2977 existingTable.setEndpointKind(newTable.getEndpointKind()); 2978 } 2979 } else if (existingTemplate) { 2980 // A plain occurrence of the same name arrived: mixed static/templated. The 2981 // template claim is withdrawn and the PLAIN side's kind stands — the same 2982 // outcome the reverse arrival order and the batch merge produce. Monotone: 2983 // once withdrawn, a later templated occurrence stays out. 2984 existingTable.setTemplatedParts(null); 2985 existingTable.setEndpointKind(newTable.getEndpointKind()); 2986 } else if (incomingTemplate) { 2987 // Mixed, other order: keep the existing (plain) side's kind; never adopt the 2988 // templated side's kind or parts. 2989 } else if (existingTable.getEndpointKind() == null) { 2990 existingTable.setEndpointKind(newTable.getEndpointKind()); 2991 } 2992 } 2993 2994 private static void addTemplatedParts(Set<String> parts, String joined) { 2995 if (SQLUtil.isEmpty(joined)) { 2996 return; 2997 } 2998 for (String part : joined.split("\\|")) { 2999 if (!part.isEmpty()) { 3000 parts.add(part); 3001 } 3002 } 3003 } 3004 3005 private static String joinTemplatedParts(Set<String> parts) { 3006 StringBuilder sb = new StringBuilder(); 3007 for (String part : TEMPLATED_PART_ORDER) { 3008 if (parts.contains(part)) { 3009 if (sb.length() > 0) { 3010 sb.append('|'); 3011 } 3012 sb.append(part); 3013 } 3014 } 3015 return sb.length() == 0 ? null : sb.toString(); 3016 } 3017 3018 /** 3019 * 将 newTable 的列合并到 existingTable 上(已有列 ID 保持不变)。 3020 */ 3021 private static void mergeColumnsIncremental(table existingTable, table newTable, 3022 Map<String, String> columnIdMap, 3023 Map<String, column> columnMergeIdMap, 3024 Map<String, Set<String>> identityColumnNames) { 3025 if (newTable.getColumns() == null) { 3026 return; 3027 } 3028 if (existingTable.getColumns() == null) { 3029 existingTable.setColumns(new ArrayList<column>()); 3030 } 3031 3032 // A named column with an identity joins the column of that identity whatever its name; any other column 3033 // joins the column of its name and role that carries no identity. Whether a column without an identity is 3034 // the column of the one identity published under its name is decided once every file is merged 3035 // (joinColumnsWithoutIdentity): deciding it here would depend on the order the files arrive in. A star, a 3036 // system column and a named column are never one column (SQL Server [*] is a column, not the star). 3037 Map<List<String>, column> existingColMap = new LinkedHashMap<>(); 3038 Map<CanonKey, column> existingByIdentity = new HashMap<>(); 3039 for (column c : existingTable.getColumns()) { 3040 if (NAMED_ROLE.equals(roleOf(c)) && c.getIdentifierKey() != null) { 3041 if (!existingByIdentity.containsKey(c.getIdentifierKey())) { 3042 existingByIdentity.put(c.getIdentifierKey(), c); 3043 } 3044 } else { 3045 existingColMap.put(Arrays.asList(roleOf(c), c.getName()), c); 3046 } 3047 } 3048 3049 for (column newCol : newTable.getColumns()) { 3050 if (Boolean.TRUE.toString().equals(newCol.getIsFunction())) { 3051 // function 列直接追加,不做合并 3052 existingTable.getColumns().add(newCol); 3053 columnIdMap.put(newCol.getId(), newCol.getId()); 3054 columnMergeIdMap.put(newCol.getId(), newCol); 3055 continue; 3056 } 3057 3058 String role = roleOf(newCol); 3059 CanonKey identity = NAMED_ROLE.equals(role) ? newCol.getIdentifierKey() : null; 3060 column existingCol = identity != null ? existingByIdentity.get(identity) 3061 : existingColMap.get(Arrays.asList(role, newCol.getName())); 3062 if (existingCol != null) { 3063 // 同名列:合并到已有列,已有列 ID 保持不变 3064 if (identity != null) { 3065 namesOf(identityColumnNames, existingCol).add(newCol.getName()); 3066 } 3067 columnIdMap.put(newCol.getId(), existingCol.getId()); 3068 if (existingCol.getSqlSpelling() == null) { 3069 existingCol.setSqlSpelling(newCol.getSqlSpelling()); 3070 } 3071 if (!SQLUtil.isEmpty(newCol.getCoordinate())) { 3072 if (SQLUtil.isEmpty(existingCol.getCoordinate())) { 3073 existingCol.setCoordinate(newCol.getCoordinate()); 3074 } else if (existingCol.getCoordinate().indexOf(newCol.getCoordinate()) == -1) { 3075 existingCol.appendCoordinate(newCol.getCoordinate()); 3076 } 3077 } 3078 takeAttributes(existingCol, newCol); 3079 } else { 3080 // 新列:直接追加 3081 existingTable.getColumns().add(newCol); 3082 if (identity != null) { 3083 existingByIdentity.put(identity, newCol); 3084 namesOf(identityColumnNames, newCol); 3085 } else { 3086 existingColMap.put(Arrays.asList(role, newCol.getName()), newCol); 3087 } 3088 columnIdMap.put(newCol.getId(), newCol.getId()); 3089 columnMergeIdMap.put(newCol.getId(), newCol); 3090 } 3091 } 3092 } 3093 3094 3095 /** 3096 * 将 table 按类型追加到 mergedDataflow 对应的列表中。 3097 */ 3098 private static void addTableToMergedDataflow(dataflow df, table t) { 3099 if (t.isView()) { 3100 if (df.getViews() == null) df.setViews(new ArrayList<table>()); 3101 df.getViews().add(t); 3102 } else if (t.isDatabaseType()) { 3103 if (df.getDatabases() == null) df.setDatabases(new ArrayList<table>()); 3104 df.getDatabases().add(t); 3105 } else if (t.isSchemaType()) { 3106 if (df.getSchemas() == null) df.setSchemas(new ArrayList<table>()); 3107 df.getSchemas().add(t); 3108 } else if (t.isStage()) { 3109 if (df.getStages() == null) df.setStages(new ArrayList<table>()); 3110 df.getStages().add(t); 3111 } else if (t.isDataSource()) { 3112 if (df.getDatasources() == null) df.setDatasources(new ArrayList<table>()); 3113 df.getDatasources().add(t); 3114 } else if (t.isStream()) { 3115 if (df.getStreams() == null) df.setStreams(new ArrayList<table>()); 3116 df.getStreams().add(t); 3117 } else if (t.isFile()) { 3118 if (df.getPaths() == null) df.setPaths(new ArrayList<table>()); 3119 df.getPaths().add(t); 3120 } else if (t.isVariable()) { 3121 if (df.getVariables() == null) df.setVariables(new ArrayList<table>()); 3122 df.getVariables().add(t); 3123 } else if (t.isResultSet()) { 3124 if (df.getResultsets() == null) df.setResultsets(new ArrayList<table>()); 3125 df.getResultsets().add(t); 3126 } else if (t.isSequence()) { 3127 if (df.getSequences() == null) df.setSequences(new ArrayList<table>()); 3128 df.getSequences().add(t); 3129 } else { 3130 if (df.getTables() == null) df.setTables(new ArrayList<table>()); 3131 df.getTables().add(t); 3132 } 3133 } 3134 3135 /** 3136 * 将 currentDataflow 的非表实体追加到 mergedDataflow。 3137 */ 3138 private static void appendOtherEntities(dataflow mergedDataflow, dataflow currentDataflow) { 3139 if (currentDataflow.getProcesses() != null) { 3140 if (mergedDataflow.getProcesses() == null) { 3141 mergedDataflow.setProcesses(new ArrayList<process>()); 3142 } 3143 mergedDataflow.getProcesses().addAll(currentDataflow.getProcesses()); 3144 } 3145 if (currentDataflow.getProcedures() != null) { 3146 if (mergedDataflow.getProcedures() == null) { 3147 mergedDataflow.setProcedures(new ArrayList<procedure>()); 3148 } 3149 mergedDataflow.getProcedures().addAll(currentDataflow.getProcedures()); 3150 } 3151 if (currentDataflow.getPackages() != null) { 3152 if (mergedDataflow.getPackages() == null) { 3153 mergedDataflow.setPackages(new ArrayList<oraclePackage>()); 3154 } 3155 mergedDataflow.getPackages().addAll(currentDataflow.getPackages()); 3156 } 3157 if (currentDataflow.getErrors() != null) { 3158 if (mergedDataflow.getErrors() == null) { 3159 mergedDataflow.setErrors(new ArrayList<error>()); 3160 } 3161 mergedDataflow.getErrors().addAll(currentDataflow.getErrors()); 3162 if (mergedDataflow.getErrors().size() > 10000) { 3163 mergedDataflow.setErrors(mergedDataflow.getErrors().subList(0, 10000)); 3164 } 3165 } 3166 } 3167 3168 /** 3169 * 基于合并后的 dataflow 构建 function→resultSet 映射。 3170 */ 3171 private static void buildFunctionResultSetMapping(dataflow df, 3172 Map<String, String> functionIdToResultSetId, 3173 Map<String, column> funcColIdToResultSetCol, 3174 Map<String, table> allTableIdMap) { 3175 // 收集所有 procedure 3176 List<procedure> allProcedures = new ArrayList<>(); 3177 if (df.getProcedures() != null) { 3178 allProcedures.addAll(df.getProcedures()); 3179 } 3180 if (df.getPackages() != null) { 3181 for (oraclePackage pkg : df.getPackages()) { 3182 if (pkg.getProcedures() != null) { 3183 allProcedures.addAll(pkg.getProcedures()); 3184 } 3185 } 3186 } 3187 3188 // the result set of each routine, by the routine's identity (routineKey) 3189 Map<Object, String> procedureNameToResultSetId = new HashMap<>(); 3190 List<table> allTables = dataflow.getAllTables(df); 3191 for (table t : allTables) { 3192 if (t.getProcedureId() != null && !t.isFunction()) { 3193 Object procName = null; 3194 for (procedure proc : allProcedures) { 3195 if (proc.getId().equals(t.getProcedureId())) { 3196 procName = routineKeyOf(proc); 3197 break; 3198 } 3199 } 3200 if (procName != null) { 3201 procedureNameToResultSetId.put(procName, t.getId()); 3202 } 3203 } 3204 } 3205 3206 // 构建 allTableIdMap 3207 for (table t : allTables) { 3208 allTableIdMap.put(t.getId(), t); 3209 } 3210 3211 // 构建 functionId → resultSetId 映射 3212 for (table t : allTables) { 3213 if (t.isFunction()) { 3214 String resultSetId = procedureNameToResultSetId.get(calledRoutineKey(t)); 3215 if (resultSetId != null) { 3216 functionIdToResultSetId.put(t.getId(), resultSetId); 3217 table resultSet = allTableIdMap.get(resultSetId); 3218 if (resultSet != null && t.getColumns() != null && resultSet.getColumns() != null) { 3219 List<column> funcCols = t.getColumns(); 3220 List<column> rsCols = resultSet.getColumns(); 3221 int minSize = Math.min(funcCols.size(), rsCols.size()); 3222 for (int j = 0; j < minSize; j++) { 3223 funcColIdToResultSetCol.put(funcCols.get(j).getId(), rsCols.get(j)); 3224 } 3225 } 3226 } 3227 } 3228 } 3229 } 3230 3231 /** 3232 * 对单条关系做 ID 重映射和去重,如果未重复则加入 newRelations 列表。 3233 */ 3234 private static void remapAndDedupRelation(relationship rel, 3235 Map<String, String> tableIdMap, 3236 Map<String, String> columnIdMap, 3237 Map<String, column> columnMergeIdMap, 3238 Map<String, String> functionIdToResultSetId, 3239 Map<String, column> funcColIdToResultSetCol, 3240 Map<String, table> allTableIdMap, 3241 Set<Long> mergedDedupKeys, 3242 Set<Long> ownKeys, 3243 Set<Long> ownPublishedKeys, 3244 List<relationship> newRelations) { 3245 if (RelationshipType.call.name().equals(rel.getType())) { 3246 // call 类型关系 3247 targetColumn caller = rel.getCaller(); 3248 if (caller == null) { 3249 return; 3250 } 3251 String mappedCallerId = tableIdMap.get(caller.getId()); 3252 if (mappedCallerId != null) { 3253 caller.setId(mappedCallerId); 3254 } 3255 3256 List<sourceColumn> callees = rel.getCallees(); 3257 if (callees != null) { 3258 Set<sourceColumn> calleeSet = new LinkedHashSet<>(); 3259 for (sourceColumn callee : callees) { 3260 String mappedSourceId = tableIdMap.get(callee.getId()); 3261 if (mappedSourceId != null) { 3262 callee.setId(mappedSourceId); 3263 } 3264 } 3265 calleeSet.addAll(callees); 3266 rel.setCallees(new ArrayList<>(calleeSet)); 3267 } 3268 3269 if (mergedDedupKeys.add(rel.toDedupHash())) { 3270 newRelations.add(rel); 3271 } 3272 } else { 3273 // fdd/frd 等类型关系 3274 targetColumn target = rel.getTarget(); 3275 if (target == null) { 3276 return; 3277 } 3278 String mappedTargetParentId = tableIdMap.get(target.getParent_id()); 3279 if (mappedTargetParentId != null) { 3280 target.setParent_id(mappedTargetParentId); 3281 } 3282 String mappedTargetColId = columnIdMap.get(target.getId()); 3283 if (mappedTargetColId != null) { 3284 target.setId(mappedTargetColId); 3285 column mergedCol = columnMergeIdMap.get(mappedTargetColId); 3286 if (mergedCol != null) { 3287 target.setCoordinate(mergedCol.getCoordinate()); 3288 } 3289 } 3290 3291 List<sourceColumn> sources = rel.getSources(); 3292 // Same contract as DataFlowAnalyzer.mergeTables: dedup runs under the 3293 // merged column's stamped coordinate, but a source's own usable 3294 // reference-site coordinate is restored on the surviving relation. 3295 Map<sourceColumn, String> referenceCoordinates = new IdentityHashMap<>(); 3296 if (sources != null) { 3297 Set<sourceColumn> sourceSet = new LinkedHashSet<>(); 3298 for (sourceColumn source : sources) { 3299 String mappedParentId = tableIdMap.get(source.getParent_id()); 3300 if (mappedParentId != null) { 3301 source.setParent_id(mappedParentId); 3302 } 3303 String mappedSourceId = tableIdMap.get(source.getSource_id()); 3304 if (mappedSourceId != null) { 3305 source.setSource_id(mappedSourceId); 3306 } 3307 String originalParentId = source.getParent_id(); 3308 if (!functionIdToResultSetId.isEmpty()) { 3309 String resultSetId = functionIdToResultSetId.get(originalParentId); 3310 if (resultSetId != null) { 3311 source.setParent_id(resultSetId); 3312 column matched = funcColIdToResultSetCol.get(source.getId()); 3313 if (matched != null) { 3314 source.setId(matched.getId()); 3315 source.setColumn(matched.getName()); 3316 } 3317 table resultSetTable = allTableIdMap.get(source.getParent_id()); 3318 if (resultSetTable != null) { 3319 source.setParent_name(resultSetTable.getName()); 3320 } 3321 } 3322 } 3323 String mappedColId = columnIdMap.get(source.getId()); 3324 if (mappedColId != null) { 3325 source.setId(mappedColId); 3326 column mergedCol = columnMergeIdMap.get(mappedColId); 3327 if (mergedCol != null) { 3328 String ownCoordinate = source.getCoordinate(); 3329 source.setCoordinate(mergedCol.getCoordinate()); 3330 if (DlineageUtil.hasUsableCoordinate(ownCoordinate)) { 3331 referenceCoordinates.put(source, ownCoordinate); 3332 } 3333 } 3334 } 3335 } 3336 sourceSet.addAll(sources); 3337 rel.setSources(new ArrayList<>(sourceSet)); 3338 } 3339 3340 long key = rel.toDedupHash(); 3341 if (mergedDedupKeys.add(key)) { 3342 ownKeys.add(key); 3343 for (Map.Entry<sourceColumn, String> entry : referenceCoordinates.entrySet()) { 3344 entry.getKey().setCoordinate(entry.getValue()); 3345 } 3346 ownPublishedKeys.add(SQLUtil.hash64(exactKey(rel))); 3347 newRelations.add(rel); 3348 } else if (ownKeys.contains(key)) { 3349 // this dataflow published both, from their own reference sites: they stay two relationships unless 3350 // they repeat each other exactly (sameEdgeOfOneInput); against an earlier dataflow, its one is kept 3351 for (Map.Entry<sourceColumn, String> entry : referenceCoordinates.entrySet()) { 3352 entry.getKey().setCoordinate(entry.getValue()); 3353 } 3354 if (ownPublishedKeys.add(SQLUtil.hash64(exactKey(rel)))) { 3355 newRelations.add(rel); 3356 } 3357 } 3358 } 3359 } 3360 3361 /** 3362 * 只对指定表做 * 列桥接,避免全量扫描。 3363 */ 3364 private static long addStarColumnBridgingRelationsForTables(dataflow df, long startId, 3365 Set<String> targetTableIds) { 3366 if (df == null || targetTableIds.isEmpty()) { 3367 return startId; 3368 } 3369 3370 // 构建已有桥接关系索引 3371 Map<String, Set<String>> bridgeIndex = new HashMap<>(); 3372 if (df.getRelationships() != null) { 3373 for (relationship rel : df.getRelationships()) { 3374 if (!RelationshipType.fdd.name().equals(rel.getType())) { 3375 continue; 3376 } 3377 if (rel.getTarget() == null || rel.getSources() == null) { 3378 continue; 3379 } 3380 String parentId = rel.getTarget().getParent_id(); 3381 if (parentId == null) { 3382 continue; 3383 } 3384 for (sourceColumn sc : rel.getSources()) { 3385 if (parentId.equals(sc.getParent_id())) { 3386 bridgeIndex.computeIfAbsent(parentId, k -> new HashSet<>()) 3387 .add(rel.getTarget().getId() + "<-" + sc.getId()); 3388 } 3389 } 3390 } 3391 } 3392 3393 List<relationship> newRelations = new ArrayList<>(); 3394 for (table t : dataflow.getAllTables(df)) { 3395 if (!targetTableIds.contains(t.getId())) { 3396 continue; 3397 } 3398 if (t.getColumns() == null || t.getColumns().size() < 2) { 3399 continue; 3400 } 3401 column starColumn = null; 3402 List<column> regularColumns = new ArrayList<>(); 3403 for (column c : t.getColumns()) { 3404 if ("system".equals(c.getSource())) { 3405 continue; 3406 } 3407 if (Boolean.TRUE.toString().equals(c.getIsFunction())) { 3408 continue; 3409 } 3410 if (c.standsForStar()) { 3411 if (starColumn == null) { 3412 starColumn = c; 3413 } 3414 } else { 3415 regularColumns.add(c); 3416 } 3417 } 3418 if (starColumn == null || regularColumns.isEmpty()) { 3419 continue; 3420 } 3421 3422 Set<String> existingBridges = bridgeIndex.getOrDefault(t.getId(), Collections.emptySet()); 3423 String starId = starColumn.getId(); 3424 for (column target : regularColumns) { 3425 if (existingBridges.contains(target.getId() + "<-" + starId)) { 3426 continue; 3427 } 3428 relationship rel = new relationship(); 3429 rel.setId(String.valueOf(++startId)); 3430 rel.setType(RelationshipType.fdd.name()); 3431 rel.setEffectType("expand_star"); 3432 3433 targetColumn tc = new targetColumn(); 3434 tc.setId(target.getId()); 3435 tc.setColumn(target.getName()); 3436 tc.setParent_id(t.getId()); 3437 tc.setParent_name(t.getName()); 3438 rel.setTarget(tc); 3439 3440 sourceColumn sc = new sourceColumn(); 3441 sc.setId(starId); 3442 sc.setColumn(starColumn.getName()); 3443 sc.setParent_id(t.getId()); 3444 sc.setParent_name(t.getName()); 3445 rel.setSources(Arrays.asList(sc)); 3446 3447 newRelations.add(rel); 3448 } 3449 } 3450 3451 if (!newRelations.isEmpty()) { 3452 if (df.getRelationships() == null) { 3453 df.setRelationships(new ArrayList<relationship>()); 3454 } 3455 df.getRelationships().addAll(newRelations); 3456 } 3457 return startId; 3458 } 3459 3460 private static String offsetId(String id, long offset) { 3461 if (id == null || id.isEmpty()) { 3462 return id; 3463 } 3464 try { 3465 return String.valueOf(Long.parseLong(id) + offset); 3466 } catch (NumberFormatException ignore) { 3467 return id; 3468 } 3469 } 3470 3471 /** 3472 * 偏移 coordinate 第三维(fileIdx)。 3473 * <p> 3474 * coordinate 字符串格式为 {@code [line,col,fileIdx],[line,col,fileIdx]}, 3475 * 合并多个独立生成的 dataflow 时,它们的 fileIdx 都从 0 起始会互相冲突, 3476 * 通过给第三维加上 offset 来区分不同的来源。 3477 */ 3478 private static String offsetCoordinate(String coordinate, long offset) { 3479 if (coordinate == null || coordinate.isEmpty() || offset == 0L) { 3480 return coordinate; 3481 } 3482 StringBuilder sb = new StringBuilder(coordinate.length() + 8); 3483 int i = 0; 3484 int len = coordinate.length(); 3485 while (i < len) { 3486 int open = coordinate.indexOf('[', i); 3487 if (open < 0) { 3488 sb.append(coordinate, i, len); 3489 break; 3490 } 3491 int close = coordinate.indexOf(']', open); 3492 if (close < 0) { 3493 sb.append(coordinate, i, len); 3494 break; 3495 } 3496 sb.append(coordinate, i, open + 1); 3497 String inner = coordinate.substring(open + 1, close); 3498 String[] parts = inner.split(",", -1); 3499 for (int k = 0; k < parts.length; k++) { 3500 if (k > 0) sb.append(','); 3501 if (k == 2) { 3502 String p = parts[k].trim(); 3503 try { 3504 sb.append(Long.parseLong(p) + offset); 3505 } catch (NumberFormatException ignore) { 3506 sb.append(parts[k]); 3507 } 3508 } else { 3509 sb.append(parts[k]); 3510 } 3511 } 3512 sb.append(']'); 3513 i = close + 1; 3514 } 3515 return sb.toString(); 3516 } 3517 3518 /** 3519 * 偏移 dataflow 中所有节点的 ID 和 coordinate 第三维 fileIdx。 3520 * 3521 * @param df 待偏移的 dataflow(原地修改) 3522 * @param idOffset 给所有 id / xxxId 字段加上的偏移量 3523 * @param fileIdxOffset 给所有 coordinate 第三维 (fileIdx) 加上的偏移量, 3524 * 通常为合并时前面所有 dataflow 累积的源文件数, 3525 * 以便合并后不同来源的 fileIdx 保持连续且不重叠。 3526 */ 3527 private static void offsetDataflowIds(dataflow df, long idOffset, long fileIdxOffset) { 3528 if (idOffset == 0L && fileIdxOffset == 0L) { 3529 return; 3530 } 3531 3532 CommentBindingProjection.offset(df, id -> offsetId(id, idOffset)); 3533 // tables / views / databases / schemas / stages / datasources / streams / paths / variables / resultsets / sequences 3534 for (table t : dataflow.getAllTables(df)) { 3535 offsetTableIds(t, idOffset); 3536 t.setCoordinate(offsetCoordinate(t.getCoordinate(), fileIdxOffset)); 3537 if (t.getColumns() != null) { 3538 for (column c : t.getColumns()) { 3539 c.setCoordinate(offsetCoordinate(c.getCoordinate(), fileIdxOffset)); 3540 } 3541 } 3542 } 3543 3544 // processes 3545 if (df.getProcesses() != null) { 3546 for (process p : df.getProcesses()) { 3547 p.setId(offsetId(p.getId(), idOffset)); 3548 p.setProcedureId(offsetId(p.getProcedureId(), idOffset)); 3549 p.setCoordinate(offsetCoordinate(p.getCoordinate(), fileIdxOffset)); 3550 } 3551 } 3552 3553 // procedures 3554 if (df.getProcedures() != null) { 3555 for (procedure p : df.getProcedures()) { 3556 p.setId(offsetId(p.getId(), idOffset)); 3557 p.setCoordinate(offsetCoordinate(p.getCoordinate(), fileIdxOffset)); 3558 if (p.getArguments() != null) { 3559 for (argument arg : p.getArguments()) { 3560 arg.setId(offsetId(arg.getId(), idOffset)); 3561 arg.setCoordinate(offsetCoordinate(arg.getCoordinate(), fileIdxOffset)); 3562 } 3563 } 3564 } 3565 } 3566 3567 // oracle packages 3568 if (df.getPackages() != null) { 3569 for (oraclePackage pkg : df.getPackages()) { 3570 pkg.setId(offsetId(pkg.getId(), idOffset)); 3571 if (pkg.getProcedures() != null) { 3572 for (procedure pp : pkg.getProcedures()) { 3573 pp.setId(offsetId(pp.getId(), idOffset)); 3574 pp.setCoordinate(offsetCoordinate(pp.getCoordinate(), fileIdxOffset)); 3575 if (pp.getArguments() != null) { 3576 for (argument arg : pp.getArguments()) { 3577 arg.setId(offsetId(arg.getId(), idOffset)); 3578 arg.setCoordinate(offsetCoordinate(arg.getCoordinate(), fileIdxOffset)); 3579 } 3580 } 3581 } 3582 } 3583 } 3584 } 3585 3586 // relationships 3587 if (df.getRelationships() != null) { 3588 for (relationship rel : df.getRelationships()) { 3589 rel.setId(offsetId(rel.getId(), idOffset)); 3590 rel.setProcessId(offsetId(rel.getProcessId(), idOffset)); 3591 rel.setProcedureId(offsetId(rel.getProcedureId(), idOffset)); 3592 if (rel.getTarget() != null) { 3593 offsetTargetColumnIds(rel.getTarget(), idOffset); 3594 } 3595 if (rel.getCaller() != null) { 3596 offsetTargetColumnIds(rel.getCaller(), idOffset); 3597 } 3598 if (rel.getSources() != null) { 3599 for (sourceColumn sc : rel.getSources()) { 3600 offsetSourceColumnIds(sc, idOffset); 3601 } 3602 } 3603 if (rel.getCallees() != null) { 3604 for (sourceColumn sc : rel.getCallees()) { 3605 offsetSourceColumnIds(sc, idOffset); 3606 } 3607 } 3608 } 3609 } 3610 } 3611 3612 /** 3613 * 扫描 dataflow 中所有 coordinate,返回最大的第三维 (fileIdx) 值; 3614 * 若没有任何 coordinate 则返回 -1。 3615 */ 3616 private static int findMaxFileIdx(dataflow df) { 3617 if (df == null) { 3618 return -1; 3619 } 3620 int max = -1; 3621 for (table t : dataflow.getAllTables(df)) { 3622 max = Math.max(max, maxFileIdxInCoordinate(t.getCoordinate())); 3623 if (t.getColumns() != null) { 3624 for (column c : t.getColumns()) { 3625 max = Math.max(max, maxFileIdxInCoordinate(c.getCoordinate())); 3626 } 3627 } 3628 } 3629 if (df.getProcesses() != null) { 3630 for (process p : df.getProcesses()) { 3631 max = Math.max(max, maxFileIdxInCoordinate(p.getCoordinate())); 3632 } 3633 } 3634 if (df.getProcedures() != null) { 3635 for (procedure p : df.getProcedures()) { 3636 max = Math.max(max, maxFileIdxInCoordinate(p.getCoordinate())); 3637 if (p.getArguments() != null) { 3638 for (argument arg : p.getArguments()) { 3639 max = Math.max(max, maxFileIdxInCoordinate(arg.getCoordinate())); 3640 } 3641 } 3642 } 3643 } 3644 if (df.getPackages() != null) { 3645 for (oraclePackage pkg : df.getPackages()) { 3646 if (pkg.getProcedures() != null) { 3647 for (procedure pp : pkg.getProcedures()) { 3648 max = Math.max(max, maxFileIdxInCoordinate(pp.getCoordinate())); 3649 if (pp.getArguments() != null) { 3650 for (argument arg : pp.getArguments()) { 3651 max = Math.max(max, maxFileIdxInCoordinate(arg.getCoordinate())); 3652 } 3653 } 3654 } 3655 } 3656 } 3657 } 3658 return max; 3659 } 3660 3661 /** 3662 * 从 coordinate 字符串中解析出最大的第三维 (fileIdx) 值; 3663 * 若 coordinate 为空或没有任何可解析的数值则返回 -1。 3664 */ 3665 private static int maxFileIdxInCoordinate(String coordinate) { 3666 if (coordinate == null || coordinate.isEmpty()) { 3667 return -1; 3668 } 3669 int max = -1; 3670 int i = 0; 3671 int len = coordinate.length(); 3672 while (i < len) { 3673 int open = coordinate.indexOf('[', i); 3674 if (open < 0) break; 3675 int close = coordinate.indexOf(']', open); 3676 if (close < 0) break; 3677 String inner = coordinate.substring(open + 1, close); 3678 String[] parts = inner.split(",", -1); 3679 if (parts.length >= 3) { 3680 try { 3681 int v = Integer.parseInt(parts[2].trim()); 3682 if (v > max) max = v; 3683 } catch (NumberFormatException ignore) { 3684 // 不是数字,跳过 3685 } 3686 } 3687 i = close + 1; 3688 } 3689 return max; 3690 } 3691 3692 private static void offsetTableIds(table t, long offset) { 3693 if (t == null) { 3694 return; 3695 } 3696 t.setId(offsetId(t.getId(), offset)); 3697 if (t.getProcessIds() != null) { 3698 List<String> newProcessIds = new ArrayList<>(t.getProcessIds().size()); 3699 for (String pid : t.getProcessIds()) { 3700 newProcessIds.add(offsetId(pid, offset)); 3701 } 3702 t.setProcessIds(newProcessIds); 3703 } 3704 if (t.getColumns() != null) { 3705 for (column c : t.getColumns()) { 3706 c.setId(offsetId(c.getId(), offset)); 3707 } 3708 } 3709 } 3710 3711 private static void offsetTargetColumnIds(targetColumn col, long offset) { 3712 col.setId(offsetId(col.getId(), offset)); 3713 col.setParent_id(offsetId(col.getParent_id(), offset)); 3714 col.setTarget_id(offsetId(col.getTarget_id(), offset)); 3715 } 3716 3717 private static void offsetSourceColumnIds(sourceColumn col, long offset) { 3718 col.setId(offsetId(col.getId(), offset)); 3719 col.setParent_id(offsetId(col.getParent_id(), offset)); 3720 col.setSource_id(offsetId(col.getSource_id(), offset)); 3721 } 3722 3723 public static dataflow readDataflowFromCsvMetadata(String csvMetadata, EDbVendor vendor) { 3724 if (!MetadataReader.isMetadata(csvMetadata)) { 3725 throw new IllegalArgumentException("Illegal csv metadata."); 3726 } 3727 return new SQLDepMetadataAnalyzer().analyzeMetadata(vendor, csvMetadata); 3728 } 3729 3730 public static Dataflow mergeDataflowsAndCsv(List<Pair<EDbVendor,dataflow>> pairs, String csvMetadata) { 3731 //重置ID防止重复 3732 Long index = 0L; 3733 List<CommentBindingBundle> commentBundles = null; 3734 for(int i=0; i<pairs.size(); i++){ 3735 dataflow dataflow = pairs.get(i).second; 3736 Map<String, table> objIDMap = getDataflowDbObjMap(dataflow); 3737 Map<String, String> idMaps = new HashMap<>(); 3738 Map<String,String> commentRelations = dataflow.getCommentBindings()==null ? null : new HashMap<String,String>(); 3739 for (Map.Entry<String, table> entry : objIDMap.entrySet()) { 3740 index = index + 1; 3741 String id = index.toString(); 3742 idMaps.put(entry.getKey(), id); 3743 table table = entry.getValue(); 3744 table.setId(id); 3745 if(table.getColumns() != null){ 3746 for(column col: table.getColumns()){ 3747 index = index + 1; 3748 id = index.toString(); 3749 idMaps.put(col.getId(), id); 3750 col.setId(id); 3751 } 3752 } 3753 } 3754 if(dataflow.getProcedures() != null){ 3755 for(procedure procedure: dataflow.getProcedures()){ 3756 index = index + 1; 3757 String id = index.toString(); 3758 idMaps.put(procedure.getId(), id); 3759 procedure.setId(id); 3760 } 3761 } 3762 if(dataflow.getProcesses() != null){ 3763 for(process process: dataflow.getProcesses()){ 3764 index = index + 1; 3765 String id = index.toString(); 3766 idMaps.put(process.getId(), id); 3767 process.setId(id); 3768 } 3769 } 3770 if(dataflow.getPackages() != null){ 3771 for(oraclePackage oraclePackage: dataflow.getPackages()){ 3772 index = index + 1; 3773 String id = index.toString(); 3774 idMaps.put(oraclePackage.getId(), id); 3775 oraclePackage.setId(id); 3776 if(oraclePackage.getProcedures() != null){ 3777 for(procedure procedure: oraclePackage.getProcedures()){ 3778 index = index + 1; 3779 id = index.toString(); 3780 idMaps.put(procedure.getId(), id); 3781 procedure.setId(id); 3782 } 3783 } 3784 } 3785 } 3786 if(dataflow.getRelationships() != null && dataflow.getRelationships().size()>0){ 3787 for(relationship rel: dataflow.getRelationships()){ 3788 index = index + 1; 3789 String id = index.toString(); 3790 if (commentRelations != null) commentRelations.put(rel.getId(),id); 3791 rel.setId(id); 3792 rel.setProcedureId(idMaps.get(rel.getProcedureId())); 3793 rel.setProcessId(idMaps.get(rel.getProcessId())); 3794 rel.getTarget().setParent_id(idMaps.get(rel.getTarget().getParent_id())); 3795 rel.getTarget().setTarget_id(idMaps.get(rel.getTarget().getTarget_id())); 3796 rel.getTarget().setId(idMaps.get(rel.getTarget().getId())); 3797 for(sourceColumn column: rel.getSources()){ 3798 column.setParent_id(idMaps.get(column.getParent_id())); 3799 column.setId(idMaps.get(column.getId())); 3800 } 3801 } 3802 } 3803 if (commentRelations != null) { 3804 CommentBindingProjection.remap(dataflow, idMaps, idMaps, commentRelations); 3805 CommentBindingProjection.retain(dataflow); 3806 if (commentBundles == null) commentBundles = new ArrayList<CommentBindingBundle>(); 3807 commentBundles.add(dataflow.getCommentBindings()); 3808 } 3809 } 3810 Dataflow mDataflow = DataFlowAnalyzer.getSqlflowJSONModel(pairs.get(0).first, pairs.get(0).second, false); 3811 Map<String, table> objNameMap = getDataflowDbObjNameMap(pairs.get(0).second); 3812 Sqlflow dbobjs = mDataflow.getDbobjs(); 3813 List<Relationship> mRelationshipList = new ArrayList<>(); 3814 if(mDataflow.getRelationships() != null && mDataflow.getRelationships().length>0){ 3815 mRelationshipList = new LinkedList<>(Arrays.asList(mDataflow.getRelationships())); 3816 } 3817 3818 List<Error> mErrorList = new ArrayList<>(); 3819 if(mDataflow.getErrors() != null && mDataflow.getErrors().length>0){ 3820 mErrorList = new LinkedList<>(Arrays.asList(mDataflow.getErrors())); 3821 } 3822 3823 List<Process> mProcessList = new ArrayList<>(); 3824 if(mDataflow.getProcesses() != null && mDataflow.getProcesses().length>0){ 3825 mProcessList = new LinkedList<>(Arrays.asList(mDataflow.getProcesses())); 3826 } 3827 3828 for(int i=1; i<pairs.size(); i++){ 3829 objNameMap.putAll(getDataflowDbObjNameMap(pairs.get(i).second)); 3830 Dataflow dataflow = DataFlowAnalyzer.getSqlflowJSONModel(pairs.get(i).first, pairs.get(i).second, false); 3831 if(dataflow.getDbobjs().getServers() != null && dataflow.getDbobjs().getServers().size()>0){ 3832 if(dbobjs.getServers() == null){ 3833 dbobjs.setServers(new ArrayList<>()); 3834 } 3835 dbobjs.getServers().addAll(dataflow.getDbobjs().getServers()); 3836 } 3837 3838 if(dataflow.getDbobjs().getErrorMessages() != null && dataflow.getDbobjs().getErrorMessages().size()>0){ 3839 if(dbobjs.getErrorMessages() == null){ 3840 dbobjs.setErrorMessages(new ArrayList<>()); 3841 } 3842 dbobjs.getErrorMessages().addAll(dataflow.getDbobjs().getErrorMessages()); 3843 } 3844 3845 if(dataflow.getRelationships() != null && dataflow.getRelationships().length>0){ 3846 List<Relationship> relationshipList = new LinkedList<>(Arrays.asList(dataflow.getRelationships())); 3847 mRelationshipList.addAll(relationshipList); 3848 } 3849 3850 if(dataflow.getErrors() != null && dataflow.getErrors().length>0){ 3851 List<Error> errorList = new LinkedList<>(Arrays.asList(dataflow.getErrors())); 3852 mErrorList.addAll(errorList); 3853 } 3854 3855 if(dataflow.getProcesses() != null && dataflow.getProcesses().length>0){ 3856 List<Process> processList = new LinkedList<>(Arrays.asList(dataflow.getProcesses())); 3857 mProcessList.addAll(processList); 3858 } 3859 3860 } 3861 if(!SQLUtil.isEmpty(csvMetadata)){ 3862 /** 3863 * excel内的血缘 只合并关系 就是只做关联,如果找不到obj 就算了 3864 */ 3865 dataflow df = readDataflowFromCsvMetadata(csvMetadata); 3866 Map<String, table> objIDMap = getDataflowDbObjMap(df); 3867 if(df.getRelationships() != null && df.getRelationships().size()>0){ 3868 for(relationship rel: df.getRelationships()){ 3869 rel.setId("m"+rel.getId()); 3870 table sTable = objIDMap.get(rel.getTarget().getParent_id()); 3871 table tTable = objNameMap.get(sTable.getFullName()); 3872 rel.getTarget().setParent_id(tTable.getId()); 3873 for(column col: tTable.getColumns()){ 3874 if(col.getName().equalsIgnoreCase(rel.getTarget().getColumn())){ 3875 rel.getTarget().setId(col.getId()); 3876 break; 3877 } 3878 } 3879 for(sourceColumn column: rel.getSources()){ 3880 sTable = objIDMap.get(column.getParent_id()); 3881 tTable = objNameMap.get(sTable.getFullName()); 3882 column.setParent_id(tTable.getId()); 3883 for(column col: tTable.getColumns()){ 3884 if(col.getName().equalsIgnoreCase(column.getColumn())){ 3885 column.setId(col.getId()); 3886 break; 3887 } 3888 } 3889 } 3890 mRelationshipList.add(toRelationship(rel)); 3891 } 3892 } 3893 } 3894 3895 mDataflow.setDbobjs(dbobjs); 3896 mDataflow.setRelationships(mRelationshipList.toArray(new Relationship[mRelationshipList.size()])); 3897 mDataflow.setProcesses(mProcessList.toArray(new Process[mProcessList.size()])); 3898 mDataflow.setErrors(mErrorList.toArray(new Error[mErrorList.size()])); 3899 if (commentBundles != null) mDataflow.setCommentBindings(CommentBindingProjection.merge(commentBundles)); 3900 return mDataflow; 3901 } 3902 3903 public static dataflow readDataflowFromCsvMetadata(String csvMetadata) { 3904 if (!MetadataReader.isMetadata(csvMetadata)) { 3905 throw new IllegalArgumentException("Illegal csv metadata."); 3906 } 3907 return new SQLDepMetadataAnalyzer().analyzeMetadata(null, csvMetadata); 3908 } 3909 3910 private static Relationship toRelationship(relationship relation){ 3911 Relationship relationModel; 3912 if (relation.getType().equals("join")) { 3913 JoinRelationship joinRelationModel = new JoinRelationship(); 3914 joinRelationModel.setCondition(relation.getCondition()); 3915 joinRelationModel.setJoinType(relation.getJoinType()); 3916 joinRelationModel.setClause(relation.getClause()); 3917 relationModel = joinRelationModel; 3918 } else { 3919 relationModel = new Relationship(); 3920 } 3921 relationModel.setId(relation.getId()); 3922 relationModel.setProcessId(relation.getProcessId()); 3923 relationModel.setProcessType(relation.getProcessType()); 3924 relationModel.setType(relation.getType()); 3925 relationModel.setEffectType(relation.getEffectType()); 3926 relationModel.setPartition(relation.getPartition()); 3927 relationModel.setFunction(relation.getFunction()); 3928 relationModel.setProcedureId(relation.getProcedureId()); 3929 relationModel.setSqlHash(relation.getSqlHash()); 3930 relationModel.setCondition(relation.getCondition()); 3931 relationModel.setSqlComment(relation.getSqlComment()); 3932 relationModel.setTimestampMax(relation.getTimestampMax()); 3933 relationModel.setTimestampMin(relation.getTimestampMin()); 3934 3935 if (relation.getTarget() != null && relation.getSources() != null && !relation.getSources().isEmpty()) { 3936 RelationshipElement targetModel = new RelationshipElement(); 3937 targetColumn target = relation.getTarget(); 3938 targetModel.setColumn(target.getColumn()); 3939 targetModel.setParentName(target.getParent_name()); 3940 targetModel.setTargetName(target.getTarget_name()); 3941 targetModel.setId(target.getId()); 3942 targetModel.setTargetId(target.getTarget_id()); 3943 targetModel.setParentId(target.getParent_id()); 3944 targetModel.setCoordinates(Coordinate.parse(target.getCoordinate())); 3945 targetModel.setFunction(target.getFunction()); 3946 targetModel.setType(target.getType()); 3947 relationModel.setTarget(targetModel); 3948 3949 List<RelationshipElement> sourceModels = new ArrayList<>(); 3950 for (sourceColumn source : relation.getSources()) { 3951 RelationshipElement sourceModel = new RelationshipElement(); 3952 sourceModel.setColumn(source.getColumn()); 3953 sourceModel.setParentName(source.getParent_name()); 3954 sourceModel.setSourceName(source.getSource_name()); 3955 sourceModel.setColumnType(source.getColumn_type()); 3956 sourceModel.setId(source.getId()); 3957 sourceModel.setParentId(source.getParent_id()); 3958 sourceModel.setSourceId(source.getSource_id()); 3959 sourceModel.setCoordinates(Coordinate.parse(source.getCoordinate())); 3960 sourceModel.setClauseType(source.getClauseType()); 3961 sourceModel.setType(source.getType()); 3962 sourceModels.add(sourceModel); 3963 if (source.getTransforms() != null && !source.getTransforms().isEmpty()) { 3964 List<Transform> transforms = new ArrayList<gudusoft.gsqlparser.dlineage.dataflow.model.json.Transform>(); 3965 for (transform transform : source.getTransforms()) { 3966 Transform item = new Transform(); 3967 item.setCode(transform.getCode()); 3968 item.setType(transform.getType()); 3969 item.setCoordinate(transform.getCoordinate(true)); 3970 transforms.add(item); 3971 } 3972 sourceModel.setTransforms(transforms.toArray(new Transform[0])); 3973 } 3974 } 3975 relationModel.setSources(sourceModels.toArray(new RelationshipElement[0])); 3976 } else if (relation.getCaller() != null && relation.getCallees() != null 3977 && !relation.getCallees().isEmpty()) { 3978 RelationshipElement targetModel = new RelationshipElement(); 3979 targetColumn target = relation.getCaller(); 3980 targetModel.setName(target.getName()); 3981 targetModel.setId(target.getId()); 3982 targetModel.setCoordinates(Coordinate.parse(target.getCoordinate())); 3983 targetModel.setType(target.getType()); 3984 relationModel.setCaller(targetModel); 3985 List<RelationshipElement> sourceModels = new ArrayList<gudusoft.gsqlparser.dlineage.dataflow.model.json.RelationshipElement>(); 3986 for (sourceColumn source : relation.getCallees()) { 3987 RelationshipElement sourceModel = new RelationshipElement(); 3988 sourceModel.setName(source.getName()); 3989 sourceModel.setId(source.getId()); 3990 sourceModel.setCoordinates(Coordinate.parse(source.getCoordinate())); 3991 sourceModel.setType(source.getType()); 3992 sourceModels.add(sourceModel); 3993 } 3994 relationModel.setCallees(sourceModels.toArray(new RelationshipElement[0])); 3995 } 3996 return relationModel; 3997 } 3998 3999// public static void main(String[] args) throws Exception { 4000// // 基于文件的迭代合并(节约内存,适合大数据量) 4001// dataflow dataflow = iterativeMergeDataflowsFromFiles(Arrays.asList(new File("C:\\Users\\KK\\xwechat_files\\wxid_z9ci6s8b7g0d21_9365\\msg\\file\\2026-06\\dataflow_item").listFiles()), EDbVendor.dbvoracle); 4002// System.out.println(XML2Model.saveXML(dataflow)); 4003// } 4004// 4005// public static void main(String[] args) throws Exception { 4006// // 基于内存的合并(一次性加载所有文件到内存,适合小数据量) 4007// File[] files = new File("C:\\Users\\KK\\xwechat_files\\wxid_z9ci6s8b7g0d21_9365\\msg\\file\\2026-06\\dataflow_item").listFiles(); 4008// if (files == null || files.length == 0) { 4009// System.out.println("No dataflow files found."); 4010// return; 4011// } 4012// 4013// List<dataflow> dataflows = new ArrayList<>(); 4014// for (File file : files) { 4015// dataflow df = XML2Model.loadXML(dataflow.class, file); 4016// if (df != null) { 4017// dataflows.add(df); 4018// } 4019// } 4020// 4021// dataflow mergedDataflow = mergeDataflows(dataflows, EDbVendor.dbvoracle); 4022// System.out.println(XML2Model.saveXML(mergedDataflow)); 4023// } 4024// 4025// public static void main(String[] args) throws Exception { 4026// String sqlDirPath = "C:\\Users\\KK\\Desktop\\同义词验证"; 4027// File sqlDir = new File(sqlDirPath); 4028// 4029// if (!sqlDir.exists() || !sqlDir.isDirectory()) { 4030// System.err.println("目录不存在或不是目录: " + sqlDirPath); 4031// return; 4032// } 4033// 4034// File[] sqlFiles = sqlDir.listFiles((dir, name) -> name.toLowerCase().endsWith(".sql")); 4035// 4036// if (sqlFiles == null || sqlFiles.length == 0) { 4037// System.err.println("目录下没有找到 SQL 文件: " + sqlDirPath); 4038// return; 4039// } 4040// 4041// System.out.println("找到 " + sqlFiles.length + " 个 SQL 文件"); 4042// for (File file : sqlFiles) { 4043// System.out.println(" - " + file.getName()); 4044// } 4045// 4046// Option option = new Option(); 4047// option.setVendor(EDbVendor.dbvoracle); 4048// option.setOutput(true); 4049// option.setSimpleOutput(true); 4050// option.setSimpleShowSynonym(true); 4051// option.setParallel(4); 4052// 4053// ParallelDataFlowAnalyzer analyzer = new ParallelDataFlowAnalyzer( 4054// sqlFiles, 4055// option 4056// ); 4057// 4058// System.out.println("\n开始分析..."); 4059// long startTime = System.currentTimeMillis(); 4060// analyzer.generateDataFlow(false, true); 4061// long endTime = System.currentTimeMillis(); 4062// 4063// dataflow dataflow = analyzer.getDataFlow(); 4064// 4065// analyzer.dispose(); 4066// 4067// if (dataflow == null) { 4068// System.err.println("分析失败,未生成数据流"); 4069// return; 4070// } 4071// 4072// Option option1 = new Option(); 4073// option1.setVendor(EDbVendor.dbvoracle); 4074// option1.setSimpleOutput(true); 4075// dataflow simpleDataflow = new DataFlowAnalyzer("", option1).getSimpleDataflow(dataflow, true); 4076// 4077// 4078// System.out.println("分析完成,耗时: " + (endTime - startTime) + " ms"); 4079// System.out.println("表数量: " + (simpleDataflow.getTables() != null ? simpleDataflow.getTables().size() : 0)); 4080// System.out.println("视图数量: " + (simpleDataflow.getViews() != null ? simpleDataflow.getViews().size() : 0)); 4081// System.out.println("关系数量: " + (simpleDataflow.getRelationships() != null ? simpleDataflow.getRelationships().size() : 0)); 4082// System.out.println("错误数量: " + (simpleDataflow.getErrors() != null ? simpleDataflow.getErrors().size() : 0)); 4083// 4084// System.out.println(XML2Model.saveXML(simpleDataflow)); 4085// 4086// } 4087}