0
# Database Metadata Access
1

2
Comprehensive database metadata functionality for discovering available catalogs, schemas, and database capabilities with specific Flink SQL Gateway integration. Provides detailed information about the Flink database system and its supported features.
3

4
## Capabilities
5

6
### Database Information Access
7

8
FlinkDatabaseMetaData provides comprehensive information about the Flink database system, driver version, and capabilities.
9

10
```java { .api }
11
public class FlinkDatabaseMetaData extends BaseDatabaseMetaData {
12
    // Connection access
13
    public Connection getConnection() throws SQLException;
14
    
15
    // Database product information
16
    public String getDatabaseProductName() throws SQLException; // Returns "Apache Flink"
17
    public String getDatabaseProductVersion() throws SQLException;
18
    public int getDatabaseMajorVersion() throws SQLException;
19
    public int getDatabaseMinorVersion() throws SQLException;
20
    
21
    // Driver information
22
    public String getDriverName() throws SQLException;
23
    public String getDriverVersion() throws SQLException;  
24
    public int getDriverMajorVersion();
25
    public int getDriverMinorVersion();
26
    
27
    // Connection information
28
    public String getURL() throws SQLException;
29
    public String getUserName() throws SQLException; // May return null
30
}
31
```
32

33
**Usage Example:**
34
```java
35
Connection connection = DriverManager.getConnection("jdbc:flink://localhost:8083");
36
DatabaseMetaData metadata = connection.getMetaData();
37

38
System.out.println("Database: " + metadata.getDatabaseProductName());
39
System.out.println("Database Version: " + metadata.getDatabaseProductVersion());
40
System.out.println("Driver: " + metadata.getDriverName());
41
System.out.println("Driver Version: " + metadata.getDriverVersion());
42
System.out.println("JDBC URL: " + metadata.getURL());
43

44
System.out.printf("Database Version: %d.%d%n",
45
    metadata.getDatabaseMajorVersion(),
46
    metadata.getDatabaseMinorVersion());
47
```
48

49
### Catalog and Schema Discovery
50

51
Discover available catalogs and schemas in the Flink SQL Gateway environment.
52

53
```java { .api }
54
public class FlinkDatabaseMetaData extends BaseDatabaseMetaData {
55
    public ResultSet getCatalogs() throws SQLException;
56
    public ResultSet getSchemas() throws SQLException;
57
    public ResultSet getSchemas(String catalog, String schemaPattern) throws SQLException; // Not supported
58
}
59
```
60

61
**Catalog Discovery:**
62
```java
63
DatabaseMetaData metadata = connection.getMetaData();
64
ResultSet catalogs = metadata.getCatalogs();
65

66
System.out.println("Available Catalogs:");
67
while (catalogs.next()) {
68
    String catalogName = catalogs.getString("TABLE_CAT");
69
    System.out.println("  - " + catalogName);
70
}
71
catalogs.close();
72
```
73

74
**Schema Discovery:**
75
```java
76
DatabaseMetaData metadata = connection.getMetaData();
77
ResultSet schemas = metadata.getSchemas();
78

79
System.out.println("Available Schemas:");
80
while (schemas.next()) {
81
    String schemaName = schemas.getString("TABLE_SCHEM");
82
    String catalogName = schemas.getString("TABLE_CATALOG");
83
    System.out.printf("  - %s (catalog: %s)%n", schemaName, catalogName);
84
}
85
schemas.close();
86
```
87

88
### Database Capability Information
89

90
Query database capabilities and supported SQL features for application compatibility.
91

92
```java { .api }
93
public class FlinkDatabaseMetaData extends BaseDatabaseMetaData {
94
    // General capabilities
95
    public boolean isReadOnly() throws SQLException; // Returns true
96
    public boolean allTablesAreSelectable() throws SQLException; // Returns true
97
    public boolean allProceduresAreCallable() throws SQLException; // Returns false
98
    
99
    // Null handling
100
    public boolean nullsAreSortedLow() throws SQLException; // Returns true
101
    public boolean nullsAreSortedHigh() throws SQLException; // Returns false
102
    public boolean nullsAreSortedAtStart() throws SQLException; // Returns false
103
    public boolean nullsAreSortedAtEnd() throws SQLException; // Returns false
104
    
105
    // Transaction support  
106
    public boolean supportsTransactions() throws SQLException; // Returns false
107
    public boolean supportsDataDefinitionAndDataManipulationTransactions() throws SQLException; // Returns false
108
    public boolean supportsDataManipulationTransactionsOnly() throws SQLException; // Returns false
109
    public boolean dataDefinitionCausesTransactionCommit() throws SQLException; // Returns false
110
    public boolean dataDefinitionIgnoredInTransactions() throws SQLException; // Returns false
111
    
112
    // Advanced features (most return false)
113
    public boolean supportsStoredFunctionsUsingCallSyntax() throws SQLException; // Returns false
114
    public boolean autoCommitFailureClosesAllResultSets() throws SQLException; // Returns false
115
    public boolean generatedKeyAlwaysReturned() throws SQLException; // Returns false
116
}
117
```
118

