001package gudusoft.gsqlparser.sqlenv.parser.sharded;
002
003import gudusoft.gsqlparser.EDbVendor;
004import gudusoft.gsqlparser.sqlenv.TSQLEnv;
005import gudusoft.gsqlparser.sqlenv.TSQLCatalog;
006import gudusoft.gsqlparser.sqlenv.TSQLSchema;
007import gudusoft.gsqlparser.sqlenv.TSQLTable;
008import gudusoft.gsqlparser.util.SQLUtil;
009import gudusoft.gsqlparser.util.json.JSON;
010
011import java.io.File;
012import java.io.IOException;
013import java.nio.charset.StandardCharsets;
014import java.nio.file.Files;
015import java.nio.file.Paths;
016import java.util.ArrayList;
017import java.util.List;
018import java.util.Map;
019import java.util.logging.Level;
020import java.util.logging.Logger;
021
022/**
023 * SQLEnv implementation for sqlflow-sharded format.
024 * <p>
025 * Parses catalog.json files from the sharded export format, loading metadata
026 * including tables, views, columns, and synonyms. Supports different database
027 * topologies based on the manifest configuration.
028 * <p>
029 * The sharded format structure:
030 * <pre>
031 * manifest.json
032 * catalog/&lt;shardId&gt;.catalog.json
033 * source/&lt;shardId&gt;.source.jsonl
034 * index/&lt;shardId&gt;.source.idx
035 * </pre>
036 */
037@SuppressWarnings("rawtypes")
038public class SqlflowShardedSQLEnv extends TSQLEnv {
039
040    private Map server;
041    private String baseDir;
042    private Boolean init = false;
043
044    /**
045     * Constructor with explicit base directory for catalog files.
046     *
047     * @param defaultServer   default server name
048     * @param defaultDatabase default database name
049     * @param defaultSchema   default schema name
050     * @param dbVendor        database vendor
051     * @param server          server configuration from manifest
052     * @param baseDir         base directory containing catalog/, source/, index/ subdirectories
053     */
054    public SqlflowShardedSQLEnv(String defaultServer, String defaultDatabase, String defaultSchema,
055                                 EDbVendor dbVendor, Map server, String baseDir) {
056        super(dbVendor);
057        initDefaults(defaultServer, defaultDatabase, defaultSchema);
058        this.server = server;
059        this.baseDir = baseDir;
060        initSQLEnv();
061    }
062
063    private void initDefaults(String defaultServer, String defaultDatabase, String defaultSchema) {
064        if (!SQLUtil.isEmpty(defaultServer) && !defaultServer.equals(TSQLEnv.DEFAULT_SERVER_NAME)) {
065            setDefaultServerName(defaultServer);
066        }
067        if (!SQLUtil.isEmpty(defaultDatabase) && !defaultDatabase.equals(TSQLEnv.DEFAULT_DB_NAME)) {
068            setDefaultCatalogName(defaultDatabase);
069        }
070        if (!SQLUtil.isEmpty(defaultSchema) && !defaultSchema.equals(TSQLEnv.DEFAULT_SCHEMA_NAME)) {
071            setDefaultSchemaName(defaultSchema);
072        }
073    }
074
075    @Override
076    public void initSQLEnv() {
077        synchronized (init) {
078            if (init)
079                return;
080
081            if (server == null) {
082                init = true;
083                return;
084            }
085
086            try {
087                // Parse dbVendor from server config
088                String dbVendor = (String) server.get("dbVendor");
089                EDbVendor vendor = getDBVendor();
090                if (dbVendor != null) {
091                    vendor = EDbVendor.valueOf(dbVendor);
092                }
093
094                // Set server name from config
095                if (!SQLUtil.isEmpty((String) server.get("name"))) {
096                    this.setDefaultServerName((String) server.get("name"));
097                }
098
099                boolean supportsCatalogs = TSQLEnv.supportCatalog(vendor);
100                boolean supportsSchemas = TSQLEnv.supportSchema(vendor);
101
102                // The manifest is authoritative about how THIS export was sharded,
103                // which may differ from the vendor's GSP defaults. Oracle defaults
104                // to catalog+schema, but its natural export is schema-only shards
105                // (servers[].schemas[]). Honor the declared flags so the correct
106                // process* branch is chosen; otherwise a schema-topology manifest
107                // routes to processDatabases(), finds no "databases", and loads
108                // nothing (silent empty env -> empty lineage). Mirror the same
109                // override already done in SqlflowShardedMetadataAnalyzer.
110                if (server.containsKey("supportsCatalogs")) {
111                    supportsCatalogs = Boolean.TRUE.equals(server.get("supportsCatalogs"));
112                }
113                if (server.containsKey("supportsSchemas")) {
114                    supportsSchemas = Boolean.TRUE.equals(server.get("supportsSchemas"));
115                }
116
117                // Get nameLevels for topology determination
118                List<String> nameLevels = getNameLevels(server);
119
120                // Resolve base directory
121                String resolvedBaseDir = resolveBaseDir();
122
123                // List for pending synonyms (resolved after all tables are loaded)
124                List<Object[]> pendingSynonyms = new ArrayList<>();
125
126                // Route on the PHYSICAL shard container first, not the vendor
127                // default. Schema-topology exports (Oracle) list shards under
128                // server.schemas[]; catalog-topology exports (SQL Server, MySQL)
129                // list them under server.databases[]. Only after the container
130                // is known do the catalog/schema flags decide how a databases[]
131                // shard's catalog file is interpreted. Deciding purely from the
132                // vendor default sent Oracle schema shards to processDatabases()
133                // (which reads databases[]) and loaded nothing; deciding purely
134                // from the flags would send a databases[]-wrapped schema-only
135                // manifest to processSchemas() (which reads schemas[]) and load
136                // nothing. Treat an empty list as absent.
137                List schemaShards = (List) server.get("schemas");
138                List databaseShards = (List) server.get("databases");
139                boolean hasSchemaShards = schemaShards != null && !schemaShards.isEmpty();
140                boolean hasDatabaseShards = databaseShards != null && !databaseShards.isEmpty();
141
142                if (hasSchemaShards) {
143                    // Schema-topology shards (e.g., Oracle)
144                    processSchemas(server, resolvedBaseDir, pendingSynonyms);
145                } else if (hasDatabaseShards) {
146                    if (supportsCatalogs && supportsSchemas) {
147                        // Catalog + Schema topology (e.g., SQL Server, PostgreSQL)
148                        processDatabases(server, resolvedBaseDir, pendingSynonyms);
149                    } else if (supportsCatalogs) {
150                        // Catalog-only topology (e.g., MySQL)
151                        processDatabasesAsCatalog(server, resolvedBaseDir, pendingSynonyms);
152                    } else {
153                        // Schema-only data wrapped in databases[]: the catalog
154                        // file nests a schemas array; processDatabases already
155                        // reads catalogJson.schemas (mirrors the metadata
156                        // analyzer's processCatalogSchemaOnly).
157                        processDatabases(server, resolvedBaseDir, pendingSynonyms);
158                    }
159                }
160
161                // Resolve pending synonyms
162                for (Object[] pending : pendingSynonyms) {
163                    CatalogMetadataHelper.appendSynonym((TSQLSchema) pending[0], (Map) pending[1]);
164                }
165
166            } catch (Exception e) {
167                Logger.getLogger(SqlflowShardedSQLEnv.class.getName())
168                        .log(Level.WARNING, "Parse sharded catalog failed", e);
169            }
170
171            init = true;
172        }
173    }
174
175    /**
176     * Get nameLevels from server config, defaulting based on vendor capabilities.
177     */
178    private List<String> getNameLevels(Map server) {
179        List<String> nameLevels = (List<String>) server.get("nameLevels");
180        if (nameLevels != null && !nameLevels.isEmpty()) {
181            return nameLevels;
182        }
183
184        // Default based on supportsCatalogs/supportsSchemas
185        Boolean supportsCatalogs = (Boolean) server.get("supportsCatalogs");
186        Boolean supportsSchemas = (Boolean) server.get("supportsSchemas");
187
188        List<String> defaultLevels = new ArrayList<>();
189        if (supportsCatalogs != null && supportsCatalogs) {
190            defaultLevels.add("catalog");
191        }
192        if (supportsSchemas != null && supportsSchemas) {
193            defaultLevels.add("schema");
194        }
195
196        return defaultLevels;
197    }
198
199    /**
200     * Get base directory.
201     */
202    private String resolveBaseDir() {
203        return baseDir;
204    }
205
206    /**
207     * Process databases in catalog+schema topology.
208     */
209    private void processDatabases(Map server, String baseDir, List<Object[]> pendingSynonyms) {
210        List databases = (List) server.get("databases");
211        if (databases == null) {
212            return;
213        }
214
215        for (int i = 0; i < databases.size(); i++) {
216            Map jsonDatabase = (Map) databases.get(i);
217            String databaseName = (String) jsonDatabase.get("name");
218
219            // Get catalog path
220            Map catalogInfo = (Map) jsonDatabase.get("catalog");
221            if (catalogInfo == null) {
222                continue;
223            }
224
225            String catalogPath = (String) catalogInfo.get("path");
226            if (SQLUtil.isEmpty(catalogPath)) {
227                continue;
228            }
229
230            // Resolve full path to catalog file
231            String fullCatalogPath = resolvePath(baseDir, catalogPath);
232            Map catalogJson = loadCatalogJson(fullCatalogPath);
233            if (catalogJson == null) {
234                continue;
235            }
236
237            // Process schemas within catalog
238            List schemas = (List) catalogJson.get("schemas");
239            if (schemas == null) {
240                // Fallback: database might be directly in catalog root (nameLevels = ["database"])
241                // In this case, create a default schema to hold tables/views
242                TSQLCatalog sqlCatalog = getSQLCatalog(databaseName, true);
243                TSQLSchema sqlSchema = sqlCatalog.getSchema(TSQLEnv.DEFAULT_SCHEMA_NAME, true);
244                appendTables(catalogJson, sqlSchema);
245                appendViews(catalogJson, sqlSchema);
246                CatalogMetadataHelper.appendPackages(catalogJson, sqlSchema);
247                CatalogMetadataHelper.appendProcedures(catalogJson, sqlSchema, null);
248                collectSynonyms(catalogJson, sqlSchema, pendingSynonyms);
249            } else {
250                TSQLCatalog sqlCatalog = getSQLCatalog(databaseName, true);
251                for (int j = 0; j < schemas.size(); j++) {
252                    Map jsonSchema = (Map) schemas.get(j);
253                    String schemaName = (String) jsonSchema.get("name");
254                    TSQLSchema sqlSchema = sqlCatalog.getSchema(schemaName, true);
255                    appendTables(jsonSchema, sqlSchema);
256                    appendViews(jsonSchema, sqlSchema);
257                    CatalogMetadataHelper.appendPackages(jsonSchema, sqlSchema);
258                    CatalogMetadataHelper.appendProcedures(jsonSchema, sqlSchema, null);
259                    collectSynonyms(jsonSchema, sqlSchema, pendingSynonyms);
260                }
261            }
262        }
263    }
264
265    /**
266     * Process databases in catalog-only topology (no schema level).
267     */
268    private void processDatabasesAsCatalog(Map server, String baseDir, List<Object[]> pendingSynonyms) {
269        List databases = (List) server.get("databases");
270        if (databases == null) {
271            return;
272        }
273
274        for (int i = 0; i < databases.size(); i++) {
275            Map jsonDatabase = (Map) databases.get(i);
276            String databaseName = (String) jsonDatabase.get("name");
277
278            Map catalogInfo = (Map) jsonDatabase.get("catalog");
279            if (catalogInfo == null) {
280                continue;
281            }
282
283            String catalogPath = (String) catalogInfo.get("path");
284            if (SQLUtil.isEmpty(catalogPath)) {
285                continue;
286            }
287
288            String fullCatalogPath = resolvePath(baseDir, catalogPath);
289            Map catalogJson = loadCatalogJson(fullCatalogPath);
290            if (catalogJson == null) {
291                continue;
292            }
293
294            // In catalog-only mode, each database is treated as a catalog
295            // Tables/views are at the catalog level (not in schemas)
296            TSQLCatalog sqlCatalog = getSQLCatalog(databaseName, true);
297            TSQLSchema sqlSchema = sqlCatalog.getSchema(TSQLEnv.DEFAULT_SCHEMA_NAME, true);
298
299            // Tables/views might be at root level or in schemas
300            List tables = (List) catalogJson.get("tables");
301            List views = (List) catalogJson.get("views");
302
303            if (tables != null) {
304                for (Object table : tables) {
305                    appendTableObject((Map) table, sqlSchema, false);
306                }
307            }
308            if (views != null) {
309                for (Object view : views) {
310                    appendTableObject((Map) view, sqlSchema, true);
311                }
312            }
313            CatalogMetadataHelper.appendPackages(catalogJson, sqlSchema);
314            CatalogMetadataHelper.appendProcedures(catalogJson, sqlSchema, null);
315        }
316    }
317
318    /**
319     * Process schemas in schema-only topology (e.g., Oracle).
320     */
321    private void processSchemas(Map server, String baseDir, List<Object[]> pendingSynonyms) {
322        List schemas = (List) server.get("schemas");
323        if (schemas == null) {
324            return;
325        }
326
327        for (int i = 0; i < schemas.size(); i++) {
328            Map jsonSchema = (Map) schemas.get(i);
329            String schemaName = (String) jsonSchema.get("name");
330
331            // Get catalog path
332            Map catalogInfo = (Map) jsonSchema.get("catalog");
333            String fullCatalogPath = null;
334            Map catalogJson = null;
335
336            if (catalogInfo != null) {
337                String catalogPath = (String) catalogInfo.get("path");
338                if (!SQLUtil.isEmpty(catalogPath)) {
339                    fullCatalogPath = resolvePath(baseDir, catalogPath);
340                    catalogJson = loadCatalogJson(fullCatalogPath);
341                }
342            }
343
344            // Create synthetic catalog for schema-only vendors
345            TSQLCatalog sqlCatalog = getSQLCatalog(TSQLEnv.DEFAULT_DB_NAME, true);
346            TSQLSchema sqlSchema = sqlCatalog.getSchema(schemaName, true);
347
348            if (catalogJson != null) {
349                appendTables(catalogJson, sqlSchema);
350                appendViews(catalogJson, sqlSchema);
351                CatalogMetadataHelper.appendPackages(catalogJson, sqlSchema);
352                CatalogMetadataHelper.appendProcedures(catalogJson, sqlSchema, null);
353                collectSynonyms(catalogJson, sqlSchema, pendingSynonyms);
354            } else {
355                // Process directly from schema config
356                appendTables(jsonSchema, sqlSchema);
357                appendViews(jsonSchema, sqlSchema);
358                CatalogMetadataHelper.appendPackages(jsonSchema, sqlSchema);
359                CatalogMetadataHelper.appendProcedures(jsonSchema, sqlSchema, null);
360                collectSynonyms(jsonSchema, sqlSchema, pendingSynonyms);
361            }
362        }
363    }
364
365    /**
366     * Load and parse a catalog JSON file.
367     */
368    private Map loadCatalogJson(String path) {
369        if (path == null) {
370            return null; // unresolvable relative path with no base directory
371        }
372        try {
373            File file = new File(path);
374            if (!file.exists()) {
375                Logger.getLogger(SqlflowShardedSQLEnv.class.getName())
376                        .warning("Catalog file not found: " + path);
377                return null;
378            }
379
380            String content = new String(Files.readAllBytes(file.toPath()), StandardCharsets.UTF_8);
381            Object parsed = JSON.parseObject(content);
382            if (parsed instanceof Map) {
383                return (Map) parsed;
384            }
385        } catch (IOException e) {
386            Logger.getLogger(SqlflowShardedSQLEnv.class.getName())
387                    .log(Level.WARNING, "Failed to read catalog file: " + path, e);
388        }
389        return null;
390    }
391
392    /**
393     * Resolve relative path against base directory.
394     */
395    private String resolvePath(String baseDir, String relativePath) {
396        // An absolute path stands on its own regardless of base directory.
397        if (new File(relativePath).isAbsolute()) {
398            return relativePath;
399        }
400        if (baseDir == null) {
401            // Do NOT resolve a relative shard path against the process working
402            // directory: that could read a same-named catalog from CWD and merge
403            // a wrong environment. Without a base directory the shard is simply
404            // unreachable (no catalog is better than a wrong one).
405            return null;
406        }
407
408        // Handle both forward slash and backslash
409        String normalizedBase = baseDir.replace('\\', '/');
410        String normalizedRelative = relativePath.replace('\\', '/');
411
412        return Paths.get(normalizedBase, normalizedRelative).toString();
413    }
414
415    /**
416     * Append tables from a schema/catalog JSON object.
417     */
418    private void appendTables(Map jsonSchema, TSQLSchema sqlSchema) {
419        List tables = (List) jsonSchema.get("tables");
420        if (tables == null) {
421            return;
422        }
423
424        for (int i = 0; i < tables.size(); i++) {
425            Map jsonTable = (Map) tables.get(i);
426            appendTableObject(jsonTable, sqlSchema, false);
427        }
428    }
429
430    /**
431     * Append views from a schema/catalog JSON object.
432     */
433    private void appendViews(Map jsonSchema, TSQLSchema sqlSchema) {
434        List views = (List) jsonSchema.get("views");
435        if (views == null) {
436            return;
437        }
438
439        for (int i = 0; i < views.size(); i++) {
440            Map jsonTable = (Map) views.get(i);
441            appendTableObject(jsonTable, sqlSchema, true);
442        }
443    }
444
445    /**
446     * Append a single table or view object.
447     */
448    private void appendTableObject(Map jsonTable, TSQLSchema sqlSchema, boolean isView) {
449        String tableName = (String) jsonTable.get("name");
450        if (SQLUtil.isEmpty(tableName)) {
451            return;
452        }
453
454        TSQLTable sqlTable = sqlSchema.createTable(tableName, isView ? 3 : 0);
455        if (isView) {
456            sqlTable.setView(true);
457        }
458
459        // Set table type if present
460        String type = (String) jsonTable.get("type");
461        if (type != null && type.toLowerCase().indexOf("view") != -1) {
462            sqlTable.setView(true);
463        }
464
465        // Add columns
466        List columns = (List) jsonTable.get("columns");
467        if (columns != null) {
468            for (int j = 0; j < columns.size(); j++) {
469                Map jsonColumn = (Map) columns.get(j);
470                String columnName = (String) jsonColumn.get("name");
471                if (!SQLUtil.isEmpty(columnName)) {
472                    sqlTable.addColumn(columnName);
473                }
474            }
475        }
476    }
477
478    /**
479     * Collect synonyms for later resolution.
480     */
481    private void collectSynonyms(Map jsonSchema, TSQLSchema sqlSchema, List<Object[]> pendingSynonyms) {
482        List synonyms = (List) jsonSchema.get("synonyms");
483        if (synonyms == null) {
484            return;
485        }
486
487        for (int i = 0; i < synonyms.size(); i++) {
488            Map jsonSynonym = (Map) synonyms.get(i);
489            if (jsonSynonym == null) {
490                continue;
491            }
492            pendingSynonyms.add(new Object[]{sqlSchema, jsonSynonym});
493        }
494    }
495}