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.model.*;
009import gudusoft.gsqlparser.dlineage.dataflow.model.json.Coordinate;
010import gudusoft.gsqlparser.dlineage.dataflow.model.xml.*;
011import gudusoft.gsqlparser.dlineage.dataflow.sqlenv.SQLEnvParser;
012import gudusoft.gsqlparser.dlineage.util.*;
013import gudusoft.gsqlparser.sqlenv.TSQLEnv;
014import gudusoft.gsqlparser.sqlenv.parser.TJSONSQLEnvParser;
015import gudusoft.gsqlparser.util.IndexedLinkedHashMap;
016import gudusoft.gsqlparser.util.Logger;
017import gudusoft.gsqlparser.util.LoggerFactory;
018import gudusoft.gsqlparser.util.SQLUtil;
019import gudusoft.gsqlparser.util.json.JSON;
020
021import java.io.File;
022import java.io.FileInputStream;
023import java.io.IOException;
024import java.io.BufferedReader;
025import java.io.InputStreamReader;
026import java.util.zip.GZIPInputStream;
027import java.util.*;
028import java.util.concurrent.*;
029import java.util.concurrent.atomic.AtomicInteger;
030import java.util.stream.Collectors;
031
032public class ParallelDataFlowAnalyzer implements IDataFlowAnalyzer {
033    private static final Logger logger = LoggerFactory.getLogger(ParallelDataFlowAnalyzer.class);
034    private SqlInfo[] sqlInfos;
035    private Option option = new Option();
036    private TSQLEnv sqlenv = null;
037    private dataflow dataflow;
038    private String dataflowString;
039    private List<ErrorInfo> errorInfos = new CopyOnWriteArrayList<ErrorInfo>();
040    private IndexedLinkedHashMap<String, List<SqlInfo>> sqlInfoMap = new IndexedLinkedHashMap<String, List<SqlInfo>>();
041    private final Map<String, String> hashSQLMap = new HashMap();
042
043    public ParallelDataFlowAnalyzer(SqlInfo[] sqlInfos, EDbVendor dbVendor, boolean simpleOutput) {
044        this.sqlInfos = sqlInfos;
045        option.setVendor(dbVendor);
046        option.setSimpleOutput(simpleOutput);
047        ModelBindingManager.setGlobalOption(option);
048    }
049
050    public ParallelDataFlowAnalyzer(String[] sqlContents, EDbVendor dbVendor, boolean simpleOutput, String defaultServer, String defaultDatabase, String defaltSchema) {
051        SqlInfo[] sqlInfos = new SqlInfo[sqlContents.length];
052        for (int i = 0; i < sqlContents.length; i++) {
053            SqlInfo info = new SqlInfo();
054            info.setSql(sqlContents[i]);
055            info.setOriginIndex(0);
056            sqlInfos[i] = info;
057        }
058        option.setVendor(dbVendor);
059        option.setSimpleOutput(simpleOutput);
060                option.setDefaultServer(defaultServer);
061                option.setDefaultDatabase(defaultDatabase);
062                option.setDefaultSchema(defaltSchema);
063        this.sqlInfos = convertSQL(dbVendor, JSON.toJSONString(sqlInfos)).toArray(new SqlInfo[0]);
064    }
065    
066    public ParallelDataFlowAnalyzer(String[] sqlContents, EDbVendor dbVendor, boolean simpleOutput) {
067        SqlInfo[] sqlInfos = new SqlInfo[sqlContents.length];
068        for (int i = 0; i < sqlContents.length; i++) {
069            SqlInfo info = new SqlInfo();
070            info.setSql(sqlContents[i]);
071            info.setOriginIndex(0);
072            sqlInfos[i] = info;
073        }
074        option.setVendor(dbVendor);
075        option.setSimpleOutput(simpleOutput);
076        this.sqlInfos = convertSQL(dbVendor, JSON.toJSONString(sqlInfos)).toArray(new SqlInfo[0]);
077    }
078
079    public ParallelDataFlowAnalyzer(SqlInfo[] sqlInfos, Option option) {
080        this.sqlInfos = sqlInfos;
081        this.option = option;
082        ModelBindingManager.setGlobalOption(option);
083    }
084
085    public ParallelDataFlowAnalyzer(File[] sqlFiles, Option option) {
086        SqlInfo[] sqlInfos = new SqlInfo[sqlFiles.length];
087        for (int i = 0; i < sqlFiles.length; i++) {
088            SqlInfo info = new SqlInfo();
089            info.setSql(SQLUtil.getFileContent(sqlFiles[i]));
090            info.setFileName(sqlFiles[i].getName());
091            info.setFilePath(sqlFiles[i].getAbsolutePath());
092            info.setOriginIndex(0);
093            sqlInfos[i] = info;
094        }
095        this.sqlInfos = sqlInfos;
096        this.option = option;
097        ModelBindingManager.setGlobalOption(option);
098    }
099    
100    public ParallelDataFlowAnalyzer(File[] sqlFiles, Option option, int splitSizeMB, File splitDir) {
101        List<SqlInfo> sqlInfoList = new ArrayList<SqlInfo>();
102        for (int i = 0; i < sqlFiles.length; i++) {
103            try {
104                // Ensure split directory exists
105                if (!splitDir.exists() && !splitDir.mkdirs()) {
106                    throw new IOException("Failed to create split directory: " + splitDir.getAbsolutePath());
107                }
108                
109                // Split the file if it's large
110                List<File> splitFiles = gudusoft.gsqlparser.util.FileSplitter.splitFile(sqlFiles[i], splitDir, splitSizeMB, option.getVendor());
111                
112                if (splitFiles.isEmpty()) {
113                    // If no split files were created, process the original file
114                    SqlInfo info = new SqlInfo();
115                    info.setSql(SQLUtil.getFileContent(sqlFiles[i]));
116                    info.setFileName(sqlFiles[i].getName());
117                    info.setFilePath(sqlFiles[i].getAbsolutePath());
118                    info.setOriginIndex(0);
119                    sqlInfoList.add(info);
120                } else {
121                    // Process each split file
122                    for (File splitFile : splitFiles) {
123                        SqlInfo info = new SqlInfo();
124                        info.setSql(SQLUtil.getFileContent(splitFile));
125                        info.setFileName(splitFile.getName());
126                        info.setFilePath(splitFile.getAbsolutePath());
127                        info.setOriginIndex(0);
128                        sqlInfoList.add(info);
129                    }
130                }
131            } catch (Exception e) {
132                sqlInfoList.clear();
133                logger.error("Error splitting file: " + sqlFiles[i].getAbsolutePath(), e);
134                // Fallback to original file if splitting fails
135                SqlInfo info = new SqlInfo();
136                info.setSql(SQLUtil.getFileContent(sqlFiles[i]));
137                info.setFileName(sqlFiles[i].getName());
138                info.setFilePath(sqlFiles[i].getAbsolutePath());
139                info.setOriginIndex(0);
140                sqlInfoList.add(info);
141            }
142        }
143        this.sqlInfos = sqlInfoList.toArray(new SqlInfo[0]);
144        this.option = option;
145        ModelBindingManager.setGlobalOption(option);
146    }
147    
148    public ParallelDataFlowAnalyzer(File[] sqlFiles, Option option, int splitSizeMB, String splitDirPath) {
149        this(sqlFiles, option, splitSizeMB, new File(splitDirPath));
150    }
151    
152    protected List<SqlInfo> convertSQL(EDbVendor vendor, String json) {
153                List<SqlInfo> sqlInfos = new ArrayList<SqlInfo>();
154                List sqlContents = (List) JSON.parseObject(json);
155                for (int j = 0; j < sqlContents.size(); j++) {
156                        Map sqlContent = (Map) sqlContents.get(j);
157                        String sql = (String) sqlContent.get("sql");
158                        String fileName = (String) sqlContent.get("fileName");
159                        String filePath = (String) sqlContent.get("filePath");
160                        if (sql != null && sql.trim().startsWith("{")) {
161                                if (sql.indexOf("createdBy") != -1) {
162                                        if (this.sqlenv == null) {
163                                                TSQLEnv[] sqlenvs = new TJSONSQLEnvParser(option.getDefaultServer(), option.getDefaultDatabase(), option.getDefaultSchema()).parseSQLEnv(vendor, sql);
164                                                if (sqlenvs != null) {
165                                                        this.sqlenv = sqlenvs[0];
166                                                }
167                                        }
168                        if (sql.toLowerCase().indexOf("sqldep") != -1 || sql.toLowerCase().indexOf("grabit") != -1 ) {
169                                Map queryObject = (Map) JSON.parseObject(sql);
170                                        List querys = (List) queryObject.get("queries");
171                                        if (querys != null) {
172                                                for (int i = 0; i < querys.size(); i++) {
173                                                        Map object = (Map) querys.get(i);
174                                                        SqlInfo info = new SqlInfo();
175                                                        info.setSql(JSON.toJSONString(object));
176                                                        info.setFileName(fileName);
177                                                        info.setFilePath(filePath);
178                                                        info.setOriginIndex(i);
179                                                        sqlInfos.add(info);
180                                                }
181                                                queryObject.remove("queries");
182                                                SqlInfo info = new SqlInfo();
183                                                info.setSql(JSON.toJSONString(queryObject));
184                                                info.setFileName(fileName);
185                                                info.setFilePath(filePath);
186                                                info.setOriginIndex(querys.size());
187                                                sqlInfos.add(info);
188                                        } else {
189                                                SqlInfo info = new SqlInfo();
190                                                info.setSql(JSON.toJSONString(queryObject));
191                                                info.setFileName(fileName);
192                                                info.setFilePath(filePath);
193                                                info.setOriginIndex(0);
194                                                sqlInfos.add(info);
195                                        }
196                        }
197                        else if (sql.toLowerCase().indexOf("sqlflow") != -1) {
198                                                Map sqlflow = (Map) JSON.parseObject(sql);
199                                                if ("sqlflow-sharded".equals(sqlflow.get("format"))) {
200                                                        String baseDir = null;
201                                                        if (filePath != null) {
202                                                                baseDir = new File(filePath).getParent();
203                                                        }
204                                                        String sourceCompression = (String) sqlflow.get("sourceCompression");
205                                                        List<Map> servers = (List<Map>) sqlflow.get("servers");
206                                                        if (servers != null) {
207                                                                for (Map serverObject : servers) {
208                                                                        String name = (String) serverObject.get("name");
209                                                                        String dbVendor = (String) serverObject.get("dbVendor");
210                                                                        List<Map> databases = (List<Map>) serverObject.get("databases");
211                                                                        if (databases != null) {
212                                                                                for (Map database : databases) {
213                                                                                        Map source = (Map) database.get("source");
214                                                                                        if (source != null) {
215                                                                                                String sourcePath = (String) source.get("path");
216                                                                                                if (sourcePath != null && baseDir != null) {
217                                                    File sourceFile = new File(baseDir, sourcePath);
218                                                    String fullPath = sourceFile.getAbsolutePath();
219                                                    if ("block".equals(sourceCompression)) {
220                                                                                                                readGzipBlockSource(fullPath, fileName, filePath, name, dbVendor, sqlInfos);
221                                                                                                        } else {
222                                                                                                                String sourceContent = SQLUtil.getFileContent(fullPath);
223                                                                                                                if (sourceContent != null) {
224                                                                                                                        String[] lines = sourceContent.split("\\r?\\n");
225                                                                                                                        for (int i = 0; i < lines.length; i++) {
226                                                                                                                                String line = lines[i].trim();
227                                                                                                                                if (line.isEmpty()) {
228                                                                                                                                        continue;
229                                                                                                                                }
230                                                                                                                                try {
231                                                                                                                                        Map sourceObject = (Map) JSON.parseObject(line);
232                                                                                                                                        String sourceCode = (String) sourceObject.get("sourceCode");
233                                                                                                                                        if (sourceCode != null && !sourceCode.isEmpty()) {
234                                                                                                                                                SqlInfo info = new SqlInfo();
235                                                                                                                                                info.setSql(sourceCode);
236                                                                                                                                                info.setFileName(sourceFile.getName());
237                                                                                                                                                info.setFilePath(sourceFile.getAbsolutePath());
238                                                                                                                                                info.setOriginIndex(i);
239                                                                                                                                                info.setDbVendor(dbVendor);
240                                                                                                                                                info.setServer(name);
241                                                                                                                                                sqlInfos.add(info);
242                                                                                                                                        }
243                                                                                                                                } catch (Exception e) {
244                                                                                                                                        logger.warn("Parse source jsonl line failed.", e);
245                                                                                                                                }
246                                                                                                                        }
247                                                                                                                }
248                                                                                                        }
249                                                                                                }
250                                                                                        }
251                                                                                }
252                                                                        }
253                                                                        SqlInfo serverInfo = new SqlInfo();
254                                                                        serverInfo.setSql(JSON.toJSONString(serverObject));
255                                                                        serverInfo.setFileName(fileName);
256                                                                        serverInfo.setFilePath(filePath);
257                                                                        serverInfo.setDbVendor(dbVendor);
258                                                                        serverInfo.setServer(name);
259                                                                        sqlInfos.add(serverInfo);
260                                                                }
261                                                        }
262                                                } else {
263                                                        List<Map> servers = (List<Map>) sqlflow.get("servers");
264                                                        if (servers != null) {
265                                                                for (Map queryObject : servers) {
266                                                                        String name = (String) queryObject.get("name");
267                                                                        String dbVendor = (String) queryObject.get("dbVendor");
268                                                                        List querys = (List) queryObject.get("queries");
269                                                                        if (querys != null) {
270                                                                                for (int i = 0; i < querys.size(); i++) {
271                                                                                        Map object = (Map) querys.get(i);
272                                                                                        SqlInfo info = new SqlInfo();
273                                                                                        info.setSql(JSON.toJSONString(object));
274                                                                                        info.setFileName(fileName);
275                                                                                        info.setFilePath(filePath);
276                                                                                        info.setOriginIndex(i);
277                                                                                        info.setDbVendor(dbVendor);
278                                                                                        info.setServer(name);
279                                                                                        sqlInfos.add(info);
280                                                                                }
281                                                                                queryObject.remove("queries");
282                                                                                SqlInfo info = new SqlInfo();
283                                                                                info.setSql(JSON.toJSONString(queryObject));
284                                                                                info.setFileName(fileName);
285                                                                                info.setFilePath(filePath);
286                                                                                info.setOriginIndex(querys.size());
287                                                                                info.setDbVendor(dbVendor);
288                                                                                info.setServer(filePath);
289                                                                                sqlInfos.add(info);
290                                                                        } else {
291                                                                                SqlInfo info = new SqlInfo();
292                                                                                info.setSql(JSON.toJSONString(queryObject));
293                                                                                info.setFileName(fileName);
294                                                                                info.setFilePath(filePath);
295                                                                                info.setOriginIndex(0);
296                                                                                sqlInfos.add(info);
297                                                                        }
298                                                                }
299                                                        }
300                                                }
301                                        }
302                    }
303                        } else if (sql != null) {
304                                SqlInfo info = new SqlInfo();
305                                info.setSql(sql);
306                                info.setFileName(fileName);
307                                info.setFilePath(filePath);
308                                info.setOriginIndex(0);
309                                sqlInfos.add(info);
310                        }
311                }
312                return sqlInfos;
313        }
314
315        private void readGzipBlockSource(String fullPath, String fileName, String filePath, String serverName, String dbVendor, List<SqlInfo> sqlInfos) {
316                try (FileInputStream fis = new FileInputStream(fullPath);
317                         GZIPInputStream gzis = new GZIPInputStream(fis);
318                         BufferedReader reader = new BufferedReader(new InputStreamReader(gzis, "UTF-8"))) {
319                        String line;
320                        int j = 0;
321                        while ((line = reader.readLine()) != null) {
322                                line = line.trim();
323                                if (line.isEmpty()) {
324                                        continue;
325                                }
326                                try {
327                                        Map sourceObject = (Map) JSON.parseObject(line);
328                                        String sourceCode = (String) sourceObject.get("sourceCode");
329                                        if (sourceCode != null && !sourceCode.isEmpty()) {
330                                                SqlInfo info = new SqlInfo();
331                                                info.setSql(sourceCode);
332                                                info.setFileName(fileName);
333                                                info.setFilePath(filePath);
334                                                info.setOriginIndex(j);
335                                                info.setDbVendor(dbVendor);
336                                                info.setServer(serverName);
337                                                sqlInfos.add(info);
338                                        }
339                                } catch (Exception e) {
340                                        logger.warn("Parse gzip source jsonl line failed.", e);
341                                }
342                                j++;
343                        }
344                } catch (Exception e) {
345                        logger.warn("Read gzip source file failed: " + fullPath, e);
346                }
347        }
348
349    @Override
350    public boolean isIgnoreRecordSet() {
351        return option.isIgnoreRecordSet();
352    }
353
354    @Override
355    public void setIgnoreRecordSet(boolean ignoreRecordSet) {
356        option.setIgnoreRecordSet(ignoreRecordSet);
357    }
358
359    @Override
360    public boolean isSimpleShowTopSelectResultSet() {
361        return option.isSimpleShowTopSelectResultSet();
362    }
363
364    @Override
365    public void setSimpleShowTopSelectResultSet(boolean simpleShowTopSelectResultSet) {
366        option.setSimpleShowTopSelectResultSet(simpleShowTopSelectResultSet);
367    }
368
369    @Override
370    public boolean isSimpleShowFunction() {
371        return option.isSimpleShowFunction();
372    }
373
374    @Override
375    public void setSimpleShowFunction(boolean simpleShowFunction) {
376        option.setSimpleShowFunction(simpleShowFunction);
377    }
378
379    @Override
380    public boolean isShowJoin() {
381        return option.isShowJoin();
382    }
383
384    @Override
385    public void setShowJoin(boolean showJoin) {
386        option.setShowJoin(showJoin);
387    }
388
389    @Override
390    public boolean isShowImplicitSchema() {
391        return option.isShowImplicitSchema();
392    }
393
394    @Override
395    public void setShowImplicitSchema(boolean showImplicitSchema) {
396        option.setShowImplicitSchema(showImplicitSchema);
397    }
398
399    @Override
400    public boolean isShowConstantTable() {
401        return option.isShowConstantTable();
402    }
403
404    @Override
405    public void setShowConstantTable(boolean showConstantTable) {
406        option.setShowConstantTable(showConstantTable);
407    }
408
409    @Override
410    public boolean isShowCountTableColumn() {
411        return option.isShowCountTableColumn();
412    }
413
414    @Override
415    public void setShowCountTableColumn(boolean showCountTableColumn) {
416        option.setShowCountTableColumn(showCountTableColumn);
417    }
418
419    @Override
420    public boolean isTransform() {
421        return option.isTransform();
422    }
423
424    @Override
425    public void setTransform(boolean transform) {
426        option.setTransform(transform);
427        if (option.isTransformCoordinate()) {
428            option.setTransform(true);
429        }
430    }
431
432    @Override
433    public boolean isTransformCoordinate() {
434        return option.isTransformCoordinate();
435    }
436
437    @Override
438    public void setTransformCoordinate(boolean transformCoordinate) {
439        option.setTransformCoordinate(transformCoordinate);
440        if (transformCoordinate) {
441            option.setTransform(true);
442        }
443    }
444
445    @Override
446    public boolean isLinkOrphanColumnToFirstTable() {
447        return option.isLinkOrphanColumnToFirstTable();
448    }
449
450    @Override
451    public void setLinkOrphanColumnToFirstTable(boolean linkOrphanColumnToFirstTable) {
452        option.setLinkOrphanColumnToFirstTable(linkOrphanColumnToFirstTable);
453    }
454
455    @Override
456    public boolean isIgnoreCoordinate() {
457        return option.isIgnoreCoordinate();
458    }
459
460    @Override
461    public void setIgnoreCoordinate(boolean ignoreCoordinate) {
462        option.setIgnoreCoordinate(ignoreCoordinate);
463    }
464
465    @Override
466    public void setHandleListener(DataFlowHandleListener listener) {
467        option.setHandleListener(listener);
468    }
469
470    @Override
471    public void setSqlEnv(TSQLEnv sqlenv) {
472        this.sqlenv = sqlenv;
473    }
474
475    @Override
476    public void setOption(Option option) {
477        this.option = option;
478    }
479
480    @Override
481    public Option getOption() {
482        return option;
483    }
484
485    @Override
486    public List<ErrorInfo> getErrorMessages() {
487        return errorInfos;
488    }
489
490    @Override
491    public synchronized String generateSqlInfos() {
492        return JSON.toJSONString(sqlInfoMap);
493    }
494
495    @Override
496    public synchronized String generateDataFlow() {
497        return generateDataFlow(false);
498    }
499
500    @Override
501    public Map<String, List<SqlInfo>> getSqlInfos() {
502        return sqlInfoMap;
503    }
504
505    @Override
506    /**
507     * @deprecated please use SqlInfoHelper.getSelectedDbObjectInfo
508     */
509    public DbObjectPosition getSelectedDbObjectInfo(Coordinate start, Coordinate end) {
510        if (start == null || end == null) {
511            throw new IllegalArgumentException("Coordinate can't be null.");
512        }
513
514        String hashCode = start.getHashCode();
515
516        if (hashCode == null) {
517            throw new IllegalArgumentException("Coordinate hashcode can't be null.");
518        }
519
520        int dbObjectStartLine = (int) start.getX() - 1;
521        int dbObjectStarColumn = (int) start.getY() - 1;
522        int dbObjectEndLine = (int) end.getX() - 1;
523        int dbObjectEndColumn = (int) end.getY() - 1;
524        List<SqlInfo> sqlInfoList;
525        if (hashCode.matches("\\d+")) {
526            sqlInfoList = sqlInfoMap.getValueAtIndex(Integer.valueOf(hashCode));
527        } else {
528            sqlInfoList = sqlInfoMap.get(hashCode);
529        }
530        for (int j = 0; j < sqlInfoList.size(); j++) {
531            SqlInfo sqlInfo = sqlInfoList.get(j);
532            int startLine = sqlInfo.getLineStart();
533            int endLine = sqlInfo.getLineEnd();
534            if (dbObjectStartLine >= startLine && dbObjectStartLine <= endLine) {
535                DbObjectPosition position = new DbObjectPosition();
536                position.setFile(sqlInfo.getFileName());
537                position.setFilePath(sqlInfo.getFilePath());
538                position.setSql(sqlInfo.getSql());
539                position.setIndex(sqlInfo.getOriginIndex());
540                List<Pair<Integer, Integer>> positions = position.getPositions();
541                positions.add(new Pair<Integer, Integer>(
542                        dbObjectStartLine - startLine + sqlInfo.getOriginLineStart() + 1, dbObjectStarColumn + 1));
543                positions.add(new Pair<Integer, Integer>(dbObjectEndLine - startLine + sqlInfo.getOriginLineStart() + 1,
544                        dbObjectEndColumn + 1));
545                return position;
546            }
547        }
548        return null;
549    }
550
551    @Override
552    public synchronized String generateDataFlow(final boolean withExtraInfo) {
553        return generateDataFlow(withExtraInfo, true);
554    }
555    
556    public synchronized String generateDataFlow(final boolean withExtraInfo, boolean useSaveMemoryMode) {
557        sqlInfoMap.clear();
558        errorInfos.clear();
559        Map<String, Pair3<StringBuilder, AtomicInteger, String>> databaseMap = new LinkedHashMap<String, Pair3<StringBuilder, AtomicInteger, String>>();
560        for (int i = 0; i < sqlInfos.length; i++) {
561            SqlInfo sqlInfo = sqlInfos[i];
562                        if (sqlInfo != null && sqlInfo.getSql() == null && sqlInfo.getFilePath() != null) {
563                                sqlInfo.setSql(SQLUtil.getFileContent(sqlInfo.getFilePath()));
564                        }
565            if (sqlInfo != null && sqlInfo.getSql() == null && sqlInfo.getFileName() != null) {
566                sqlInfo.setSql(SQLUtil.getFileContent(sqlInfo.getFileName()));
567            }
568            if (sqlInfo == null || sqlInfo.getSql() == null) {
569                sqlInfoMap.put(String.valueOf(i), new ArrayList<SqlInfo>());
570                continue;
571            }
572            String sql = sqlInfo.getSql();
573            if (sql != null && sql.trim().startsWith("{")) {
574                if (MetadataReader.isGrabit(sql) || MetadataReader.isSqlflow(sql) || MetadataReader.isSqlflowSharded(sql)) {
575                    String hash = SHA256.getMd5(sql);
576                    String fileHash = SHA256.getMd5(hash);
577                    if (!sqlInfoMap.containsKey(fileHash)) {
578                        sqlInfoMap.put(fileHash, new ArrayList<SqlInfo>());
579                        sqlInfoMap.get(fileHash).add(sqlInfo);
580                    }
581                } else {
582                    Map queryObject = (Map) JSON.parseObject(sql);
583                    appendSqlInfo(databaseMap, i, sqlInfo, queryObject);
584                }
585            } else {
586                String content = sql;
587                String hash = SHA256.getMd5(content);
588                String fileHash = SHA256.getMd5(hash);
589                if (!sqlInfoMap.containsKey(fileHash)) {
590                    sqlInfoMap.put(fileHash, new ArrayList<SqlInfo>());
591
592                    String database = TSQLEnv.DEFAULT_DB_NAME;
593                    String schema = TSQLEnv.DEFAULT_SCHEMA_NAME;
594                    if(sqlenv!=null) {
595                        // Prefer the per-analysis Option default database over the attached
596                        // env's default catalog (mirrors DataFlowAnalyzer), so a shared
597                        // multi-catalog TSQLEnv can be reused across analyses with different
598                        // default databases.
599                        if (!SQLUtil.isEmpty(option.getDefaultDatabase())
600                                && !TSQLEnv.DEFAULT_DB_NAME.equals(option.getDefaultDatabase())) {
601                            database = option.getDefaultDatabase();
602                        } else {
603                            database = sqlenv.getDefaultCatalogName();
604                        }
605                        if(database == null) {
606                            database = TSQLEnv.DEFAULT_DB_NAME;
607                        }
608                        schema = sqlenv.getDefaultSchemaName();
609                        if(schema == null) {
610                            schema = TSQLEnv.DEFAULT_SCHEMA_NAME;
611                        }
612                    }
613                    boolean supportCatalog = TSQLEnv.supportCatalog(option.getVendor());
614                    boolean supportSchema = TSQLEnv.supportSchema(option.getVendor());
615                    StringBuilder builder = new StringBuilder();
616                    if (supportCatalog) {
617                        builder.append(database);
618                    }
619                    if (supportSchema) {
620                        if (builder.length() > 0) {
621                            builder.append(".");
622                        }
623                        builder.append(schema);
624                    }
625                    String group = builder.toString();
626                    SqlInfo sqlInfoItem = new SqlInfo();
627                    sqlInfoItem.setFileName(sqlInfo.getFileName());
628                    sqlInfoItem.setFilePath(sqlInfo.getFilePath());
629                    sqlInfoItem.setSql(sqlInfo.getSql());
630                    sqlInfoItem.setOriginIndex(0);
631                    sqlInfoItem.setOriginLineStart(0);
632                    sqlInfoItem.setOriginLineEnd(sqlInfo.getSql().split("\n").length - 1);
633                    sqlInfoItem.setIndex(0);
634                    sqlInfoItem.setLineStart(0);
635                    sqlInfoItem.setLineEnd(sqlInfo.getSql().split("\n").length - 1);
636                    sqlInfoItem.setHash(fileHash);
637                    sqlInfoItem.setGroup(group);
638
639                    sqlInfoMap.get(fileHash).add(sqlInfoItem);
640                }
641            }
642        }
643
644        final TSQLEnv[] env = new TSQLEnv[]{sqlenv};
645        if (sqlenv == null) {
646                TSQLEnv[] envs = new SQLEnvParser(option.getDefaultServer(), option.getDefaultDatabase(), option.getDefaultSchema()).parseSQLEnv(option.getVendor(), sqlInfos);
647                        if (envs != null && envs.length > 0) {
648                                env[0] = envs[0];
649                        }
650        }
651        ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors
652                .newFixedThreadPool(option.getParallel() < sqlInfos.length
653                        ? option.getParallel()
654                        : sqlInfos.length);
655        
656        List<File> tempFiles = Collections.synchronizedList(new ArrayList<File>());
657        final Map<SqlInfo, dataflow> dataflowMap = useSaveMemoryMode ? null : new ConcurrentHashMap<>();
658        
659        try {
660            final CountDownLatch latch = new CountDownLatch(sqlInfos.length);
661
662            logger.info("start parallel analyze " + sqlInfos.length + " sqlinfos");
663
664            for (int i = 0; i < sqlInfos.length; i++) {
665                final SqlInfo[] sqlInfoCopy = new SqlInfo[sqlInfos.length];
666                final SqlInfo item = sqlInfos[i];
667                sqlInfoCopy[i] = item;
668                final Option optionCopy = (Option) option.clone();
669                optionCopy.setStartId(5000000L * i);
670                optionCopy.setOutput(false);
671                final int index = i;
672                Runnable task = new Runnable() {
673                    @Override
674                    public void run() {
675                        try {
676                            logger.info("start analyze sqlinfo[" + index + "]" + (sqlInfos[index].getFilePath() != null ? ", file name = " + new File(sqlInfos[index].getFilePath()).getName() : ""));
677                            DataFlowAnalyzer analyzer = new DataFlowAnalyzer(sqlInfoCopy, optionCopy);
678                            analyzer.setSqlEnv(env[0]);
679                            analyzer.generateDataFlow(withExtraInfo);
680                            dataflow dataflow = analyzer.getDataFlow();
681                            if (dataflow == null) {
682                                logger.warn("analyze sqlinfo[" + index + "] done, but dataflow is null" + (sqlInfos[index].getFilePath() != null ? ", file name = " + new File(sqlInfos[index].getFilePath()).getName() : ""));
683                                errorInfos.addAll(analyzer.getErrorMessages());
684                                analyzer.dispose();
685                                return;
686                            }
687                            logger.info("analyze sqlinfo[" + index + "] done, relation count: " + dataflow.getRelationships().size()+", error count: "+ analyzer.getErrorMessages().size());
688                            if (analyzer.getErrorMessages().size() > 10000) {
689                                dataflow.setErrors(new ArrayList<>(dataflow.getErrors().subList(0, 10000)));
690                                errorInfos.addAll(analyzer.getErrorMessages().subList(0, 10000));
691                                logger.warn("Too many errors in dataflow ("+dataflow.getErrors().size()+"), truncating to first 10000 errors" + (sqlInfos[index].getFileName() != null ? ", file name = " + sqlInfos[index].getFileName() : ""));
692                            }
693                            else{
694                                errorInfos.addAll(analyzer.getErrorMessages());
695                            }
696                            
697                            if (useSaveMemoryMode) {
698                                File tempFile = File.createTempFile("dataflow_" + index + "_" + System.currentTimeMillis() + "_", ".xml.zip");
699                                XML2Model.saveXML(dataflow, tempFile);
700                                tempFiles.add(tempFile);
701                            } else {
702                                dataflowMap.put(item, dataflow);
703                            }
704                            
705                            hashSQLMap.putAll(analyzer.getHashSQLMap());
706                            analyzer.dispose();
707                        }
708                        catch (Exception e) {
709                            logger.error("analyze sqlinfo[" + index + "] failed.", e);
710                        }
711                        finally {
712                            latch.countDown();
713                        }
714                    }
715                };
716                executor.submit(task);
717            }
718            latch.await();
719        } catch (Exception e) {
720            logger.error("execute task failed.", e);
721        }
722        executor.shutdown();
723
724        ModelBindingManager modelManager = new ModelBindingManager();
725        modelManager.setGlobalVendor(option.getVendor());
726        modelManager.setGlobalOption(option);
727        ModelBindingManager.set(modelManager);
728        
729        if (useSaveMemoryMode) {
730            // 使用节约内存模式，迭代合并
731            logger.info("start merge dataflow, dataflow count: " + tempFiles.size()+", useSaveMemoryMode: "+ useSaveMemoryMode+", gsp version: "+ TBaseType.versionid);
732            this.dataflow = iterativeMergeDataFlows(tempFiles, 5000000L * sqlInfos.length);
733            if (this.dataflow != null && this.dataflow.getRelationships() != null) {
734                logger.info("merge dataflow done, dataflow count: " + tempFiles.size() + ", relation count: " + this.dataflow.getRelationships().size());
735            }
736            for (File tempFile : tempFiles) {
737                tempFile.delete();
738            }
739        } else {
740            // 保持原有逻辑
741            logger.info("start merge dataflow, dataflow count: " + dataflowMap.size()+", useSaveMemoryMode: "+ useSaveMemoryMode+", gsp version: "+ TBaseType.versionid);
742            this.dataflow = mergeDataFlows(dataflowMap, 5000000L * sqlInfos.length);
743            if (this.dataflow != null && this.dataflow.getRelationships() != null) {
744                logger.info("merge dataflow done, dataflow count: " + tempFiles.size() + ", relation count: " + this.dataflow.getRelationships().size());
745            }
746            dataflowMap.clear();
747        }
748        
749        if (this.dataflow != null) {
750            logger.info("merge done, relation count: " + this.dataflow.getRelationships().size());
751        }
752        
753        if (dataflow != null && option.isOutput()) {
754            if (option.isTextFormat()) {
755                dataflowString = DataFlowAnalyzer.getTextOutput(dataflow);
756            } else {
757                try {
758                    dataflowString = XML2Model.saveXML(dataflow);
759                }catch (Exception e){
760                    logger.error("save dataflow as xml failed.", e);
761                    dataflowString = null;
762                }
763            }
764        }
765        ModelBindingManager.remove();
766        return dataflowString;
767    }
768    
769    private dataflow iterativeMergeDataFlows(List<File> tempFiles, long startId) {
770        return DataflowUtility.iterativeMergeDataflowsFromFilesByStartId(tempFiles, startId);
771    }
772
773    private void appendSqlInfo(Map<String, Pair3<StringBuilder, AtomicInteger, String>> databaseMap, int index,
774                               SqlInfo sqlInfo, Map queryObject) {
775        EDbVendor vendor = option.getVendor();
776        if(!SQLUtil.isEmpty(sqlInfo.getDbVendor())){
777            vendor = EDbVendor.valueOf(sqlInfo.getDbVendor());
778        }
779
780        boolean supportCatalog = TSQLEnv.supportCatalog(vendor);
781        boolean supportSchema = TSQLEnv.supportSchema(vendor);
782
783        String groupName = (String) queryObject.get("groupName");
784        if (DlineageUtil.isProcedureExcluded(groupName)) {
785            return;
786        }
787
788        String content = (String) queryObject.get("sourceCode");
789        if (SQLUtil.isEmpty(content)) {
790            return;
791        }
792
793        StringBuilder builder = new StringBuilder();
794        if (supportCatalog) {
795            String database = (String) queryObject.get("database");
796            if (database.indexOf(".") != -1) {
797                String delimitedChar = TSQLEnv.delimitedChar(vendor);
798                database = delimitedChar + SQLUtil.trimColumnStringQuote(database) + delimitedChar;
799            }
800            builder.append(database);
801        }
802        if (supportSchema) {
803            String schema = (String) queryObject.get("schema");
804            if (schema.indexOf(".") != -1) {
805                String delimitedChar = TSQLEnv.delimitedChar(vendor);
806                schema = delimitedChar + SQLUtil.trimColumnStringQuote(schema) + delimitedChar;
807            }
808            if (builder.length() > 0) {
809                builder.append(".");
810            }
811            builder.append(schema);
812        }
813        String group = builder.toString();
814        String sqlHash = SHA256.getMd5(content);
815        String hash = SHA256.getMd5(sqlHash);
816        if (!databaseMap.containsKey(sqlHash)) {
817            databaseMap.put(sqlHash,
818                    new Pair3<StringBuilder, AtomicInteger, String>(new StringBuilder(), new AtomicInteger(), group));
819        }
820        String delimiterChar = String.valueOf(TGSqlParser.getDelimiterChar(option.getVendor()));
821        StringBuilder buffer = new StringBuilder(content);
822        if (content.trim().endsWith(delimiterChar) || content.trim().endsWith(";")) {
823            buffer.append("\n");
824        } else if(vendor == EDbVendor.dbvredshift
825                || vendor == EDbVendor.dbvgaussdb
826                || vendor == EDbVendor.dbvedb
827                || vendor == EDbVendor.dbvpostgresql
828                || vendor == EDbVendor.dbvmysql
829                || vendor == EDbVendor.dbvteradata){
830            buffer.append("\n\n-- " + TBaseType.sqlflow_stmt_delimiter_str + "\n\n");
831        } else{
832            SQLUtil.endTrim(buffer);
833            buffer.append(";").append("\n");
834        }
835
836        int lineStart = databaseMap.get(sqlHash).first.toString().split("\n", -1).length - 1;
837        if (databaseMap.get(sqlHash).first.toString().length() == 0) {
838            lineStart = 0;
839        }
840        databaseMap.get(sqlHash).first.append(buffer.toString());
841        SqlInfo sqlInfoItem = new SqlInfo();
842        sqlInfoItem.setFileName(sqlInfo.getFileName());
843        sqlInfoItem.setFilePath(sqlInfo.getFilePath());
844        sqlInfoItem.setSql(buffer.toString());
845        sqlInfoItem.setOriginIndex(index);
846        sqlInfoItem.setOriginLineStart(0);
847        sqlInfoItem.setOriginLineEnd(buffer.toString().split("\n", -1).length - 1);
848        sqlInfoItem.setIndex(databaseMap.get(sqlHash).second.getAndIncrement());
849        sqlInfoItem.setLineStart(lineStart);
850        sqlInfoItem.setLineEnd(databaseMap.get(sqlHash).first.toString().split("\n", -1).length - 1);
851        sqlInfoItem.setGroup(group);
852        sqlInfoItem.setHash(hash);
853
854        if (!sqlInfoMap.containsKey(hash)) {
855            sqlInfoMap.put(hash, new ArrayList<SqlInfo>());
856        }
857        sqlInfoMap.get(hash).add(sqlInfoItem);
858    }
859
860    @Override
861    public void dispose() {
862        ModelBindingManager.remove();
863    }
864
865    private dataflow mergeDataFlows(Map<SqlInfo, dataflow> dataflowMap, long startId) {
866        return DataflowUtility.mergeDataflowsByStartId(dataflowMap.values(), startId);
867    }
868
869    @Override
870    public synchronized dataflow getDataFlow() {
871        if (dataflow != null) {
872            return dataflow;
873        } else if (dataflowString != null) {
874            return XML2Model.loadXML(dataflow.class, dataflowString);
875        }
876        return null;
877    }
878
879    @Override
880    public Map<String, String> getHashSQLMap() {
881        return hashSQLMap;
882    }
883
884    public static void main(String[] args) throws Exception {
885        Option option = new Option();
886        option.setVendor(EDbVendor.dbvmssql);
887        option.setOutput(false);
888//        option.setSimpleOutput(true);
889        File parentDir = new File("C:\\Users\\KK\\Desktop\\sql");
890        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"));
891        analyzer.generateDataFlow(false, true);
892        dataflow dataflow = analyzer.getDataFlow();
893//        System.out.println(XML2Model.saveXML(dataflow));
894//        dataflow dataflow = XML2Model.loadXML(gudusoft.gsqlparser.dlineage.dataflow.model.xml.dataflow.class,new File("D:\\dataflow.xml.zip"));
895        XML2Model.saveXML(dataflow, new File("D:\\dataflow.xml.zip"));
896//        dataflow = DataflowUtility.convertToTableLevelDataflow(dataflow);
897//        dataflow = DataflowUtility.convertTableLevelToFunctionCallDataflow(dataflow, true, EDbVendor.dbvoracle);
898//        System.out.println(XML2Model.saveXML(dataflow));
899    }
900
901}