119
**Usage Example:**
120
```java
121
DatabaseMetaData metadata = connection.getMetaData();
122

123
System.out.println("Database Capabilities:");
124
System.out.println("  Read-only: " + metadata.isReadOnly());
125
System.out.println("  Supports transactions: " + metadata.supportsTransactions());
126
System.out.println("  All tables selectable: " + metadata.allTablesAreSelectable());
127
System.out.println("  Nulls sorted low: " + metadata.nullsAreSortedLow());
128
System.out.println("  Supports stored procedures: " + metadata.supportsStoredFunctionsUsingCallSyntax());
129
```
130

131
### SQL Feature Support
132

133
Query support for various SQL language features and syntax elements.
134

135
```java { .api }
136
public class FlinkDatabaseMetaData extends BaseDatabaseMetaData {
137
    // SQL syntax support
138
    public boolean supportsANSI92EntryLevelSQL() throws SQLException;
139
    public boolean supportsANSI92IntermediateSQL() throws SQLException; 
140
    public boolean supportsANSI92FullSQL() throws SQLException;
141
    
142
    // JOIN support
143
    public boolean supportsOuterJoins() throws SQLException;
144
    public boolean supportsFullOuterJoins() throws SQLException;
145
    public boolean supportsLimitedOuterJoins() throws SQLException;
146
    
147
    // Subquery support
148
    public boolean supportsCorrelatedSubqueries() throws SQLException;
149
    public boolean supportsSubqueriesInComparisons() throws SQLException;
150
    public boolean supportsSubqueriesInExists() throws SQLException;
151
    public boolean supportsSubqueriesInIns() throws SQLException;
152
    public boolean supportsSubqueriesInQuantifieds() throws SQLException;
153
    
154
    // GROUP BY support
155
    public boolean supportsGroupBy() throws SQLException;
156
    public boolean supportsGroupByUnrelated() throws SQLException;
157
    public boolean supportsGroupByBeyondSelect() throws SQLException;
158
    
159
    // ORDER BY support
160
    public boolean supportsOrderByUnrelated() throws SQLException;
161
    
162
    // UNION support
163
    public boolean supportsUnion() throws SQLException;
164
    public boolean supportsUnionAll() throws SQLException;
165
}
166
```
167

168
**Usage Example:**
169
```java
170
DatabaseMetaData metadata = connection.getMetaData();
171

172
System.out.println("SQL Feature Support:");
173
System.out.println("  ANSI SQL 92 Entry Level: " + metadata.supportsANSI92EntryLevelSQL());
174
System.out.println("  Outer Joins: " + metadata.supportsOuterJoins());
175
System.out.println("  Full Outer Joins: " + metadata.supportsFullOuterJoins());
176
System.out.println("  Correlated Subqueries: " + metadata.supportsCorrelatedSubqueries());
177
System.out.println("  GROUP BY: " + metadata.supportsGroupBy());
178
System.out.println("  UNION: " + metadata.supportsUnion());
179
System.out.println("  UNION ALL: " + metadata.supportsUnionAll());
180
```
181

182
### Identifier and String Handling
183

184
Information about identifier quoting, case sensitivity, and string handling.
185

186
```java { .api }
187
public class FlinkDatabaseMetaData extends BaseDatabaseMetaData {
188
    // Identifier handling
189
    public String getIdentifierQuoteString() throws SQLException; // Returns "`"
190
    public boolean supportsMixedCaseIdentifiers() throws SQLException; // Returns true
191
    public boolean storesMixedCaseIdentifiers() throws SQLException; // Returns true
192
    public boolean storesUpperCaseIdentifiers() throws SQLException; // Returns false
193
    public boolean storesLowerCaseIdentifiers() throws SQLException; // Returns false
194
    public boolean supportsMixedCaseQuotedIdentifiers() throws SQLException;
195
    public boolean storesMixedCaseQuotedIdentifiers() throws SQLException;
196
    public boolean storesUpperCaseQuotedIdentifiers() throws SQLException;
197
    public boolean storesLowerCaseQuotedIdentifiers() throws SQLException;
198
    
199
    // String functions and literals
200
    public String getStringFunctions() throws SQLException;
201
    public String getNumericFunctions() throws SQLException;
202
    public String getTimeDateFunctions() throws SQLException;
203
    public String getSystemFunctions() throws SQLException;
204
    public String getSQLKeywords() throws SQLException;
205
}
206
```
207

208
**Usage Example:**
209
```java
210
DatabaseMetaData metadata = connection.getMetaData();
211

