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/<shardId>.catalog.json 033 * source/<shardId>.source.jsonl 034 * index/<shardId>.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}