Source code

001package gudusoft.gsqlparser.dlineage;
002
003import gudusoft.gsqlparser.EDbVendor;
004import gudusoft.gsqlparser.TBaseType;
005import gudusoft.gsqlparser.TGSqlParser;
006import gudusoft.gsqlparser.dlineage.dataflow.listener.DataFlowHandleListener;
007import gudusoft.gsqlparser.dlineage.dataflow.metadata.MetadataReader;
008import gudusoft.gsqlparser.dlineage.dataflow.metadata.grabit.GrabitMetadataAnalyzer;
009import gudusoft.gsqlparser.dlineage.dataflow.metadata.sqlflow.SqlflowMetadataAnalyzer;
010import gudusoft.gsqlparser.dlineage.dataflow.metadata.sqlflow.sharded.SqlflowShardedMetadataAnalyzer;
011import gudusoft.gsqlparser.dlineage.dataflow.model.*;
012import gudusoft.gsqlparser.dlineage.dataflow.model.json.Coordinate;
013import gudusoft.gsqlparser.dlineage.dataflow.model.xml.*;
014import gudusoft.gsqlparser.dlineage.dataflow.sqlenv.SQLEnvParser;
015import gudusoft.gsqlparser.dlineage.util.*;
016import gudusoft.gsqlparser.sqlenv.ESQLDataObjectType;
017import gudusoft.gsqlparser.sqlenv.TSQLEnv;
018
019import gudusoft.gsqlparser.util.IndexedLinkedHashMap;
020import gudusoft.gsqlparser.util.Logger;
021import gudusoft.gsqlparser.util.LoggerFactory;
022import gudusoft.gsqlparser.util.SQLUtil;
023import gudusoft.gsqlparser.util.json.JSON;
024
025import java.io.File;
026import java.io.FileInputStream;
027import java.io.IOException;
028import java.io.BufferedReader;
029import java.io.InputStreamReader;
030import java.util.zip.GZIPInputStream;
031import java.util.*;
032import java.util.concurrent.*;
033import java.util.concurrent.atomic.AtomicInteger;
034
035public class ParallelDataFlowAnalyzer implements IDataFlowAnalyzer {
036    private static final Logger logger = LoggerFactory.getLogger(ParallelDataFlowAnalyzer.class);
037    private SqlInfo[] sqlInfos;
038    private Option option = new Option();
039    private TSQLEnv sqlenv = null;
040    private dataflow dataflow;
041    private String dataflowString;
042    private List<ErrorInfo> errorInfos = new CopyOnWriteArrayList<ErrorInfo>();
043    private IndexedLinkedHashMap<String, List<SqlInfo>> sqlInfoMap = new IndexedLinkedHashMap<String, List<SqlInfo>>();
044    //autogenerate 并发任务在 run() 中并发 putAll，必须使用并发容器，否则 HashMap 并发写入会丢失/损坏条目
045    private final Map<String, String> hashSQLMap = new ConcurrentHashMap<>();
046    private dataflow metadataFlow;
047
048    public ParallelDataFlowAnalyzer(SqlInfo[] sqlInfos, EDbVendor dbVendor, boolean simpleOutput) {
049        this.sqlInfos = sqlInfos;
050        option.setVendor(dbVendor);
051        option.setSimpleOutput(simpleOutput);
052        ModelBindingManager.setGlobalOption(option);
053    }
054
055    public ParallelDataFlowAnalyzer(String[] sqlContents, EDbVendor dbVendor, boolean simpleOutput, String defaultServer, String defaultDatabase, String defaltSchema) {
056        SqlInfo[] sqlInfos = new SqlInfo[sqlContents.length];
057        for (int i = 0; i < sqlContents.length; i++) {
058            SqlInfo info = new SqlInfo();
059            info.setSql(sqlContents[i]);
060            info.setOriginIndex(0);
061            sqlInfos[i] = info;
062        }
063        option.setVendor(dbVendor);
064        option.setSimpleOutput(simpleOutput);
065                option.setDefaultServer(defaultServer);
066                option.setDefaultDatabase(defaultDatabase);
067                option.setDefaultSchema(defaltSchema);
068        this.sqlInfos = convertSQL(dbVendor, JSON.toJSONString(sqlInfos)).toArray(new SqlInfo[0]);
069    }
070    
071    public ParallelDataFlowAnalyzer(String[] sqlContents, EDbVendor dbVendor, boolean simpleOutput) {
072        SqlInfo[] sqlInfos = new SqlInfo[sqlContents.length];
073        for (int i = 0; i < sqlContents.length; i++) {
074            SqlInfo info = new SqlInfo();
075            info.setSql(sqlContents[i]);
076            info.setOriginIndex(0);
077            sqlInfos[i] = info;
078        }
079        option.setVendor(dbVendor);
080        option.setSimpleOutput(simpleOutput);
081        this.sqlInfos = convertSQL(dbVendor, JSON.toJSONString(sqlInfos)).toArray(new SqlInfo[0]);
082    }
083
084    public ParallelDataFlowAnalyzer(SqlInfo[] sqlInfos, Option option) {
085        this.sqlInfos = sqlInfos;
086        this.option = option;
087        ModelBindingManager.setGlobalOption(option);
088    }
089
090    public ParallelDataFlowAnalyzer(File[] sqlFiles, Option option) {
091        SqlInfo[] sqlInfos = new SqlInfo[sqlFiles.length];
092        for (int i = 0; i < sqlFiles.length; i++) {
093            SqlInfo info = new SqlInfo();
094            info.setSql(SQLUtil.getFileContent(sqlFiles[i]));
095            info.setFileName(sqlFiles[i].getName());
096            info.setFilePath(sqlFiles[i].getAbsolutePath());
097            info.setOriginIndex(0);
098            sqlInfos[i] = info;
099        }
100        this.sqlInfos = sqlInfos;
101        this.option = option;
102        ModelBindingManager.setGlobalOption(option);
103    }
104    
105    public ParallelDataFlowAnalyzer(File[] sqlFiles, Option option, int splitSizeMB, File splitDir) {
106        List<SqlInfo> sqlInfoList = new ArrayList<SqlInfo>();
107        for (int i = 0; i < sqlFiles.length; i++) {
108            try {
109                // Ensure split directory exists
110                if (!splitDir.exists() && !splitDir.mkdirs()) {
111                    throw new IOException("Failed to create split directory: " + splitDir.getAbsolutePath());
112                }
113                
114                // Split the file if it's large
115                List<File> splitFiles = gudusoft.gsqlparser.util.FileSplitter.splitFile(sqlFiles[i], splitDir, splitSizeMB, option.getVendor());
116                
117                if (splitFiles.isEmpty()) {
118                    // If no split files were created, process the original file
119                    SqlInfo info = new SqlInfo();
120                    info.setSql(SQLUtil.getFileContent(sqlFiles[i]));
121                    info.setFileName(sqlFiles[i].getName());
122                    info.setFilePath(sqlFiles[i].getAbsolutePath());
123                    info.setOriginIndex(0);
124                    sqlInfoList.add(info);
125                } else {
126                    // Process each split file
127                    for (File splitFile : splitFiles) {
128                        SqlInfo info = new SqlInfo();
129                        info.setSql(SQLUtil.getFileContent(splitFile));
130                        info.setFileName(splitFile.getName());
131                        info.setFilePath(splitFile.getAbsolutePath());
132                        info.setOriginIndex(0);
133                        sqlInfoList.add(info);
134                    }
135                }
136            } catch (Exception e) {
137                sqlInfoList.clear();
138                logger.error("Error splitting file: " + sqlFiles[i].getAbsolutePath(), e);
139                // Fallback to original file if splitting fails
140                SqlInfo info = new SqlInfo();
141                info.setSql(SQLUtil.getFileContent(sqlFiles[i]));
142                info.setFileName(sqlFiles[i].getName());
143                info.setFilePath(sqlFiles[i].getAbsolutePath());
144                info.setOriginIndex(0);
145                sqlInfoList.add(info);
146            }
147        }
148        this.sqlInfos = sqlInfoList.toArray(new SqlInfo[0]);
149        this.option = option;
150        ModelBindingManager.setGlobalOption(option);
151    }
152    
153    public ParallelDataFlowAnalyzer(File[] sqlFiles, Option option, int splitSizeMB, String splitDirPath) {
154        this(sqlFiles, option, splitSizeMB, new File(splitDirPath));
155    }
156    
157    protected List<SqlInfo> convertSQL(EDbVendor vendor, String json) {
158                List<SqlInfo> sqlInfos = new ArrayList<SqlInfo>();
159                List sqlContents = (List) JSON.parseObject(json);
160                for (int j = 0; j < sqlContents.size(); j++) {
161                        Map sqlContent = (Map) sqlContents.get(j);
162                        String sql = (String) sqlContent.get("sql");
163                        String fileName = (String) sqlContent.get("fileName");
164                        String filePath = (String) sqlContent.get("filePath");
165                        if (sql != null && sql.trim().startsWith("{")) {
166                                if (sql.indexOf("createdBy") != -1) {
167                        // Exclude sqlflow-sharded manifests from the brand branch so a
168                        // SQLdep/grabit-branded sharded export routes to the sharded
169                        // reader below instead of the legacy queries[] path.
170                        if ((sql.toLowerCase().indexOf("sqldep") != -1 || sql.toLowerCase().indexOf("grabit") != -1)
171                                        && !gudusoft.gsqlparser.dlineage.dataflow.metadata.MetadataReader.isSqlflowSharded(sql)) {
172                                Map queryObject = (Map) JSON.parseObject(sql);
173                                        List querys = (List) queryObject.get("queries");
174                                        if (querys != null) {
175                                                for (int i = 0; i < querys.size(); i++) {
176                                                        Map object = (Map) querys.get(i);
177                                                        SqlInfo info = new SqlInfo();
178                                                        info.setSql(JSON.toJSONString(object));
179                                                        info.setFileName(fileName);
180                                                        info.setFilePath(filePath);
181                                                        info.setOriginIndex(i);
182                                                        sqlInfos.add(info);
183                                                }
184                                                queryObject.remove("queries");
185                                                SqlInfo info = new SqlInfo();
186                                                info.setSql(JSON.toJSONString(queryObject));
187                                                info.setFileName(fileName);
188                                                info.setFilePath(filePath);
189                                                info.setOriginIndex(querys.size());
190                                                sqlInfos.add(info);
191                                        } else {
192                                                SqlInfo info = new SqlInfo();
193                                                info.setSql(JSON.toJSONString(queryObject));
194                                                info.setFileName(fileName);
195                                                info.setFilePath(filePath);
196                                                info.setOriginIndex(0);
197                                                sqlInfos.add(info);
198                                        }
199                        }
200                        else if (sql.toLowerCase().indexOf("sqlflow") != -1) {
201                                                Map sqlflow = (Map) JSON.parseObject(sql);
202                                                if ("sqlflow-sharded".equals(sqlflow.get("format"))
203                                                                && !gudusoft.gsqlparser.dlineage.dataflow.metadata.MetadataReader.isSupportedSqlflowSharded(sql)) {
204                                                        // Unknown future formatVersion: do not consume it as a
205                                                        // version we understand.
206                                                } else if ("sqlflow-sharded".equals(sqlflow.get("format"))) {
207                                                        String baseDir = null;
208                                                        if (filePath != null) {
209                                                                baseDir = new File(filePath).getParent();
210                                                        }
211                                                        String sourceCompression = (String) sqlflow.get("sourceCompression");
212                                                        List<Map> servers = (List<Map>) sqlflow.get("servers");
213                                                        if (servers != null) {
214                                                                for (Map serverObject : servers) {
215                                                                        String name = (String) serverObject.get("name");
216                                                                        String dbVendor = (String) serverObject.get("dbVendor");
217                                                                        List<Map> databases = (List<Map>) serverObject.get("databases");
218                                                                        // Source shards are under databases[] (catalog topology) or
219                                                                        // schemas[] (schema topology, the natural Oracle shape). Read
220                                                                        // whichever is present; databases-only dropped every source
221                                                                        // record of a schema-topology manifest -> empty lineage.
222                                                                        List<Map> schemaShards = (List<Map>) serverObject.get("schemas");
223                                                                        if (schemaShards != null && !schemaShards.isEmpty()) {
224                                                                                databases = schemaShards;
225                                                                        }
226                                                                        if (databases != null) {
227                                                                                for (Map database : databases) {
228                                                                                        Map source = (Map) database.get("source");
229                                                                                        if (source != null) {
230                                                                                                String sourcePath = (String) source.get("path");
231                                                                                                if (sourcePath != null && baseDir != null) {
232                                                    File sourceFile = new File(baseDir, sourcePath);
233                                                    String fullPath = sourceFile.getAbsolutePath();
234                                                    if ("block".equals(sourceCompression)) {
235                                                                                                                readGzipBlockSource(fullPath, fileName, filePath, name, dbVendor, sqlInfos);
236                                                                                                        } else {
237                                                                                                                String sourceContent = SQLUtil.getFileContent(fullPath);
238                                                                                                                if (sourceContent != null) {
239                                                                                                                        String[] lines = sourceContent.split("\\r?\\n");
240                                                                                                                        for (int i = 0; i < lines.length; i++) {
241                                                                                                                                String line = lines[i].trim();
242                                                                                                                                if (line.isEmpty()) {
243                                                                                                                                        continue;
244                                                                                                                                }
245                                                                                                                                try {
246                                                                                                                                        Map sourceObject = (Map) JSON.parseObject(line);
247                                                                                                                                        String sourceCode = (String) sourceObject.get("sourceCode");
248                                                                                                                                        if (sourceCode != null && !sourceCode.isEmpty() && !DataFlowAnalyzer.isSourceUnavailable(sourceObject)) {
249                                                                                                                                                SqlInfo info = new SqlInfo();
250                                                                                                                                                info.setSql(sourceCode);
251                                                                                                                                                info.setFileName(sourceFile.getName());
252                                                                                                                                                info.setFilePath(sourceFile.getAbsolutePath());
253                                                                                                                                                info.setOriginIndex(i);
254                                                                                                                                                info.setDbVendor(dbVendor);
255                                                                                                                                                info.setServer(name);
256                                                                                                                                                sqlInfos.add(info);
257                                                                                                                                        }
258                                                                                                                                } catch (Exception e) {
259                                                                                                                                        logger.warn("Parse source jsonl line failed.", e);
260                                                                                                                                }
261                                                                                                                        }
262                                                                                                                }
263                                                                                                        }
264                                                                                                }
265                                                                                        }
266                                                                                }
267                                                                        }
268                                                                        SqlInfo serverInfo = new SqlInfo();
269                                                                        serverInfo.setSql(JSON.toJSONString(serverObject));
270                                                                        serverInfo.setFileName(fileName);
271                                                                        serverInfo.setFilePath(filePath);
272                                                                        serverInfo.setDbVendor(dbVendor);
273                                                                        serverInfo.setServer(name);
274                                                                        sqlInfos.add(serverInfo);
275                                                                }
276                                                        }
277                                                } else {
278                                                        List<Map> servers = (List<Map>) sqlflow.get("servers");
279                                                        if (servers != null) {
280                                                                for (Map queryObject : servers) {
281                                                                        String name = (String) queryObject.get("name");
282                                                                        String dbVendor = (String) queryObject.get("dbVendor");
283                                                                        List querys = (List) queryObject.get("queries");
284                                                                        if (querys != null) {
285                                                                                for (int i = 0; i < querys.size(); i++) {
286                                                                                        Map object = (Map) querys.get(i);
287                                                                                        SqlInfo info = new SqlInfo();
288                                                                                        info.setSql(JSON.toJSONString(object));
289                                                                                        info.setFileName(fileName);
290                                                                                        info.setFilePath(filePath);
291                                                                                        info.setOriginIndex(i);
292                                                                                        info.setDbVendor(dbVendor);
293                                                                                        info.setServer(name);
294                                                                                        sqlInfos.add(info);
295                                                                                }
296                                                                                queryObject.remove("queries");
297                                                                                SqlInfo info = new SqlInfo();
298                                                                                info.setSql(JSON.toJSONString(queryObject));
299                                                                                info.setFileName(fileName);
300                                                                                info.setFilePath(filePath);
301                                                                                info.setOriginIndex(querys.size());
302                                                                                info.setDbVendor(dbVendor);
303                                                                                info.setServer(filePath);
304                                                                                sqlInfos.add(info);
305                                                                        } else {
306                                                                                SqlInfo info = new SqlInfo();
307                                                                                info.setSql(JSON.toJSONString(queryObject));
308                                                                                info.setFileName(fileName);
309                                                                                info.setFilePath(filePath);
310                                                                                info.setOriginIndex(0);
311                                                                                sqlInfos.add(info);
312                                                                        }
313                                                                }
314                                                        }
315                                                }
316                                        }
317                    }
318                        } else if (sql != null) {
319                                SqlInfo info = new SqlInfo();
320                                info.setSql(sql);
321                                info.setFileName(fileName);
322                                info.setFilePath(filePath);
323                                info.setOriginIndex(0);
324                                sqlInfos.add(info);
325                        }
326                }
327                return sqlInfos;
328        }
329
330        private void readGzipBlockSource(String fullPath, String fileName, String filePath, String serverName, String dbVendor, List<SqlInfo> sqlInfos) {
331                try (FileInputStream fis = new FileInputStream(fullPath);
332                         GZIPInputStream gzis = new GZIPInputStream(fis);
333                         BufferedReader reader = new BufferedReader(new InputStreamReader(gzis, "UTF-8"))) {
334                        String line;
335                        int j = 0;
336                        while ((line = reader.readLine()) != null) {
337                                line = line.trim();
338                                if (line.isEmpty()) {
339                                        continue;
340                                }
341                                try {
342                                        Map sourceObject = (Map) JSON.parseObject(line);
343                                        String sourceCode = (String) sourceObject.get("sourceCode");
344                                        if (sourceCode != null && !sourceCode.isEmpty() && !DataFlowAnalyzer.isSourceUnavailable(sourceObject)) {
345                                                SqlInfo info = new SqlInfo();
346                                                info.setSql(sourceCode);
347                                                info.setFileName(fileName);
348                                                info.setFilePath(filePath);
349                                                info.setOriginIndex(j);
350                                                info.setDbVendor(dbVendor);
351                                                info.setServer(serverName);
352                                                sqlInfos.add(info);
353                                        }
354                                } catch (Exception e) {
355                                        logger.warn("Parse gzip source jsonl line failed.", e);
356                                }
357                                j++;
358                        }
359                } catch (Exception e) {
360                        logger.warn("Read gzip source file failed: " + fullPath, e);
361                }
362        }
363
364    @Override
365    public boolean isIgnoreRecordSet() {
366        return option.isIgnoreRecordSet();
367    }
368
369    @Override
370    public void setIgnoreRecordSet(boolean ignoreRecordSet) {
371        option.setIgnoreRecordSet(ignoreRecordSet);
372    }
373
374    @Override
375    public boolean isSimpleShowTopSelectResultSet() {
376        return option.isSimpleShowTopSelectResultSet();
377    }
378
379    @Override
380    public void setSimpleShowTopSelectResultSet(boolean simpleShowTopSelectResultSet) {
381        option.setSimpleShowTopSelectResultSet(simpleShowTopSelectResultSet);
382    }
383
384    @Override
385    public boolean isSimpleShowFunction() {
386        return option.isSimpleShowFunction();
387    }
388
389    @Override
390    public void setSimpleShowFunction(boolean simpleShowFunction) {
391        option.setSimpleShowFunction(simpleShowFunction);
392    }
393
394    @Override
395    public boolean isShowJoin() {
396        return option.isShowJoin();
397    }
398
399    @Override
400    public void setShowJoin(boolean showJoin) {
401        option.setShowJoin(showJoin);
402    }
403
404    @Override
405    public boolean isShowImplicitSchema() {
406        return option.isShowImplicitSchema();
407    }
408
409    @Override
410    public void setShowImplicitSchema(boolean showImplicitSchema) {
411        option.setShowImplicitSchema(showImplicitSchema);
412    }
413
414    @Override
415    public boolean isShowConstantTable() {
416        return option.isShowConstantTable();
417    }
418
419    @Override
420    public void setShowConstantTable(boolean showConstantTable) {
421        option.setShowConstantTable(showConstantTable);
422    }
423
424    @Override
425    public boolean isShowCountTableColumn() {
426        return option.isShowCountTableColumn();
427    }
428
429    @Override
430    public void setShowCountTableColumn(boolean showCountTableColumn) {
431        option.setShowCountTableColumn(showCountTableColumn);
432    }
433
434    @Override
435    public boolean isTransform() {
436        return option.isTransform();
437    }
438
439    @Override
440    public void setTransform(boolean transform) {
441        option.setTransform(transform);
442        if (option.isTransformCoordinate()) {
443            option.setTransform(true);
444        }
445    }
446
447    @Override
448    public boolean isTransformCoordinate() {
449        return option.isTransformCoordinate();
450    }
451
452    @Override
453    public void setTransformCoordinate(boolean transformCoordinate) {
454        option.setTransformCoordinate(transformCoordinate);
455        if (transformCoordinate) {
456            option.setTransform(true);
457        }
458    }
459
460    @Override
461    public boolean isLinkOrphanColumnToFirstTable() {
462        return option.isLinkOrphanColumnToFirstTable();
463    }
464
465    @Override
466    public void setLinkOrphanColumnToFirstTable(boolean linkOrphanColumnToFirstTable) {
467        option.setLinkOrphanColumnToFirstTable(linkOrphanColumnToFirstTable);
468    }
469
470    @Override
471    public boolean isIgnoreCoordinate() {
472        return option.isIgnoreCoordinate();
473    }
474
475    @Override
476    public void setIgnoreCoordinate(boolean ignoreCoordinate) {
477        option.setIgnoreCoordinate(ignoreCoordinate);
478    }
479
480    @Override
481    public void setHandleListener(DataFlowHandleListener listener) {
482        option.setHandleListener(listener);
483    }
484
485    @Override
486    public void setSqlEnv(TSQLEnv sqlenv) {
487        this.sqlenv = sqlenv;
488    }
489
490    @Override
491    public void setOption(Option option) {
492        this.option = option;
493    }
494
495    @Override
496    public Option getOption() {
497        return option;
498    }
499
500    @Override
501    public List<ErrorInfo> getErrorMessages() {
502        return errorInfos;
503    }
504
505    @Override
506    public synchronized String generateSqlInfos() {
507        return JSON.toJSONString(sqlInfoMap);
508    }
509
510    @Override
511    public synchronized String generateDataFlow() {
512        return generateDataFlow(false);
513    }
514
515    @Override
516    public Map<String, List<SqlInfo>> getSqlInfos() {
517        return sqlInfoMap;
518    }
519
520    @Override
521    /**
522     * @deprecated please use SqlInfoHelper.getSelectedDbObjectInfo
523     */
524    public DbObjectPosition getSelectedDbObjectInfo(Coordinate start, Coordinate end) {
525        if (start == null || end == null) {
526            throw new IllegalArgumentException("Coordinate can't be null.");
527        }
528
529        String hashCode = start.getHashCode();
530
531        if (hashCode == null) {
532            throw new IllegalArgumentException("Coordinate hashcode can't be null.");
533        }
534
535        int dbObjectStartLine = (int) start.getX() - 1;
536        int dbObjectStarColumn = (int) start.getY() - 1;
537        int dbObjectEndLine = (int) end.getX() - 1;
538        int dbObjectEndColumn = (int) end.getY() - 1;
539        List<SqlInfo> sqlInfoList;
540        if (hashCode.matches("\\d+")) {
541            sqlInfoList = sqlInfoMap.getValueAtIndex(Integer.valueOf(hashCode));
542        } else {
543            sqlInfoList = sqlInfoMap.get(hashCode);
544        }
545        for (int j = 0; j < sqlInfoList.size(); j++) {
546            SqlInfo sqlInfo = sqlInfoList.get(j);
547            int startLine = sqlInfo.getLineStart();
548            int endLine = sqlInfo.getLineEnd();
549            if (dbObjectStartLine >= startLine && dbObjectStartLine <= endLine) {
550                DbObjectPosition position = new DbObjectPosition();
551                position.setFile(sqlInfo.getFileName());
552                position.setFilePath(sqlInfo.getFilePath());
553                position.setSql(sqlInfo.getSql());
554                position.setIndex(sqlInfo.getOriginIndex());
555                List<Pair<Integer, Integer>> positions = position.getPositions();
556                positions.add(new Pair<Integer, Integer>(
557                        dbObjectStartLine - startLine + sqlInfo.getOriginLineStart() + 1, dbObjectStarColumn + 1));
558                positions.add(new Pair<Integer, Integer>(dbObjectEndLine - startLine + sqlInfo.getOriginLineStart() + 1,
559                        dbObjectEndColumn + 1));
560                return position;
561            }
562        }
563        return null;
564    }
565
566    @Override
567    public synchronized String generateDataFlow(final boolean withExtraInfo) {
568        return generateDataFlow(withExtraInfo, true);
569    }
570    
571    public synchronized String generateDataFlow(final boolean withExtraInfo, boolean useSaveMemoryMode) {
572        sqlInfoMap.clear();
573        errorInfos.clear();
574        Map<String, Pair3<StringBuilder, AtomicInteger, String>> databaseMap = new LinkedHashMap<String, Pair3<StringBuilder, AtomicInteger, String>>();
575        for (int i = 0; i < sqlInfos.length; i++) {
576            SqlInfo sqlInfo = sqlInfos[i];
577                        if (sqlInfo != null && sqlInfo.getSql() == null && sqlInfo.getFilePath() != null) {
578                                sqlInfo.setSql(SQLUtil.getFileContent(sqlInfo.getFilePath()));
579                        }
580            if (sqlInfo != null && sqlInfo.getSql() == null && sqlInfo.getFileName() != null) {
581                sqlInfo.setSql(SQLUtil.getFileContent(sqlInfo.getFileName()));
582            }
583            if (sqlInfo == null || sqlInfo.getSql() == null) {
584                sqlInfoMap.put(String.valueOf(i), new ArrayList<SqlInfo>());
585                continue;
586            }
587            String sql = sqlInfo.getSql();
588            if (sql != null && sql.trim().startsWith("{")) {
589                if (MetadataReader.isGrabit(sql) || MetadataReader.isSqlflow(sql) || MetadataReader.isSqlflowSharded(sql)) {
590                    String hash = SHA256.getMd5(sql);
591                    String fileHash = SHA256.getMd5(hash);
592                    if (!sqlInfoMap.containsKey(fileHash)) {
593                        sqlInfoMap.put(fileHash, new ArrayList<SqlInfo>());
594                        sqlInfoMap.get(fileHash).add(sqlInfo);
595                    }
596                    if (this.metadataFlow == null) {
597                        this.metadataFlow = analyzeMetadata(sql, sqlInfo.getFilePath());
598                    }
599                } else {
600                    Map queryObject = (Map) JSON.parseObject(sql);
601                    appendSqlInfo(databaseMap, i, sqlInfo, queryObject);
602                }
603            } else {
604                String content = sql;
605                String hash = SHA256.getMd5(content);
606                String fileHash = SHA256.getMd5(hash);
607                if (!sqlInfoMap.containsKey(fileHash)) {
608                    sqlInfoMap.put(fileHash, new ArrayList<SqlInfo>());
609
610                    String database = TSQLEnv.DEFAULT_DB_NAME;
611                    String schema = TSQLEnv.DEFAULT_SCHEMA_NAME;
612                    if(sqlenv!=null) {
613                        // Prefer the per-analysis Option default database over the attached
614                        // env's default catalog (mirrors DataFlowAnalyzer), so a shared
615                        // multi-catalog TSQLEnv can be reused across analyses with different
616                        // default databases.
617                        if (!SQLUtil.isEmpty(option.getDefaultDatabase())
618                                && !TSQLEnv.DEFAULT_DB_NAME.equals(option.getDefaultDatabase())) {
619                            database = option.getDefaultDatabase();
620                        } else {
621                            database = sqlenv.getDefaultCatalogName();
622                        }
623                        if(database == null) {
624                            database = TSQLEnv.DEFAULT_DB_NAME;
625                        }
626                        schema = sqlenv.getDefaultSchemaName();
627                        if(schema == null) {
628                            schema = TSQLEnv.DEFAULT_SCHEMA_NAME;
629                        }
630                    }
631                    boolean supportCatalog = TSQLEnv.supportCatalog(option.getVendor());
632                    boolean supportSchema = TSQLEnv.supportSchema(option.getVendor());
633                    StringBuilder builder = new StringBuilder();
634                    if (supportCatalog) {
635                        builder.append(database);
636                    }
637                    if (supportSchema) {
638                        if (builder.length() > 0) {
639                            builder.append(".");
640                        }
641                        builder.append(schema);
642                    }
643                    String group = builder.toString();
644                    SqlInfo sqlInfoItem = new SqlInfo();
645                    sqlInfoItem.setFileName(sqlInfo.getFileName());
646                    sqlInfoItem.setFilePath(sqlInfo.getFilePath());
647                    sqlInfoItem.setSql(sqlInfo.getSql());
648                    sqlInfoItem.setOriginIndex(0);
649                    sqlInfoItem.setOriginLineStart(0);
650                    sqlInfoItem.setOriginLineEnd(sqlInfo.getSql().split("\n").length - 1);
651                    sqlInfoItem.setIndex(0);
652                    sqlInfoItem.setLineStart(0);
653                    sqlInfoItem.setLineEnd(sqlInfo.getSql().split("\n").length - 1);
654                    sqlInfoItem.setHash(fileHash);
655                    sqlInfoItem.setGroup(group);
656
657                    sqlInfoMap.get(fileHash).add(sqlInfoItem);
658                }
659            }
660        }
661
662        final TSQLEnv[] env = new TSQLEnv[]{sqlenv};
663        if (sqlenv == null) {
664                if (option.getHandleListener() != null) {
665                        option.getHandleListener().startParseSQLEnv();
666                }
667                TSQLEnv[] envs = new SQLEnvParser(option.getDefaultServer(), option.getDefaultDatabase(), option.getDefaultSchema()).parseSQLEnv(option.getVendor(), sqlInfos);
668                        if (envs != null && envs.length > 0) {
669                                env[0] = envs[0];
670                        }
671                        if (option.getHandleListener() != null) {
672                                option.getHandleListener().endParseSQLEnv();
673                        }
674        }
675        // Guard against a zero-size pool: newFixedThreadPool(0) throws. sqlInfos can be
676        // empty when the only input was a manifest that produced no analyzable source
677        // (e.g. an unsupported sharded version that was refused).
678        int poolSize = calculateMemoryAwarePoolSize(option.getParallel(), sqlInfos.length, option.getEstimatedMemoryPerTaskMB());
679        ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(poolSize);
680        
681        final File[] tempFilesArray = new File[sqlInfos.length];
682        final Map<SqlInfo, dataflow> dataflowMap = useSaveMemoryMode ? null : new ConcurrentHashMap<>();
683        
684        try {
685            final CountDownLatch latch = new CountDownLatch(sqlInfos.length);
686
687            logger.info("start parallel analyze " + sqlInfos.length + " sqlinfos");
688
689            for (int i = 0; i < sqlInfos.length; i++) {
690                final SqlInfo[] sqlInfoCopy = new SqlInfo[sqlInfos.length];
691                final SqlInfo item = sqlInfos[i];
692                sqlInfoCopy[i] = item;
693                final Option optionCopy = (Option) option.clone();
694                optionCopy.setStartId(5000000L * i);
695                optionCopy.setOutput(false);
696                optionCopy.setSimpleRetainIntermediate(true);
697                optionCopy.setAutoDetectLargeFile(false);
698                // Worker-local relationship IDs are remapped during merge. Until
699                // the evidence merge performs the same remap, publishing worker
700                // evidence would create invalid correlations.
701                optionCopy.setCollectAuthoritativeLineageEvidence(false);
702                final int index = i;
703                Runnable task = new Runnable() {
704                    @Override
705                    public void run() {
706                        try {
707                            logger.info("start analyze sqlinfo[" + index + "]" + (sqlInfos[index].getFilePath() != null ? ", file name = " + new File(sqlInfos[index].getFilePath()).getName() : ""));
708                            DataFlowAnalyzer analyzer = new DataFlowAnalyzer(sqlInfoCopy, optionCopy);
709                            if (env[0] != null) {
710                                analyzer.setSqlEnv(env[0].copy());
711                            }
712                            analyzer.generateDataFlow(true);
713                            dataflow dataflow = analyzer.getDataFlow();
714                            if (dataflow == null) {
715                                logger.warn("analyze sqlinfo[" + index + "] done, but dataflow is null" + (sqlInfos[index].getFilePath() != null ? ", file name = " + new File(sqlInfos[index].getFilePath()).getName() : ""));
716                                errorInfos.addAll(analyzer.getErrorMessages());
717                                analyzer.dispose();
718                                return;
719                            }
720                            logger.info("analyze sqlinfo[" + index + "] done, relation count: " + dataflow.getRelationships().size()+", error count: "+ analyzer.getErrorMessages().size());
721                            //autogenerate analyzer 错误集合与 dataflow 错误集合不同源，各自按自身 size 安全截断，避免 subList 越界
722                            List<ErrorInfo> taskErrors = analyzer.getErrorMessages();
723                            if (taskErrors.size() > 10000) {
724                                errorInfos.addAll(taskErrors.subList(0, 10000));
725                                logger.warn("Too many errors ("+taskErrors.size()+"), truncating to first 10000 errors" + (sqlInfos[index].getFileName() != null ? ", file name = " + sqlInfos[index].getFileName() : ""));
726                            } else {
727                                errorInfos.addAll(taskErrors);
728                            }
729                            if (dataflow.getErrors() != null && dataflow.getErrors().size() > 10000) {
730                                dataflow.setErrors(new ArrayList<>(dataflow.getErrors().subList(0, 10000)));
731                            }
732                            
733                            if (useSaveMemoryMode) {
734                                File tempFile = File.createTempFile("dataflow_" + index + "_" + System.currentTimeMillis() + "_", ".xml.zip");
735                                //autogenerate saveXML 失败时删除已创建的临时文件，避免泄漏
736                                try {
737                                    XML2Model.saveXML(dataflow, tempFile);
738                                    tempFilesArray[index] = tempFile;
739                                } catch (Exception ex) {
740                                    tempFile.delete();
741                                    throw ex;
742                                }
743                            } else {
744                                dataflowMap.put(item, dataflow);
745                            }
746                            
747                            hashSQLMap.putAll(analyzer.getHashSQLMap());
748                            analyzer.dispose();
749                        }
750                        catch (Exception e) {
751                            logger.error("analyze sqlinfo[" + index + "] failed.", e);
752                        }
753                        finally {
754                            latch.countDown();
755                        }
756                    }
757                };
758                executor.submit(task);
759            }
760            latch.await();
761        } catch (Exception e) {
762            logger.error("execute task failed.", e);
763        }
764        executor.shutdown();
765
766        ModelBindingManager modelManager = new ModelBindingManager();
767        modelManager.setGlobalVendor(option.getVendor());
768        modelManager.setGlobalOption(option);
769        ModelBindingManager.set(modelManager);
770        
771        if (useSaveMemoryMode) {
772            List<File> tempFiles = new ArrayList<>(sqlInfos.length + 1);
773            for (File f : tempFilesArray) {
774                if (f != null) {
775                    tempFiles.add(f);
776                }
777            }
778            if (metadataFlow != null) {
779                File metadataFile = null;
780                //autogenerate metadata 临时 ZIP 在 saveXML 失败时必须删除，避免泄漏
781                try {
782                    metadataFile = File.createTempFile("gsp-metadata-", ".xml.zip");
783                    XML2Model.saveXML(metadataFlow, metadataFile);
784                    tempFiles.add(metadataFile);
785                } catch (Exception e) {
786                    if (metadataFile != null) {
787                        metadataFile.delete();
788                    }
789                    logger.error("Failed to save metadata to temp file", e);
790                }
791            }
792            logger.info("start merge dataflow, dataflow count: " + tempFiles.size()+", useSaveMemoryMode: "+ useSaveMemoryMode+", gsp version: "+ TBaseType.versionid);
793            //autogenerate 合并与清理放入 try-finally，异常路径也必须删除临时 ZIP，避免泄漏
794            try {
795                this.dataflow = iterativeMergeDataFlows(tempFiles, 5000000L * sqlInfos.length);
796                if (this.dataflow != null && this.dataflow.getRelationships() != null) {
797                    logger.info("merge dataflow done, dataflow count: " + tempFiles.size() + ", relation count: " + this.dataflow.getRelationships().size());
798                }
799            } finally {
800                for (File tempFile : tempFiles) {
801                    tempFile.delete();
802                }
803            }
804        } else {
805            if (metadataFlow != null) {
806                dataflowMap.put(new SqlInfo(), metadataFlow);
807            }
808            logger.info("start merge dataflow, dataflow count: " + dataflowMap.size()+", useSaveMemoryMode: "+ useSaveMemoryMode+", gsp version: "+ TBaseType.versionid);
809            this.dataflow = mergeDataFlows(dataflowMap, 5000000L * sqlInfos.length);
810            if (this.dataflow != null && this.dataflow.getRelationships() != null) {
811                logger.info("merge dataflow done, dataflow count: " + dataflowMap.size() + ", relation count: " + this.dataflow.getRelationships().size());
812            }
813            dataflowMap.clear();
814        }
815        
816                if (this.dataflow != null) {
817                        logger.info("merge done, relation count: " + this.dataflow.getRelationships().size());
818                }
819
820        if(option.isSimpleOutput()){
821            List<String> showTypes = new ArrayList<>();
822            if(option.getSimpleShowRelationTypes()!=null) {
823                showTypes.addAll(option.getSimpleShowRelationTypes());
824            }
825            if(showTypes.isEmpty()) {
826                showTypes.add("fdd");
827            }
828            if(option.isShowCallRelation()) {
829                showTypes.add("call");
830            }
831            if(option.isShowERDiagram()) {
832                showTypes.add("er");
833            }
834            try {
835                this.dataflow = new DataFlowAnalyzer("", option).getSimpleDataflow(this.dataflow, option.isSimpleOutput(), showTypes);
836            } catch (Exception e) {
837                throw new RuntimeException(e);
838            }
839        }
840
841        if (dataflow != null && option.isOutput()) {
842            if (option.isTextFormat()) {
843                dataflowString = DataFlowAnalyzer.getTextOutput(dataflow);
844            } else {
845                try {
846                    dataflowString = XML2Model.saveXML(dataflow);
847                }catch (Exception e){
848                    logger.error("save dataflow as xml failed.", e);
849                    dataflowString = null;
850                }
851            }
852        }
853        ModelBindingManager.remove();
854        return dataflowString;
855    }
856    
857    private dataflow iterativeMergeDataFlows(List<File> tempFiles, long startId) {
858        return DataflowUtility.iterativeMergeDataflowsFromFilesByStartId(tempFiles, startId);
859    }
860
861    private void appendSqlInfo(Map<String, Pair3<StringBuilder, AtomicInteger, String>> databaseMap, int index,
862                               SqlInfo sqlInfo, Map queryObject) {
863        EDbVendor vendor = option.getVendor();
864        if(!SQLUtil.isEmpty(sqlInfo.getDbVendor())){
865            vendor = EDbVendor.valueOf(sqlInfo.getDbVendor());
866        }
867
868        boolean supportCatalog = TSQLEnv.supportCatalog(vendor);
869        boolean supportSchema = TSQLEnv.supportSchema(vendor);
870
871        String groupName = (String) queryObject.get("groupName");
872        if (DlineageUtil.isProcedureExcluded(groupName)) {
873            return;
874        }
875
876        if (DataFlowAnalyzer.isSourceUnavailable(queryObject)) {
877            return; // DDL body not retrieved; its text is not the definition
878        }
879
880        String content = (String) queryObject.get("sourceCode");
881        if (SQLUtil.isEmpty(content)) {
882            return;
883        }
884
885        StringBuilder builder = new StringBuilder();
886        if (supportCatalog) {
887            String database = (String) queryObject.get("database");
888            if (database.indexOf(".") != -1) {
889                database = SQLUtil.quoteDottedName(vendor, ESQLDataObjectType.dotCatalog, database);
890            }
891            builder.append(database);
892        }
893        if (supportSchema) {
894            String schema = (String) queryObject.get("schema");
895            if (schema.indexOf(".") != -1) {
896                schema = SQLUtil.quoteDottedName(vendor, ESQLDataObjectType.dotSchema, schema);
897            }
898            if (builder.length() > 0) {
899                builder.append(".");
900            }
901            builder.append(schema);
902        }
903        String group = builder.toString();
904        String sqlHash = SHA256.getMd5(content);
905        String hash = SHA256.getMd5(sqlHash);
906        if (!databaseMap.containsKey(sqlHash)) {
907            databaseMap.put(sqlHash,
908                    new Pair3<StringBuilder, AtomicInteger, String>(new StringBuilder(), new AtomicInteger(), group));
909        }
910        String delimiterChar = String.valueOf(TGSqlParser.getDelimiterChar(option.getVendor()));
911        StringBuilder buffer = new StringBuilder(content);
912        if (content.trim().endsWith(delimiterChar) || content.trim().endsWith(";")) {
913            buffer.append("\n");
914        } else if(vendor == EDbVendor.dbvredshift
915                || vendor == EDbVendor.dbvgaussdb
916                || vendor == EDbVendor.dbvedb
917                || vendor == EDbVendor.dbvpostgresql
918                || vendor == EDbVendor.dbvmysql
919                || vendor == EDbVendor.dbvteradata){
920            buffer.append("\n\n-- " + TBaseType.sqlflow_stmt_delimiter_str + "\n\n");
921        } else{
922            SQLUtil.endTrim(buffer);
923            buffer.append(";").append("\n");
924        }
925
926        int lineStart = databaseMap.get(sqlHash).first.toString().split("\n", -1).length - 1;
927        if (databaseMap.get(sqlHash).first.toString().length() == 0) {
928            lineStart = 0;
929        }
930        databaseMap.get(sqlHash).first.append(buffer.toString());
931        SqlInfo sqlInfoItem = new SqlInfo();
932        sqlInfoItem.setFileName(sqlInfo.getFileName());
933        sqlInfoItem.setFilePath(sqlInfo.getFilePath());
934        sqlInfoItem.setSql(buffer.toString());
935        sqlInfoItem.setOriginIndex(index);
936        sqlInfoItem.setOriginLineStart(0);
937        sqlInfoItem.setOriginLineEnd(buffer.toString().split("\n", -1).length - 1);
938        sqlInfoItem.setIndex(databaseMap.get(sqlHash).second.getAndIncrement());
939        sqlInfoItem.setLineStart(lineStart);
940        sqlInfoItem.setLineEnd(databaseMap.get(sqlHash).first.toString().split("\n", -1).length - 1);
941        sqlInfoItem.setGroup(group);
942        sqlInfoItem.setHash(hash);
943
944        if (!sqlInfoMap.containsKey(hash)) {
945            sqlInfoMap.put(hash, new ArrayList<SqlInfo>());
946        }
947        sqlInfoMap.get(hash).add(sqlInfoItem);
948    }
949
950    @Override
951    public void dispose() {
952        ModelBindingManager.remove();
953    }
954
955    private dataflow mergeDataFlows(Map<SqlInfo, dataflow> dataflowMap, long startId) {
956        return DataflowUtility.mergeDataflowsByStartId(dataflowMap.values(), startId);
957    }
958
959    private dataflow analyzeMetadata(String manifestJson, String manifestPath) {
960        try {
961            if (MetadataReader.isSqlflowSharded(manifestJson)) {
962                String baseDir = manifestPath != null ? new File(manifestPath).getParent() : null;
963                SqlflowShardedMetadataAnalyzer analyzer = new SqlflowShardedMetadataAnalyzer(
964                        sqlenv, baseDir);
965                return analyzer.analyzeMetadata(option.getVendor(), manifestJson);
966            } else if (MetadataReader.isSqlflow(manifestJson)) {
967                SqlflowMetadataAnalyzer analyzer = new SqlflowMetadataAnalyzer(sqlenv);
968                return analyzer.analyzeMetadata(option.getVendor(), manifestJson);
969            } else if (MetadataReader.isGrabit(manifestJson)) {
970                GrabitMetadataAnalyzer analyzer = new GrabitMetadataAnalyzer();
971                return analyzer.analyzeMetadata(option.getVendor(), manifestJson);
972            }
973        } catch (Exception e) {
974            logger.error("Failed to analyze metadata", e);
975        }
976        return null;
977    }
978
979    @Override
980    public synchronized dataflow getDataFlow() {
981        if (dataflow != null) {
982            return dataflow;
983        } else if (dataflowString != null) {
984            return XML2Model.loadXML(dataflow.class, dataflowString);
985        }
986        return null;
987    }
988
989    @Override
990    public Map<String, String> getHashSQLMap() {
991        return hashSQLMap;
992    }
993
994    static int calculateMemoryAwarePoolSize(int requestedParallel, int taskCount, long estimatedMemoryPerTaskMB) {
995        int basePoolSize = Math.max(1, requestedParallel < taskCount ? requestedParallel : taskCount);
996
997        long maxMemory = Runtime.getRuntime().maxMemory();
998        if (maxMemory == Long.MAX_VALUE) {
999            return basePoolSize;
1000        }
1001
1002        long reservedMemory = (long) (maxMemory * 0.2);
1003        long usableMemory = maxMemory - reservedMemory;
1004
1005        long estimatedBytesPerTask = estimatedMemoryPerTaskMB * 1024L * 1024L;
1006        int memoryBasedLimit = (int) Math.max(1, usableMemory / estimatedBytesPerTask);
1007
1008        int poolSize = Math.min(basePoolSize, memoryBasedLimit);
1009
1010        if (poolSize < basePoolSize) {
1011            logger.warn("Thread pool size reduced from " + basePoolSize + " to " + poolSize
1012                    + " due to memory constraints. "
1013                    + "(maxMemory=" + (maxMemory / (1024 * 1024)) + "MB, "
1014                    + "usableMemory=" + (usableMemory / (1024 * 1024)) + "MB, "
1015                    + "estimatedMemoryPerTask=" + estimatedMemoryPerTaskMB + "MB)");
1016        }
1017
1018        return poolSize;
1019    }
1020
1021    public static void main(String[] args) throws Exception {
1022        Option option = new Option();
1023        option.setVendor(EDbVendor.dbvmssql);
1024        option.setOutput(false);
1025//        option.setSimpleOutput(true);
1026        File parentDir = new File("C:\\Users\\KK\\Desktop\\sql");
1027        ParallelDataFlowAnalyzer analyzer = new ParallelDataFlowAnalyzer(new File[]{new File("C:\\Users\\KK\\Desktop\\metadata_with_query\\客户原始sql.json")}, option, 5, new File("C:\\Users\\KK\\Desktop\\metadata_with_query"));
1028        analyzer.generateDataFlow(false, true);
1029        dataflow dataflow = analyzer.getDataFlow();
1030//        System.out.println(XML2Model.saveXML(dataflow));
1031//        dataflow dataflow = XML2Model.loadXML(gudusoft.gsqlparser.dlineage.dataflow.model.xml.dataflow.class,new File("D:\\dataflow.xml.zip"));
1032        XML2Model.saveXML(dataflow, new File("D:\\dataflow.xml.zip"));
1033//        dataflow = DataflowUtility.convertToTableLevelDataflow(dataflow);
1034//        dataflow = DataflowUtility.convertTableLevelToFunctionCallDataflow(dataflow, true, EDbVendor.dbvoracle);
1035//        System.out.println(XML2Model.saveXML(dataflow));
1036    }
1037
1038}