212
System.out.println("Identifier Handling:");
213
System.out.println("  Quote string: '" + metadata.getIdentifierQuoteString() + "'");
214
System.out.println("  Mixed case identifiers: " + metadata.supportsMixedCaseIdentifiers());
215
System.out.println("  Stores mixed case: " + metadata.storesMixedCaseIdentifiers());
216

217
// Use quoted identifiers when needed
218
Statement stmt = connection.createStatement();
219
ResultSet results = stmt.executeQuery("SELECT `Column Name` FROM `My Table`");
220
```
221

222
### Type Information
223

224
Information about supported data types and their characteristics.
225

226
```java { .api }
227
public class FlinkDatabaseMetaData extends BaseDatabaseMetaData {
228
    public ResultSet getTypeInfo() throws SQLException; // Not fully implemented
229
    
230
    // Numeric limits
231
    public int getMaxBinaryLiteralLength() throws SQLException;
232
    public int getMaxCharLiteralLength() throws SQLException;
233
    public int getMaxColumnNameLength() throws SQLException;
234
    public int getMaxColumnsInGroupBy() throws SQLException;
235
    public int getMaxColumnsInIndex() throws SQLException;
236
    public int getMaxColumnsInOrderBy() throws SQLException;
237
    public int getMaxColumnsInSelect() throws SQLException;
238
    public int getMaxColumnsInTable() throws SQLException;
239
    public int getMaxConnections() throws SQLException;
240
    public int getMaxCursorNameLength() throws SQLException;
241
    public int getMaxIndexLength() throws SQLException;
242
    public int getMaxSchemaNameLength() throws SQLException;
243
    public int getMaxProcedureNameLength() throws SQLException;
244
    public int getMaxCatalogNameLength() throws SQLException;
245
    public int getMaxRowSize() throws SQLException;
246
    public int getMaxStatementLength() throws SQLException;
247
    public int getMaxStatements() throws SQLException;
248
    public int getMaxTableNameLength() throws SQLException;
249
    public int getMaxTablesInSelect() throws SQLException;
250
    public int getMaxUserNameLength() throws SQLException;
251
}
252
```
253

254
### Isolation and Locking
255

256
Transaction isolation and locking information (mostly not applicable to Flink).
257

258
```java { .api }
259
public class FlinkDatabaseMetaData extends BaseDatabaseMetaData {
260
    public int getDefaultTransactionIsolation() throws SQLException; // Returns TRANSACTION_NONE
261
    public boolean supportsTransactionIsolationLevel(int level) throws SQLException; // Returns false
262
    
263
    public boolean supportsResultSetType(int type) throws SQLException;
264
    public boolean supportsResultSetConcurrency(int type, int concurrency) throws SQLException;
265
    public boolean supportsResultSetHoldability(int holdability) throws SQLException;
266
    public int getResultSetHoldability() throws SQLException;
267
    
268
    public int getSQLStateType() throws SQLException;
269
    public boolean locatorsUpdateCopy() throws SQLException;
270
}
271
```
272

273
### Unsupported Metadata Operations
274

275
The following metadata operations are not supported and will throw exceptions:
276

277
```java { .api }
278
// Table and column metadata (throw UnsupportedOperationException)
279
public ResultSet getTables(String catalog, String schemaPattern, String tableNamePattern, String[] types) throws SQLException;
280
public ResultSet getColumns(String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern) throws SQLException;
281
public ResultSet getPrimaryKeys(String catalog, String schema, String table) throws SQLException;
282
public ResultSet getImportedKeys(String catalog, String schema, String table) throws SQLException;
283
public ResultSet getExportedKeys(String catalog, String schema, String table) throws SQLException;
284
public ResultSet getCrossReference(String parentCatalog, String parentSchema, String parentTable, String foreignCatalog, String foreignSchema, String foreignTable) throws SQLException;
285
public ResultSet getIndexInfo(String catalog, String schema, String table, boolean unique, boolean approximate) throws SQLException;
286

287
// Stored procedures and functions
288
public ResultSet getProcedures(String catalog, String schemaPattern, String procedureNamePattern) throws SQLException;
289
public ResultSet getProcedureColumns(String catalog, String schemaPattern, String procedureNamePattern, String columnNamePattern) throws SQLException;
290
public ResultSet getFunctions(String catalog, String schemaPattern, String functionNamePattern) throws SQLException;
291
public ResultSet getFunctionColumns(String catalog, String schemaPattern, String functionNamePattern, String columnNamePattern) throws SQLException;
292

293
// User-defined types
294
public ResultSet getUDTs(String catalog, String schemaPattern, String typeNamePattern, int[] types) throws SQLException;
295

296
// Privileges
297
public ResultSet getTablePrivileges(String catalog, String schemaPattern, String tableNamePattern) throws SQLException;
298
public ResultSet getColumnPrivileges(String catalog, String schema, String table, String columnNamePattern) throws SQLException;
299

