FMWK-290 Align Aerospike namespace naming with JDBC catalog

aerospike · Dec 18, 2023 · 0a6caff · 0a6caff
1 parent 4da86f6
commit 0a6caff
Show file tree

Hide file tree

Showing 25 changed files with 130 additions and 104 deletions.
diff --git a/src/main/java/com/aerospike/jdbc/AerospikeConnection.java b/src/main/java/com/aerospike/jdbc/AerospikeConnection.java
@@ -35,7 +35,7 @@ public class AerospikeConnection implements Connection, SimpleWrapper {
     private final IAerospikeClient client;
     private final DatabaseMetadataBuilder metadataBuilder;
     private final AerospikeVersion aerospikeVersion;
-    private final AtomicReference<String> schema = new AtomicReference<>(null); // namespace
+    private final AtomicReference<String> catalog = new AtomicReference<>(null);
 
     private volatile boolean readOnly = false;
     private volatile Map<String, Class<?>> typeMap = emptyMap();
@@ -49,7 +49,7 @@ public AerospikeConnection(String url, Properties props) {
         client = config.parse(url);
         metadataBuilder = new DatabaseMetadataBuilder(config.getDriverPolicy());
         aerospikeVersion = new AerospikeVersion(client);
-        schema.set(config.getSchema()); // namespace
+        catalog.set(config.getCatalog()); // namespace
     }
 
     @Override
@@ -125,12 +125,12 @@ public void setReadOnly(boolean readOnly) throws SQLException {
 
     @Override
     public String getCatalog() {
-        return schema.get();
+        return catalog.get();
     }
 
     @Override
     public void setCatalog(String catalog) {
-        schema.set(catalog);
+        this.catalog.set(catalog);
     }
 
     @Override

diff --git a/src/main/java/com/aerospike/jdbc/AerospikePreparedStatement.java b/src/main/java/com/aerospike/jdbc/AerospikePreparedStatement.java
@@ -231,8 +231,8 @@ public ResultSetMetaData getMetaData() throws SQLException {
         AerospikeQuery query = parseQuery(prepareQueryString());
         List<DataColumn> columns = ((AerospikeDatabaseMetadata) connection.getMetaData())
                 .getSchemaBuilder()
-                .getSchema(query.getSchemaTable());
-        return new AerospikeResultSetMetaData(query.getSchema(), query.getTable(), columns);
+                .getSchema(query.getCatalogTable());
+        return new AerospikeResultSetMetaData(query.getCatalog(), query.getTable(), columns);
     }
 
     @Override
@@ -265,7 +265,7 @@ public ParameterMetaData getParameterMetaData() throws SQLException {
         AerospikeQuery query = parseQuery(prepareQueryString());
         List<DataColumn> columns = ((AerospikeDatabaseMetadata) connection.getMetaData())
                 .getSchemaBuilder()
-                .getSchema(query.getSchemaTable());
+                .getSchema(query.getCatalogTable());
         return new SimpleParameterMetaData(columns);
     }
 

diff --git a/src/main/java/com/aerospike/jdbc/AerospikeStatement.java b/src/main/java/com/aerospike/jdbc/AerospikeStatement.java
@@ -32,7 +32,7 @@ public class AerospikeStatement implements Statement, SimpleWrapper {
     protected final IAerospikeClient client;
     protected final AerospikeConnection connection;
 
-    protected String schema;
+    protected String catalog;
     protected ResultSet resultSet;
     protected int updateCount;
 
@@ -42,7 +42,7 @@ public class AerospikeStatement implements Statement, SimpleWrapper {
     public AerospikeStatement(IAerospikeClient client, AerospikeConnection connection) {
         this.client = client;
         this.connection = connection;
-        this.schema = connection.getCatalog();
+        this.catalog = connection.getCatalog();
     }
 
     @Override
@@ -67,8 +67,8 @@ protected AerospikeQuery parseQuery(String sql) throws SQLException {
         } catch (SqlParseException e) {
             query = AuxStatementParser.hack(sql);
         }
-        if (query.getSchema() == null) {
-            query.setSchema(schema);
+        if (query.getCatalog() == null) {
+            query.setCatalog(catalog);
         }
         return query;
     }

diff --git a/src/main/java/com/aerospike/jdbc/async/ScanQueryHandler.java b/src/main/java/com/aerospike/jdbc/async/ScanQueryHandler.java
@@ -42,21 +42,21 @@ public RecordSet execute(ScanPolicy scanPolicy, AerospikeQuery query) {
             long maxRecords = scanPolicy.maxRecords;
             PartitionFilter filter = getPartitionFilter(query);
             while (isScanRequired(maxRecords)) {
-                client.scanPartitions(scanPolicy, filter, query.getSchema(), query.getSetName(),
+                client.scanPartitions(scanPolicy, filter, query.getCatalog(), query.getSetName(),
                         callback, query.columnBins());
                 scanPolicy.maxRecords = maxRecords > 0 ? maxRecords - count : maxRecords;
                 filter = PartitionFilter.id(++currentPartition);
             }
             listener.onSuccess();
         } else {
-            client.scanAll(EventLoopProvider.getEventLoop(), listener, scanPolicy, query.getSchema(),
+            client.scanAll(EventLoopProvider.getEventLoop(), listener, scanPolicy, query.getCatalog(),
                     query.getSetName(), query.columnBins());
         }
         return listener.getRecordSet();
     }
 
     private PartitionFilter getPartitionFilter(AerospikeQuery query) {
-        Key key = new Key(query.getSchema(), query.getSetName(), query.getOffset());
+        Key key = new Key(query.getCatalog(), query.getSetName(), query.getOffset());
         currentPartition = Partition.getPartitionId(key.digest);
         return PartitionFilter.after(key);
     }

diff --git a/src/main/java/com/aerospike/jdbc/async/SecondaryIndexQueryHandler.java b/src/main/java/com/aerospike/jdbc/async/SecondaryIndexQueryHandler.java
@@ -30,7 +30,7 @@ public RecordSet execute(QueryPolicy queryPolicy, AerospikeQuery query,
         statement.setRecordsPerSecond(client.getScanPolicyDefault().recordsPerSecond);
 
         statement.setIndexName(secondaryIndex.getIndexName());
-        statement.setNamespace(query.getSchema());
+        statement.setNamespace(query.getCatalog());
         statement.setSetName(query.getTable());
         statement.setBinNames(query.columnBins());
 

diff --git a/src/main/java/com/aerospike/jdbc/model/AerospikeQuery.java b/src/main/java/com/aerospike/jdbc/model/AerospikeQuery.java
@@ -31,7 +31,6 @@ public class AerospikeQuery {
     private static final String ASTERISK = "*";
 
     private String catalog;
-    private String schema;
     private String table;
     private QueryType queryType;
     private Integer offset;
@@ -59,29 +58,15 @@ public void setCatalog(String catalog) {
         this.catalog = catalog;
     }
 
-    public String getSchema() {
-        return schema;
-    }
-
-    public void setSchema(String schema) {
-        this.catalog = schema; // TODO ?
-        this.schema = schema;
-    }
-
     public String getTable() {
         return table;
     }
 
     public void setTable(String table) {
         String[] spec = table.split("\\.");
         switch (spec.length) {
-            case 3:
-                this.catalog = spec[0];
-                this.schema = spec[1];
-                this.table = spec[2];
-                break;
             case 2:
-                this.schema = spec[0];
+                this.catalog = spec[0];
                 this.table = spec[1];
                 break;
             case 1:
@@ -99,8 +84,8 @@ public String getSetName() {
         return table;
     }
 
-    public SchemaTableName getSchemaTable() {
-        return new SchemaTableName(schema, table);
+    public CatalogTableName getCatalogTable() {
+        return new CatalogTableName(catalog, table);
     }
 
     public QueryType getQueryType() {

diff --git a/src/main/java/com/aerospike/jdbc/model/AerospikeSecondaryIndex.java b/src/main/java/com/aerospike/jdbc/model/AerospikeSecondaryIndex.java
@@ -38,7 +38,7 @@ public class AerospikeSecondaryIndex {
     private final IndexType indexType;
 
     @Order(3)
-    @JsonProperty("schema_name")
+    @JsonProperty("catalog_name")
     private final String namespace;
 
     @Order(4)

diff --git a/src/main/java/com/aerospike/jdbc/model/AerospikeSqlVisitor.java b/src/main/java/com/aerospike/jdbc/model/AerospikeSqlVisitor.java
@@ -78,7 +78,7 @@ public AerospikeQuery visit(SqlCall sqlCall) {
                 query.setQueryType(QueryType.DROP_TABLE);
             } else if (sqlCall instanceof SqlDropSchema) {
                 SqlDropSchema sql = (SqlDropSchema) sqlCall;
-                query.setSchema(requireNonNull(sql.name).toString());
+                query.setCatalog(requireNonNull(sql.name).toString());
                 query.setQueryType(QueryType.DROP_SCHEMA);
             } else if (sqlCall instanceof SqlOrderBy) {
                 SqlOrderBy sql = (SqlOrderBy) sqlCall;

diff --git a/src/main/java/com/aerospike/jdbc/model/CatalogTableName.java b/src/main/java/com/aerospike/jdbc/model/CatalogTableName.java
@@ -0,0 +1,41 @@
+package com.aerospike.jdbc.model;
+
+import java.util.Objects;
+
+public class CatalogTableName {
+
+    private final String catalogName;
+    private final String tableName;
+
+    public CatalogTableName(String catalogName, String tableName) {
+        this.catalogName = catalogName;
+        this.tableName = tableName;
+    }
+
+    public String getCatalogName() {
+        return catalogName;
+    }
+
+    public String getTableName() {
+        return tableName;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        CatalogTableName that = (CatalogTableName) o;
+        return Objects.equals(catalogName, that.catalogName) &&
+                Objects.equals(tableName, that.tableName);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(catalogName, tableName);
+    }
+
+    @Override
+    public String toString() {
+        return String.format("%s(%s, %s)", getClass().getSimpleName(), catalogName, tableName);
+    }
+}
diff --git a/src/main/java/com/aerospike/jdbc/model/DataColumn.java b/src/main/java/com/aerospike/jdbc/model/DataColumn.java
@@ -62,8 +62,8 @@ public int getType() {
         return type;
     }
 
-    public SchemaTableName getSchemaTableName() {
-        return new SchemaTableName(catalog, table);
+    public CatalogTableName getSchemaTableName() {
+        return new CatalogTableName(catalog, table);
     }
 
     @Override

diff --git a/src/main/java/com/aerospike/jdbc/model/DriverConfiguration.java b/src/main/java/com/aerospike/jdbc/model/DriverConfiguration.java
@@ -31,12 +31,12 @@ public final class DriverConfiguration {
 
     private static final String DEFAULT_AEROSPIKE_PORT = "3000";
 
-    private static final Pattern AS_JDBC_URL = Pattern.compile("^jdbc:aerospike:(?://)?([^/?]+)");
-    private static final Pattern AS_JDBC_SCHEMA = Pattern.compile("/([^?]+)");
+    private static final Pattern AEROSPIKE_JDBC_URL = Pattern.compile("^jdbc:aerospike:(?://)?([^/?]+)");
+    private static final Pattern AEROSPIKE_JDBC_CATALOG = Pattern.compile("/([^?]+)");
 
     private final Map<Object, Object> clientInfo = new ConcurrentHashMap<>();
     private volatile IAerospikeClient client;
-    private volatile String schema;
+    private volatile String catalog;
     private volatile ClientPolicy clientPolicy;
     private volatile DriverPolicy driverPolicy;
 
@@ -48,7 +48,7 @@ public DriverConfiguration(Properties props) {
     @SuppressWarnings("java:S2696")
     public IAerospikeClient parse(String url) {
         logger.info(() -> format("Parse URL: %s", url));
-        schema = parseSchema(url);
+        catalog = parseCatalog(url);
         updateClientInfo(url);
 
         Value.UseBoolBin = Optional.ofNullable(clientInfo.get("useBoolBin"))
@@ -113,7 +113,7 @@ private TlsPolicy buildTlsPolicy() {
     }
 
     private Host[] parseHosts(String url, final String tlsName) {
-        Matcher m = AS_JDBC_URL.matcher(url);
+        Matcher m = AEROSPIKE_JDBC_URL.matcher(url);
         if (!m.find()) {
             throw new IllegalArgumentException("Cannot parse URL " + url);
         }
@@ -124,8 +124,8 @@ private Host[] parseHosts(String url, final String tlsName) {
                 .toArray(Host[]::new);
     }
 
-    private String parseSchema(String url) {
-        Matcher m = AS_JDBC_SCHEMA.matcher(url);
+    private String parseCatalog(String url) {
+        Matcher m = AEROSPIKE_JDBC_CATALOG.matcher(url);
         return m.find() ? m.group(1) : null;
     }
 
@@ -141,8 +141,8 @@ private void updateClientInfo(String url) {
         }
     }
 
-    public String getSchema() {
-        return schema;
+    public String getCatalog() {
+        return catalog;
     }
 
     public Properties getClientInfo() {

diff --git a/src/main/java/com/aerospike/jdbc/query/BaseQueryHandler.java b/src/main/java/com/aerospike/jdbc/query/BaseQueryHandler.java
@@ -48,7 +48,7 @@ protected Bin[] getBins(AerospikeQuery query) {
     }
 
     protected ListRecordSet emptyRecordSet(AerospikeQuery query) {
-        return new ListRecordSet(statement, query.getSchema(), query.getTable(),
+        return new ListRecordSet(statement, query.getCatalog(), query.getTable(),
                 emptyList(), emptyList());
     }
 

diff --git a/src/main/java/com/aerospike/jdbc/query/DeleteQueryHandler.java b/src/main/java/com/aerospike/jdbc/query/DeleteQueryHandler.java
@@ -34,7 +34,7 @@ public Pair<ResultSet, Integer> execute(AerospikeQuery query) {
             logger.info("DELETE primary key");
             FutureDeleteListener listener = new FutureDeleteListener(keyObjects.size());
             for (Object keyObject : keyObjects) {
-                Key key = new Key(query.getSchema(), query.getSetName(), Value.get(keyObject));
+                Key key = new Key(query.getCatalog(), query.getSetName(), Value.get(keyObject));
                 try {
                     client.delete(EventLoopProvider.getEventLoop(), listener, writePolicy, key);
                 } catch (AerospikeException e) {
@@ -49,7 +49,7 @@ public Pair<ResultSet, Integer> execute(AerospikeQuery query) {
             ScanPolicy scanPolicy = policyBuilder.buildScanPolicy(query);
             scanPolicy.includeBinData = false;
 
-            client.scanAll(EventLoopProvider.getEventLoop(), listener, scanPolicy, query.getSchema(),
+            client.scanAll(EventLoopProvider.getEventLoop(), listener, scanPolicy, query.getCatalog(),
                     query.getSetName());
 
             final WritePolicy deletePolicy = policyBuilder.buildDeleteWritePolicy();

diff --git a/src/main/java/com/aerospike/jdbc/query/InsertQueryHandler.java b/src/main/java/com/aerospike/jdbc/query/InsertQueryHandler.java
@@ -49,7 +49,7 @@ public Pair<ResultSet, Integer> putConsecutively(AerospikeQuery query) {
             @SuppressWarnings("unchecked")
             List<Object> values = (List<Object>) aerospikeRecord;
             Value recordKey = extractInsertKey(query, values);
-            Key key = new Key(query.getSchema(), query.getSetName(), recordKey);
+            Key key = new Key(query.getCatalog(), query.getSetName(), recordKey);
             Bin[] bins = buildBinArray(binNames, values);
 
             try {
@@ -73,7 +73,7 @@ public Pair<ResultSet, Integer> putBatch(AerospikeQuery query) {
             @SuppressWarnings("unchecked")
             List<Object> values = (List<Object>) aerospikeRecord;
             Value recordKey = extractInsertKey(query, values);
-            Key key = new Key(query.getSchema(), query.getSetName(), recordKey);
+            Key key = new Key(query.getCatalog(), query.getSetName(), recordKey);
             batchRecords.add(
                     new BatchWrite(
                             batchWritePolicy,

diff --git a/src/main/java/com/aerospike/jdbc/query/SelectQueryHandler.java b/src/main/java/com/aerospike/jdbc/query/SelectQueryHandler.java
@@ -54,7 +54,7 @@ public SelectQueryHandler(IAerospikeClient client, Statement statement) {
 
     @Override
     public Pair<ResultSet, Integer> execute(AerospikeQuery query) {
-        columns = databaseMetadata.getSchemaBuilder().getSchema(query.getSchemaTable());
+        columns = databaseMetadata.getSchemaBuilder().getSchema(query.getCatalogTable());
         Collection<Object> keyObjects = query.getPrimaryKeys();
         Optional<AerospikeSecondaryIndex> sIndex = secondaryIndex(query);
         Pair<ResultSet, Integer> result;
@@ -74,7 +74,7 @@ private Pair<ResultSet, Integer> executeCountQuery(AerospikeQuery query) {
         String countLabel = query.getColumns().get(0);
         int recordNumber;
         if (Objects.isNull(query.getPredicate())) {
-            recordNumber = getTableRecordsNumber(client, query.getSchema(), query.getTable());
+            recordNumber = getTableRecordsNumber(client, query.getCatalog(), query.getTable());
         } else {
             ScanPolicy policy = policyBuilder.buildScanNoBinDataPolicy(query);
             RecordSet recordSet = ScanQueryHandler.create(client, config.getDriverPolicy())
@@ -91,7 +91,7 @@ private Pair<ResultSet, Integer> executeCountQuery(AerospikeQuery query) {
         recordSet.put(new KeyRecord(null, aeroRecord));
         recordSet.close();
 
-        columns = Collections.singletonList(new DataColumn(query.getSchema(), query.getTable(),
+        columns = Collections.singletonList(new DataColumn(query.getCatalog(), query.getTable(),
                 Types.INTEGER, countLabel, countLabel));
 
         return queryResult(recordSet, query);
@@ -102,7 +102,7 @@ private Pair<ResultSet, Integer> executeSelectByPrimaryKey(AerospikeQuery query,
         final BatchReadPolicy policy = policyBuilder.buildBatchReadPolicy(query);
         List<BatchRead> batchReadList = keyObjects.stream()
                 .map(k -> {
-                    Key key = new Key(query.getSchema(), query.getSetName(), Value.get(k));
+                    Key key = new Key(query.getCatalog(), query.getSetName(), Value.get(k));
                     return new BatchRead(policy, key, true);
                 })
                 .collect(Collectors.toList());
@@ -135,7 +135,7 @@ private Pair<ResultSet, Integer> executeQuery(AerospikeQuery query,
 
     private Optional<AerospikeSecondaryIndex> secondaryIndex(AerospikeQuery query) {
         if (aerospikeVersion.isSIndexSupported() && query.isIndexable()) {
-            Collection<AerospikeSecondaryIndex> indexes = databaseMetadata.getSecondaryIndexes(query.getSchema());
+            Collection<AerospikeSecondaryIndex> indexes = databaseMetadata.getSecondaryIndexes(query.getCatalog());
             List<String> binNames = query.getPredicate().getBinNames();
             if (!binNames.isEmpty() && indexes != null && !indexes.isEmpty()) {
                 List<AerospikeSecondaryIndex> indexList = indexes.stream()
@@ -154,7 +154,7 @@ private Optional<AerospikeSecondaryIndex> secondaryIndex(AerospikeQuery query) {
     }
 
     private Pair<ResultSet, Integer> queryResult(RecordSet recordSet, AerospikeQuery query) {
-        return new Pair<>(new AerospikeRecordResultSet(recordSet, statement, query.getSchema(),
+        return new Pair<>(new AerospikeRecordResultSet(recordSet, statement, query.getCatalog(),
                 query.getTable(), filterColumns(query)), -1);
     }