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