300
// Version columns and best row identifier
301
public ResultSet getVersionColumns(String catalog, String schema, String table) throws SQLException;
302
public ResultSet getBestRowIdentifier(String catalog, String schema, String table, int scope, boolean nullable) throws SQLException;
303

304
// Attributes and super types/tables
305
public ResultSet getAttributes(String catalog, String schemaPattern, String typeNamePattern, String attributeNamePattern) throws SQLException;
306
public ResultSet getSuperTypes(String catalog, String schemaPattern, String typeNamePattern) throws SQLException;
307
public ResultSet getSuperTables(String catalog, String schemaPattern, String tableNamePattern) throws SQLException;
308

309
// Client info properties
310
public ResultSet getClientInfoProperties() throws SQLException;
311

312
// Pseudo columns
313
public ResultSet getPseudoColumns(String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern) throws SQLException;
314
```
315

316
### Utility Classes
317

318
Supporting utility classes for metadata operations:
319

320
```java { .api }
321
public class DatabaseMetaDataUtils {
322
    public static FlinkResultSet createCatalogsResultSet(Statement statement, StatementResult result);
323
    public static FlinkResultSet createSchemasResultSet(Statement statement, List<String> catalogs, Map<String, List<String>> catalogSchemas);
324
}
325
```
326

327
### Error Handling
328

329
Common metadata access errors:
330

331
**Connection Required:**
332
```java
333
DatabaseMetaData metadata = connection.getMetaData();
334
connection.close();
335
try {
336
    metadata.getCatalogs(); // May fail if connection is closed
337
} catch (SQLException e) {
338
    System.err.println("Connection closed: " + e.getMessage());
339
}
340
```
341

342
**Unsupported Operations:**
343
```java
344
try {
345
    metadata.getTables(null, null, "%", null);
346
} catch (UnsupportedOperationException e) {
347
    System.err.println("Table metadata not supported: " + e.getMessage());
348
}
349
```
350

351
**Network Issues:**
352
```java
353
try {
354
    ResultSet catalogs = metadata.getCatalogs();
355
} catch (SQLException e) {
356
    System.err.println("Failed to retrieve catalogs: " + e.getMessage());
357
    // May need to retry with a new connection
358
}
359
```
360

361
### Practical Usage Patterns
362

363
**Application Compatibility Check:**
364
```java
365
public boolean isFlinkCompatible(Connection connection) throws SQLException {
366
    DatabaseMetaData metadata = connection.getMetaData();
367
    
368
    // Check if it's actually Flink
369
    if (!"Apache Flink".equals(metadata.getDatabaseProductName())) {
370
        return false;
371
    }
372
    
373
    // Check required capabilities
374
    if (!metadata.supportsGroupBy()) {
375
        return false;
376
    }
377
    
378
    if (!metadata.supportsUnion()) {
379
        return false;
380
    }
381
    
382
    return true;
383
}
384
```
385

386
**Dynamic Schema Discovery:**
387
```java
388
public void discoverSchema(Connection connection) throws SQLException {
389
    DatabaseMetaData metadata = connection.getMetaData();
390
    
391
    // List all catalogs
392
    System.out.println("Available catalogs:");
393
    try (ResultSet catalogs = metadata.getCatalogs()) {
394
        while (catalogs.next()) {
395
            String catalog = catalogs.getString("TABLE_CAT");
396
            System.out.println("  " + catalog);
397
        }
398
    }
399
    
400
    // List all schemas
401
    System.out.println("Available schemas:");
402
    try (ResultSet schemas = metadata.getSchemas()) {
403
        while (schemas.next()) {
404
            String schema = schemas.getString("TABLE_SCHEM");
405
            String catalog = schemas.getString("TABLE_CATALOG");
406
            System.out.printf("  %s.%s%n", catalog, schema);
407
        }
408
    }
409
}
410
```
411

412
**Connection Information Display:**
413
```java
414
public void displayConnectionInfo(Connection connection) throws SQLException {
415
    DatabaseMetaData metadata = connection.getMetaData();
416
    
417
    System.out.println("=== Connection Information ===");
418
    System.out.printf("Database: %s %s%n", 
419
        metadata.getDatabaseProductName(),
420
        metadata.getDatabaseProductVersion());
421
    
422
    System.out.printf("Driver: %s %s%n",
423
        metadata.getDriverName(),
424
        metadata.getDriverVersion());
425
    
426
    System.out.printf("URL: %s%n", metadata.getURL());
427
    System.out.printf("Read-only: %s%n", metadata.isReadOnly());
428
    System.out.printf("Supports transactions: %s%n", metadata.supportsTransactions());
429
    System.out.printf("Quote string: '%s'%n", metadata.getIdentifierQuoteString());
430
}