Source code

001package gudusoft.gsqlparser.dlineage.dataflow.metadata.sqlflow.sharded;
002
003import gudusoft.gsqlparser.dlineage.util.DlineageUtil;
004import gudusoft.gsqlparser.EDbVendor;
005import gudusoft.gsqlparser.sqlenv.ESQLDataObjectType;
006import gudusoft.gsqlparser.dlineage.dataflow.metadata.MetadataAnalyzer;
007import gudusoft.gsqlparser.dlineage.dataflow.metadata.MetadataReader;
008import gudusoft.gsqlparser.dlineage.dataflow.model.*;
009import gudusoft.gsqlparser.dlineage.dataflow.model.xml.*;
010import gudusoft.gsqlparser.dlineage.util.Pair3;
011import gudusoft.gsqlparser.sqlenv.IdentifierInputForm;
012import gudusoft.gsqlparser.sqlenv.TSQLEnv;
013import gudusoft.gsqlparser.sqlenv.parser.sqlflow.SqlflowNameForm;
014import gudusoft.gsqlparser.util.SQLUtil;
015import gudusoft.gsqlparser.util.json.JSON;
016
017import java.io.File;
018import java.io.IOException;
019import java.nio.charset.StandardCharsets;
020import java.nio.file.Files;
021import java.nio.file.Path;
022import java.nio.file.Paths;
023import java.util.*;
024
025/**
026 * Extracts metadata information (tables, views, procedures, relationships) from a
027 * sqlflow-sharded manifest and its catalog shards.
028 *
029 * <p>This class does <b>NOT</b> analyze SQL source code in queries, nor does it produce
030 * lineage for query objects. The lineage analysis of SQL queries is performed in
031 * {@link gudusoft.gsqlparser.dlineage.DataFlowAnalyzer DataFlowAnalyzer}, which reads
032 * the source JSONL files through the {@code databaseMap} / {@code appendSqlInfo}
033 * mechanism and invokes the SQL parser to generate dataflow relationships.
034 */
035public class SqlflowShardedMetadataAnalyzer implements MetadataAnalyzer<String> {
036
037    private Map<Object, procedure> procedureMap = new LinkedHashMap<Object, procedure>();
038    private Map<Object, oraclePackage> oraclePackageMap = new LinkedHashMap<Object, oraclePackage>();
039    private Map<Object, table> tableMap = new LinkedHashMap<Object, table>();
040    private Map<Object, column> columnMap = new LinkedHashMap<Object, column>();
041    private EDbVendor vendor;
042    private TSQLEnv sqlenv;
043    private String baseDir;
044
045    public SqlflowShardedMetadataAnalyzer(String baseDir) {
046        if (baseDir == null || baseDir.trim().isEmpty()) {
047            throw new IllegalArgumentException("baseDir is required");
048        }
049        this.baseDir = baseDir;
050    }
051
052    public SqlflowShardedMetadataAnalyzer(TSQLEnv sqlenv, String baseDir) {
053        if (baseDir == null || baseDir.trim().isEmpty()) {
054            throw new IllegalArgumentException("baseDir is required");
055        }
056        this.sqlenv = sqlenv;
057        this.baseDir = baseDir;
058    }
059
060    @Override
061    public synchronized dataflow analyzeMetadata(EDbVendor metadataVendor, String metadata) {
062        init(metadataVendor);
063        dataflow dataflow = new dataflow();
064
065        String manifestContent;
066        try {
067            if (metadata.trim().startsWith("{") && metadata.trim().endsWith("}")) {
068                manifestContent = metadata.trim();
069            } else {
070                manifestContent = new String(Files.readAllBytes(Paths.get(metadata)), StandardCharsets.UTF_8);
071            }
072        } catch (IOException e) {
073            throw new RuntimeException("Failed to read manifest: " + metadata, e);
074        }
075
076        Map manifest = (Map) JSON.parseObject(manifestContent);
077        if (manifest == null) {
078            return dataflow;
079        }
080
081        if (!"sqlflow-sharded".equals(manifest.get("format"))) {
082            return dataflow;
083        }
084
085        // Version gate: the `format` field is authoritative, so consume any
086        // `sqlflow-sharded` manifest whose `formatVersion` we understand,
087        // regardless of the `createdBy` product brand (SQLdep, sqlflow, grabit
088        // all emit this format). Anything outside the supported range — a newer
089        // major version, a bogus 0/negative, or a non-numeric value — is refused
090        // with a visible error rather than mis-parsed. ONE authoritative check
091        // (MetadataReader) is used here and at every other consumer entry point.
092        if (!MetadataReader.isSupportedSqlflowSharded(manifestContent)) {
093            dataflow.getErrors().add(
094                    unsupportedVersionError(MetadataReader.shardedFormatVersion(manifestContent)));
095            return dataflow;
096        }
097
098        // Declared once at the manifest root, for the manifest and every shard it points to.
099        IdentifierInputForm nameForm = SqlflowNameForm.toLexicalInPlace(metadataVendor, manifest);
100
101        {
102            {
103                List servers = (List) manifest.get("servers");
104                if (servers == null) {
105                    return dataflow;
106                }
107
108                for (int x = 0; x < servers.size(); x++) {
109                    Map server = (Map) servers.get(x);
110                    String serverName = (String) server.get("name");
111                    String dbVendor = (String) server.get("dbVendor");
112
113                    EDbVendor vendor = metadataVendor;
114                    if (dbVendor != null) {
115                        vendor = EDbVendor.valueOf(dbVendor);
116                    }
117
118                    boolean supportsCatalogs = TSQLEnv.supportCatalog(vendor);
119                    boolean supportsSchemas = TSQLEnv.supportSchema(vendor);
120
121                    if (server.containsKey("supportsCatalogs")) {
122                        supportsCatalogs = (Boolean) server.get("supportsCatalogs");
123                    }
124                    if (server.containsKey("supportsSchemas")) {
125                        supportsSchemas = (Boolean) server.get("supportsSchemas");
126                    }
127
128                    List databases = (List) server.get("databases");
129                    List schemaShards = (List) server.get("schemas");
130
131                    // Route on the PHYSICAL shard container. Schema-topology
132                    // exports (e.g. Oracle) list shards under servers[].schemas[],
133                    // each pointing at a FLAT single-schema catalog file
134                    // ({name, tables, views, procedures}) rather than a catalog
135                    // file that nests a "schemas" array. Prefer schema shards
136                    // when present; treat an empty databases[] as absent so a
137                    // manifest with "databases":[] alongside schemas[] still
138                    // loads. Handling only `databases` left the dataflow with no
139                    // catalog table/view nodes for schema-topology exports.
140                    boolean hasSchemaShards = schemaShards != null && !schemaShards.isEmpty();
141                    boolean hasDatabaseShards = databases != null && !databases.isEmpty();
142
143                    if (hasSchemaShards) {
144                        for (int i = 0; i < schemaShards.size(); i++) {
145                            Map jsonSchemaShard = (Map) schemaShards.get(i);
146                            String schemaName = (String) jsonSchemaShard.get("name");
147                            if (schemaName != null && SQLUtil.parseNames(schemaName).size() > 1) {
148                                schemaName = "\"" + schemaName + "\"";
149                            }
150
151                            Map catalogInfo = (Map) jsonSchemaShard.get("catalog");
152                            if (catalogInfo == null) {
153                                continue;
154                            }
155                            String catalogPath = (String) catalogInfo.get("path");
156                            if (SQLUtil.isEmpty(catalogPath)) {
157                                continue;
158                            }
159
160                            String catalogContent = loadCatalogFile(catalogPath);
161                            if (catalogContent == null) {
162                                continue;
163                            }
164
165                            Map catalog = (Map) JSON.parseObject(catalogContent);
166                            if (catalog == null) {
167                                continue;
168                            }
169                            SqlflowNameForm.shardToLexicalInPlace(vendor, catalog, null, nameForm);
170
171                            // Flat catalog IS the schema: read tables/views/
172                            // procedures directly off it, keyed by the shard's
173                            // schema name under the synthetic default catalog.
174                            processTablesAndViews(vendor, supportsCatalogs, supportsSchemas, dataflow,
175                                    serverName, TSQLEnv.DEFAULT_DB_NAME, schemaName, catalog);
176                            processProcedures(vendor, supportsCatalogs, supportsSchemas, dataflow,
177                                    serverName, TSQLEnv.DEFAULT_DB_NAME, schemaName, catalog);
178                        }
179                        continue;
180                    }
181
182                    if (!hasDatabaseShards) {
183                        continue;
184                    }
185
186                    for (int i = 0; i < databases.size(); i++) {
187                        Map jsonDatabase = (Map) databases.get(i);
188                        String databaseName = (String) jsonDatabase.get("name");
189                        if (SQLUtil.parseNames(databaseName).size() > 1) {
190                            databaseName = "\"" + databaseName + "\"";
191                        }
192
193                        Map catalogInfo = (Map) jsonDatabase.get("catalog");
194                        if (catalogInfo == null) {
195                            continue;
196                        }
197                        String catalogPath = (String) catalogInfo.get("path");
198
199                        String catalogContent = loadCatalogFile(catalogPath);
200                        if (catalogContent == null) {
201                            continue;
202                        }
203
204                        Map catalog = (Map) JSON.parseObject(catalogContent);
205                        if (catalog == null) {
206                            continue;
207                        }
208                        SqlflowNameForm.shardToLexicalInPlace(vendor, catalog, null, nameForm);
209
210                        if (supportsCatalogs && supportsSchemas) {
211                            processCatalogWithSchemas(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, catalog);
212                        } else if (supportsCatalogs) {
213                            processCatalogCatalogOnly(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, catalog);
214                        } else if (supportsSchemas) {
215                            processCatalogSchemaOnly(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, catalog);
216                        }
217                    }
218                }
219            }
220        }
221
222        sortTableColumns(dataflow);
223        return dataflow;
224    }
225
226    /** A visible, machine-readable marker that a manifest was refused on version. */
227    private static error unsupportedVersionError(int formatVersion) {
228        error e = new error();
229        e.setErrorType(ErrorInfo.METADATA_ERROR);
230        e.setErrorMessage("Unsupported sqlflow-sharded formatVersion " + formatVersion
231                + "; this build supports up to " + MetadataReader.SUPPORTED_SHARDED_FORMAT_VERSION
232                + ". Metadata not loaded.");
233        return e;
234    }
235
236    private String loadCatalogFile(String catalogPath) {
237        try {
238            if (catalogPath.startsWith("/") || catalogPath.matches("^[A-Za-z]:.*")) {
239                return new String(Files.readAllBytes(Paths.get(catalogPath)), StandardCharsets.UTF_8);
240            }
241            String normalizedPath = catalogPath.replace("/", File.separator);
242            Path fullPath = Paths.get(baseDir, normalizedPath);
243            return new String(Files.readAllBytes(fullPath), StandardCharsets.UTF_8);
244        } catch (IOException e) {
245            return null;
246        }
247    }
248
249    private void processCatalogWithSchemas(EDbVendor vendor, boolean supportsCatalogs, boolean supportsSchemas,
250                                           dataflow dataflow, String serverName, String databaseName, Map catalog) {
251        List schemas = (List) catalog.get("schemas");
252        if (schemas == null) {
253            return;
254        }
255
256        for (int j = 0; j < schemas.size(); j++) {
257            Map jsonSchema = (Map) schemas.get(j);
258            String schemaName = (String) jsonSchema.get("name");
259            if (SQLUtil.parseNames(schemaName).size() > 1) {
260                schemaName = "\"" + schemaName + "\"";
261            }
262
263            processTablesAndViews(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, schemaName, jsonSchema);
264            processProcedures(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, schemaName, jsonSchema);
265        }
266    }
267
268    private void processCatalogCatalogOnly(EDbVendor vendor, boolean supportsCatalogs, boolean supportsSchemas,
269                                           dataflow dataflow, String serverName, String databaseName, Map catalog) {
270        processTablesAndViews(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName,
271                TSQLEnv.DEFAULT_SCHEMA_NAME, catalog);
272        processProcedures(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName,
273                TSQLEnv.DEFAULT_SCHEMA_NAME, catalog);
274    }
275
276    private void processCatalogSchemaOnly(EDbVendor vendor, boolean supportsCatalogs, boolean supportsSchemas,
277                                          dataflow dataflow, String serverName, Map catalog) {
278        String databaseName = TSQLEnv.DEFAULT_DB_NAME;
279        if (SQLUtil.parseNames(databaseName).size() > 1) {
280            databaseName = "\"" + databaseName + "\"";
281        }
282
283        List schemas = (List) catalog.get("schemas");
284        if (schemas == null) {
285            return;
286        }
287
288        for (int j = 0; j < schemas.size(); j++) {
289            Map jsonSchema = (Map) schemas.get(j);
290            String schemaName = (String) jsonSchema.get("name");
291            if (SQLUtil.parseNames(schemaName).size() > 1) {
292                schemaName = "\"" + schemaName + "\"";
293            }
294
295            processTablesAndViews(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, schemaName, jsonSchema);
296            processProcedures(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, schemaName, jsonSchema);
297        }
298    }
299
300    private void processTablesAndViews(EDbVendor vendor, boolean supportsCatalogs, boolean supportsSchemas,
301                                      dataflow dataflow, String serverName, String databaseName, String schemaName,
302                                      Map jsonSchema) {
303        List tables = (List) jsonSchema.get("tables");
304        List views = (List) jsonSchema.get("views");
305
306        List dbObjs = new ArrayList();
307        if (tables != null) {
308            dbObjs.addAll(tables);
309        }
310        if (views != null) {
311            dbObjs.addAll(views);
312        }
313
314        for (int k = 0; k < dbObjs.size(); k++) {
315            Map jsonTable = (Map) dbObjs.get(k);
316            String tableName = (String) jsonTable.get("name");
317            String type = (String) jsonTable.get("type");
318            String fromDDL = (String) jsonTable.get("fromDDL");
319            boolean isView = false;
320            if (type != null && type.toLowerCase(Locale.ROOT).indexOf("view") != -1) {
321                isView = true;
322            }
323
324            List columns = (List) jsonTable.get("columns");
325            if (columns == null) {
326                continue;
327            }
328
329            for (int l = 0; l < columns.size(); l++) {
330                Map jsonColumn = (Map) columns.get(l);
331                String columnName = (String) jsonColumn.get("name");
332                String dataType = null;
333                if (jsonColumn.containsKey("dataType")) {
334                    dataType = (String) jsonColumn.get("dataType");
335                }
336                Boolean primaryKey = null;
337                if (jsonColumn.containsKey("primaryKey")) {
338                    primaryKey = (Boolean) jsonColumn.get("primaryKey");
339                }
340                Boolean unqiueKey = null;
341                if (jsonColumn.containsKey("unqiueKey")) {
342                    unqiueKey = (Boolean) jsonColumn.get("unqiueKey");
343                }
344                Boolean indexKey = null;
345                if (jsonColumn.containsKey("indexKey")) {
346                    indexKey = (Boolean) jsonColumn.get("indexKey");
347                }
348                Boolean foreignKey = null;
349                if (jsonColumn.containsKey("foreignKey")) {
350                    foreignKey = (Boolean) jsonColumn.get("foreignKey");
351                }
352
353                appendTable(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName,
354                        schemaName, tableName, isView, false, columnName, dataType, primaryKey, unqiueKey, indexKey, foreignKey, fromDDL);
355            }
356        }
357
358        List synonyms = (List) jsonSchema.get("synonyms");
359        if (synonyms != null) {
360            for (int k = 0; k < synonyms.size(); k++) {
361                Map jsonSynonym = (Map) synonyms.get(k);
362                String synonymName = (String) jsonSynonym.get("name");
363                String synonymSourceName = (String) jsonSynonym.get("sourceName");
364                String synonymSourceDb = (String) jsonSynonym.get("sourceDatabase");
365                String synonymSourceSchema = (String) jsonSynonym.get("sourceSchema");
366
367                if (SQLUtil.isEmpty(synonymSourceDb)) {
368                    synonymSourceDb = databaseName;
369                }
370                if (SQLUtil.isEmpty(synonymSourceSchema)) {
371                    synonymSourceSchema = schemaName;
372                }
373
374                Object tableKey = getFullTableName(vendor, serverName, synonymSourceDb, synonymSourceSchema, synonymSourceName);
375                table sourceTable = tableMap.get(tableKey);
376                if (sourceTable == null) {
377                    // the synonym of a synonym
378                    sourceTable = tableMap.get(getFullName(vendor, ESQLDataObjectType.dotSynonyms, serverName, synonymSourceDb, synonymSourceSchema, synonymSourceName));
379                }
380                if (sourceTable != null) {
381                    List<column> columns = sourceTable.getColumns();
382                    String fromDDL = sourceTable.getFromDDL();
383                    for (int l = 0; l < columns.size(); l++) {
384                        column jsonColumn = columns.get(l);
385                        String columnName = jsonColumn.getName();
386                        String dataType = jsonColumn.getDataType();
387                        appendTable(vendor, supportsCatalogs, supportsSchemas, dataflow,
388                                serverName, databaseName, schemaName, synonymName, false, true,
389                                columnName, dataType, false, false, false, false, fromDDL);
390                    }
391
392                    Object synonymTableKey = getFullName(vendor, ESQLDataObjectType.dotSynonyms, serverName, databaseName, schemaName, synonymName);
393                    table synonymTable = tableMap.get(synonymTableKey);
394                    if (synonymTable != null) {
395                        List<column> synonymColumns = synonymTable.getColumns();
396
397                        for (int l = 0; l < synonymColumns.size(); l++) {
398                            relationship relationElement = new relationship();
399                            relationElement.setType(RelationshipType.fdd.name());
400                            relationElement.setEffectType(EffectType.synonym.name());
401
402                            long id = ++ModelBindingManager.get().RELATION_ID;
403                            relationElement.setId(String.valueOf(id));
404
405                            column targetColumn = synonymColumns.get(l);
406                            targetColumn target = new targetColumn();
407                            target.setId(String.valueOf(targetColumn.getId()));
408                            target.setColumn(targetColumn.getName());
409                            target.setParent_id(String.valueOf(synonymTable.getId()));
410                            target.setParent_name(synonymTable.getName());
411                            relationElement.setTarget(target);
412
413                            column sourceColumn = columns.get(l);
414                            sourceColumn source = new sourceColumn();
415                            source.setId(String.valueOf(sourceColumn.getId()));
416                            source.setColumn(sourceColumn.getName());
417                            source.setParent_id(String.valueOf(sourceTable.getId()));
418                            source.setParent_name(sourceTable.getName());
419                            relationElement.addSource(source);
420
421                            dataflow.getRelationships().add(relationElement);
422                        }
423                    }
424                }
425            }
426        }
427    }
428
429    private void processProcedures(EDbVendor vendor, boolean supportsCatalogs, boolean supportsSchemas,
430                                   dataflow dataflow, String serverName, String databaseName, String schemaName,
431                                   Map jsonSchema) {
432        List<Map> procedures = (List<Map>) jsonSchema.get("procedures");
433        if (procedures != null) {
434            for (Map procedure : procedures) {
435                appendProcedure(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, schemaName, procedure, null, ESQLDataObjectType.dotProcedure);
436            }
437        }
438
439        List<Map> functions = (List<Map>) jsonSchema.get("functions");
440        if (functions != null) {
441            for (Map function : functions) {
442                appendProcedure(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, schemaName, function, null, ESQLDataObjectType.dotFunction);
443            }
444        }
445
446        List<Map> triggers = (List<Map>) jsonSchema.get("triggers");
447        if (triggers != null) {
448            for (Map trigger : triggers) {
449                appendProcedure(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, schemaName, trigger, null, ESQLDataObjectType.dotTrigger);
450            }
451        }
452
453        List<Map> oraclePackages = (List<Map>) jsonSchema.get("packages");
454        if (oraclePackages != null) {
455            for (Map oraclePackageItem : oraclePackages) {
456                String oraclePackageName = (String) oraclePackageItem.get("name");
457                oraclePackage oraclePackage = appendOraclePackage(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, schemaName, oraclePackageName);
458
459                List<Map> pkgProcedures = (List<Map>) oraclePackageItem.get("procedures");
460                if (pkgProcedures != null) {
461                    for (Map procedure : pkgProcedures) {
462                        appendProcedure(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, schemaName, procedure, oraclePackage, ESQLDataObjectType.dotProcedure);
463                    }
464                }
465
466                List<Map> pkgFunctions = (List<Map>) oraclePackageItem.get("functions");
467                if (pkgFunctions != null) {
468                    for (Map function : pkgFunctions) {
469                        appendProcedure(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, schemaName, function, oraclePackage, ESQLDataObjectType.dotFunction);
470                    }
471                }
472
473                List<Map> pkgTriggers = (List<Map>) oraclePackageItem.get("triggers");
474                if (pkgTriggers != null) {
475                    for (Map trigger : pkgTriggers) {
476                        appendProcedure(vendor, supportsCatalogs, supportsSchemas, dataflow, serverName, databaseName, schemaName, trigger, oraclePackage, ESQLDataObjectType.dotTrigger);
477                    }
478                }
479            }
480        }
481    }
482
483    private void init(EDbVendor vendor) {
484        this.vendor = vendor;
485        procedureMap.clear();
486        oraclePackageMap.clear();
487        tableMap.clear();
488        columnMap.clear();
489        if (ModelBindingManager.get() == null) {
490            ModelBindingManager.set(new ModelBindingManager());
491        }
492    }
493
494    private void sortTableColumns(dataflow dataflow) {
495        if (dataflow.getTables() != null) {
496            for (table table : dataflow.getTables()) {
497                Collections.sort(table.getColumns(), new Comparator<column>() {
498                    public int compare(column t1, column t2) {
499                        if (t1.getName().equalsIgnoreCase("RelationRows"))
500                            return 1;
501                        if (t2.getName().equalsIgnoreCase("RelationRows"))
502                            return -1;
503                        return 0;
504                    }
505                });
506            }
507        }
508
509        if (dataflow.getResultsets() != null) {
510            for (table table : dataflow.getResultsets()) {
511                Collections.sort(table.getColumns(), new Comparator<column>() {
512                    public int compare(column t1, column t2) {
513                        if (t1.getName().equalsIgnoreCase("RelationRows"))
514                            return 1;
515                        if (t2.getName().equalsIgnoreCase("RelationRows"))
516                            return -1;
517                        return 0;
518                    }
519                });
520            }
521        }
522    }
523
524    private oraclePackage appendOraclePackage(EDbVendor vendor, boolean supportsCatalogs, boolean supportsSchemas,
525                                              dataflow dataflow, String serverName, String databaseName, String schemaName,
526                                              String oraclePackageName) {
527        Object oraclePackageKey = getFullName(vendor, ESQLDataObjectType.dotOraclePackage, serverName, databaseName, schemaName, oraclePackageName);
528        if (!oraclePackageMap.containsKey(oraclePackageKey)) {
529            oraclePackage oraclePackage = new oraclePackage();
530            if (serverName != null) {
531                oraclePackage.setServer(serverName);
532            } else if (sqlenv != null && sqlenv.getDefaultServerName() != null
533                    && !TSQLEnv.DEFAULT_SERVER_NAME.equalsIgnoreCase(sqlenv.getDefaultServerName())) {
534                oraclePackage.setServer(sqlenv.getDefaultServerName());
535            }
536            if (supportsCatalogs && databaseName != null) {
537                oraclePackage.setDatabase(databaseName);
538            } else if (supportsCatalogs && sqlenv != null && sqlenv.getDefaultCatalogName() != null
539                    && !TSQLEnv.DEFAULT_DB_NAME.equalsIgnoreCase(sqlenv.getDefaultCatalogName())) {
540                oraclePackage.setDatabase(sqlenv.getDefaultCatalogName());
541            }
542
543            if (supportsSchemas && schemaName != null) {
544                oraclePackage.setSchema(schemaName);
545            } else if (supportsSchemas && sqlenv != null && sqlenv.getDefaultSchemaName() != null
546                    && !TSQLEnv.DEFAULT_SCHEMA_NAME.equalsIgnoreCase(sqlenv.getDefaultSchemaName())) {
547                oraclePackage.setSchema(sqlenv.getDefaultSchemaName());
548            }
549
550            oraclePackage.setName(oraclePackageName);
551            if (supportsSchemas && !SQLUtil.isEmpty(oraclePackage.getSchema())) {
552                oraclePackage.setName(oraclePackage.getSchema() + "." + oraclePackage.getName());
553            }
554            if (supportsCatalogs && !SQLUtil.isEmpty(oraclePackage.getDatabase())) {
555                oraclePackage.setName(oraclePackage.getDatabase() + "." + oraclePackage.getName());
556            }
557            oraclePackage.setId(String.valueOf(++ModelBindingManager.get().TABLE_COLUMN_ID));
558            if (ModelBindingManager.getGlobalSqlInfo() != null) {
559                oraclePackage.setCoordinate(new Pair3<Long, Long, Integer>(-1L, -1L,
560                        ModelBindingManager.getGlobalSqlInfo().getIndexOf(ModelBindingManager.getGlobalHash())) + ","
561                        + new Pair3<Long, Long, Integer>(-1L, -1L,
562                        ModelBindingManager.getGlobalSqlInfo().getIndexOf(ModelBindingManager.getGlobalHash())));
563            }
564            dataflow.getPackages().add(oraclePackage);
565            oraclePackageMap.put(oraclePackageKey, oraclePackage);
566        }
567        return oraclePackageMap.get(oraclePackageKey);
568    }
569
570    private void appendProcedure(EDbVendor vendor, boolean supportsCatalogs, boolean supportsSchemas,
571                                 dataflow dataflow, String serverName, String databaseName, String schemaName,
572                                 Map procedureObj, oraclePackage oraclePackage, ESQLDataObjectType role) {
573        String procedureName = (String) procedureObj.get("name");
574        if (procedureName == null) {
575            return;
576        }
577
578        Object procedureKey = getFullName(vendor, role, serverName, databaseName, schemaName, procedureName);
579        if (!procedureMap.containsKey(procedureKey)) {
580            procedure procedure = new procedure();
581            if (serverName != null) {
582                procedure.setServer(serverName);
583            } else if (sqlenv != null && sqlenv.getDefaultServerName() != null
584                    && !TSQLEnv.DEFAULT_SERVER_NAME.equalsIgnoreCase(sqlenv.getDefaultServerName())) {
585                procedure.setServer(sqlenv.getDefaultServerName());
586            }
587            if (supportsCatalogs && databaseName != null) {
588                procedure.setDatabase(databaseName);
589            } else if (supportsCatalogs && sqlenv != null && sqlenv.getDefaultCatalogName() != null
590                    && !TSQLEnv.DEFAULT_DB_NAME.equalsIgnoreCase(sqlenv.getDefaultCatalogName())) {
591                procedure.setDatabase(sqlenv.getDefaultCatalogName());
592            }
593
594            if (supportsSchemas && schemaName != null) {
595                procedure.setSchema(schemaName);
596            } else if (supportsSchemas && sqlenv != null && sqlenv.getDefaultSchemaName() != null
597                    && !TSQLEnv.DEFAULT_SCHEMA_NAME.equalsIgnoreCase(sqlenv.getDefaultSchemaName())) {
598                procedure.setSchema(sqlenv.getDefaultSchemaName());
599            }
600
601            procedure.setName(procedureName);
602            if (supportsSchemas && !SQLUtil.isEmpty(procedure.getSchema())) {
603                procedure.setName(procedure.getSchema() + "." + procedure.getName());
604            }
605            if (supportsCatalogs && !SQLUtil.isEmpty(procedure.getDatabase())) {
606                procedure.setName(procedure.getDatabase() + "." + procedure.getName());
607            }
608            procedure.setId(String.valueOf(++ModelBindingManager.get().TABLE_COLUMN_ID));
609            if (ModelBindingManager.getGlobalSqlInfo() != null) {
610                procedure.setCoordinate(new Pair3<Long, Long, Integer>(-1L, -1L,
611                        ModelBindingManager.getGlobalSqlInfo().getIndexOf(ModelBindingManager.getGlobalHash())) + ","
612                        + new Pair3<Long, Long, Integer>(-1L, -1L,
613                        ModelBindingManager.getGlobalSqlInfo().getIndexOf(ModelBindingManager.getGlobalHash())));
614            }
615            procedure.setType((String) procedureObj.get("type"));
616            if (oraclePackage != null) {
617                oraclePackage.getProcedures().add(procedure);
618            } else {
619                dataflow.getProcedures().add(procedure);
620            }
621            procedureMap.put(procedureKey, procedure);
622        }
623    }
624
625    private void appendTable(EDbVendor vendor, boolean supportsCatalogs, boolean supportsSchemas,
626                            dataflow dataflow, String serverName, String databaseName, String schemaName,
627                            String tableName, boolean isView, boolean isSynonym, String columnName,
628                            String dataType, Boolean primaryKey, Boolean unqiueKey, Boolean indexKey,
629                            Boolean foreignKey, String fromDDL) {
630        // A synonym is keyed in its own role: a profile may give synonyms rules of their own.
631        Object tableKey = getFullName(vendor, isSynonym ? ESQLDataObjectType.dotSynonyms : ESQLDataObjectType.dotTable,
632                serverName, databaseName, schemaName, tableName);
633        if (!tableMap.containsKey(tableKey)) {
634            table table = new table();
635            // This reader publishes SQL spellings (its input names, converted by their declared form).
636            table.setPublishedNameSqlSpelling(true);
637            if (serverName != null) {
638                table.setServer(serverName);
639            } else if (sqlenv != null && sqlenv.getDefaultServerName() != null
640                    && !TSQLEnv.DEFAULT_SERVER_NAME.equalsIgnoreCase(sqlenv.getDefaultServerName())) {
641                table.setServer(sqlenv.getDefaultServerName());
642            }
643            if (supportsCatalogs && databaseName != null) {
644                table.setDatabase(databaseName);
645            } else if (supportsCatalogs && sqlenv != null && sqlenv.getDefaultCatalogName() != null
646                    && !TSQLEnv.DEFAULT_DB_NAME.equalsIgnoreCase(sqlenv.getDefaultCatalogName())) {
647                table.setDatabase(sqlenv.getDefaultCatalogName());
648            }
649
650            if (supportsSchemas && schemaName != null) {
651                table.setSchema(schemaName);
652            } else if (supportsSchemas && sqlenv != null && sqlenv.getDefaultSchemaName() != null
653                    && !TSQLEnv.DEFAULT_SCHEMA_NAME.equalsIgnoreCase(sqlenv.getDefaultSchemaName())) {
654                table.setSchema(sqlenv.getDefaultSchemaName());
655            }
656
657            table.setName(tableName);
658            if (supportsSchemas && !SQLUtil.isEmpty(table.getSchema())) {
659                table.setName(table.getSchema() + "." + table.getName());
660            }
661            if (supportsCatalogs && !SQLUtil.isEmpty(table.getDatabase())) {
662                table.setName(table.getDatabase() + "." + table.getName());
663            }
664            table.setId(String.valueOf(++ModelBindingManager.get().TABLE_COLUMN_ID));
665            if (ModelBindingManager.getGlobalSqlInfo() != null) {
666                table.setCoordinate(new Pair3<Long, Long, Integer>(-1L, -1L,
667                        ModelBindingManager.getGlobalSqlInfo().getIndexOf(ModelBindingManager.getGlobalHash())) + ","
668                        + new Pair3<Long, Long, Integer>(-1L, -1L,
669                        ModelBindingManager.getGlobalSqlInfo().getIndexOf(ModelBindingManager.getGlobalHash())));
670            }
671
672            if (!SQLUtil.isEmpty(fromDDL)) {
673                table.setFromDDL(fromDDL);
674            }
675
676            if (isView) {
677                table.setType("view");
678                dataflow.getViews().add(table);
679            } else {
680                table.setType("table");
681                dataflow.getTables().add(table);
682            }
683            if (isSynonym) {
684                table.setSubType(SubType.synonym.name());
685            }
686
687            tableMap.put(tableKey, table);
688        }
689
690        table table = tableMap.get(tableKey);
691        // A column belongs to its table or synonym under that object's own key, and is named under column rules.
692        Object sourceColumnKey = java.util.Arrays.asList(tableKey,
693                DlineageUtil.lexicalNameKey(vendor, columnName, ESQLDataObjectType.dotColumn));
694        if (!columnMap.containsKey(sourceColumnKey)) {
695            column column = new column();
696            column.setId(String.valueOf(++ModelBindingManager.get().TABLE_COLUMN_ID));
697            column.setName(columnName);
698            if ("*".equals(columnName)) {
699                // a metadata column named *, not a star
700                column.setStar(Boolean.FALSE);
701            }
702            // Input names have been converted to SQL lexical form at the reader boundary.
703            column.setIdentifierKey(DlineageUtil.lexicalColumnKey(vendor, columnName));
704            column.setDataType(dataType);
705            column.setPrimaryKey(primaryKey);
706            column.setForeignKey(foreignKey);
707            column.setIndexKey(indexKey);
708            column.setUnqiueKey(unqiueKey);
709            if ("RelationRows".equalsIgnoreCase(columnName)) {
710                column.setSource("system");
711            }
712            if (ModelBindingManager.getGlobalSqlInfo() != null) {
713                column.setCoordinate(new Pair3<Long, Long, Integer>(-1L, -1L,
714                        ModelBindingManager.getGlobalSqlInfo().getIndexOf(ModelBindingManager.getGlobalHash())) + ","
715                        + new Pair3<Long, Long, Integer>(-1L, -1L,
716                        ModelBindingManager.getGlobalSqlInfo().getIndexOf(ModelBindingManager.getGlobalHash())));
717            }
718            table.getColumns().add(column);
719            columnMap.put(sourceColumnKey, column);
720        }
721    }
722
723    private Object getFullTableName(EDbVendor vendor, String... segments) {
724        return getFullName(vendor, ESQLDataObjectType.dotTable, segments);
725    }
726
727    /**
728     * The reader's key of an object: its server as given, and its name joined from the other segments under the
729     * analysis's identifier rules for {@code objectType} (the names are SQL spellings at this boundary).
730     */
731    private Object getFullName(EDbVendor vendor, ESQLDataObjectType objectType, String... segments) {
732        StringBuilder builder = new StringBuilder();
733        for (int i = 1; i < segments.length; i++) {
734                        if (segments[i] == null) {
735                                continue;
736                        }
737            builder.append(segments[i]);
738            if (i < segments.length - 1) {
739                builder.append(".");
740            }
741        }
742
743        return java.util.Arrays.asList(segments[0], DlineageUtil.lexicalNameKey(vendor, builder.toString(), objectType));
744    }
745
746    public static void main(String[] args) throws Exception {
747        String sampleManifestPath = "sqlflow-sharded/sample/manifest.json";
748        String baseDir = new File(sampleManifestPath).getParent();
749        dataflow dataflow = new SqlflowShardedMetadataAnalyzer(baseDir)
750                .analyzeMetadata(EDbVendor.dbvmssql, SQLUtil.getFileContent(new File(sampleManifestPath)));
751        System.out.println("Tables: " + dataflow.getTables().size());
752        System.out.println("Views: " + dataflow.getViews().size());
753        System.out.println("Procedures: " + dataflow.getProcedures().size());
754    }
755}