You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sd...@apache.org on 2022/01/12 12:40:16 UTC
[ignite-3] branch ignite-16250 updated: Async trash
This is an automated email from the ASF dual-hosted git repository.
sdanilov pushed a commit to branch ignite-16250
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/ignite-16250 by this push:
new 553c1bc Async trash
553c1bc is described below
commit 553c1bc02c3048eccf2d6dc60800b5f2fa69de60
Author: Semyon Danilov <sa...@yandex.ru>
AuthorDate: Wed Jan 12 15:39:31 2022 +0300
Async trash
---
.../client/proto/query/JdbcQueryEventHandler.java | 19 ++---
.../client/handler/JdbcQueryEventHandlerImpl.java | 94 ++++++++++------------
.../requests/sql/ClientSqlCloseRequest.java | 6 +-
.../sql/ClientSqlColumnMetadataRequest.java | 6 +-
.../requests/sql/ClientSqlExecuteBatchRequest.java | 6 +-
.../requests/sql/ClientSqlExecuteRequest.java | 7 +-
.../requests/sql/ClientSqlFetchRequest.java | 6 +-
.../sql/ClientSqlPrimaryKeyMetadataRequest.java | 6 +-
.../sql/ClientSqlQueryMetadataRequest.java | 6 +-
.../sql/ClientSqlSchemasMetadataRequest.java | 6 +-
.../sql/ClientSqlTableMetadataRequest.java | 6 +-
.../handler/requests/sql/JdbcMetadataCatalog.java | 84 ++++++++-----------
.../ignite/internal/client/TcpIgniteClient.java | 8 +-
.../client/query/JdbcClientQueryEventHandler.java | 55 +++++--------
.../ignite/internal/jdbc/JdbcDatabaseMetadata.java | 8 +-
.../apache/ignite/internal/jdbc/JdbcResultSet.java | 6 +-
.../apache/ignite/internal/jdbc/JdbcStatement.java | 4 +-
.../scalecube/DelegatingTransportFactory.java | 4 +
.../internal/table/distributed/TableManager.java | 43 +++++-----
19 files changed, 154 insertions(+), 226 deletions(-)
diff --git a/modules/client-common/src/main/java/org/apache/ignite/client/proto/query/JdbcQueryEventHandler.java b/modules/client-common/src/main/java/org/apache/ignite/client/proto/query/JdbcQueryEventHandler.java
index c8f5540..f7d9200 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/client/proto/query/JdbcQueryEventHandler.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/client/proto/query/JdbcQueryEventHandler.java
@@ -17,6 +17,7 @@
package org.apache.ignite.client.proto.query;
+import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.proto.query.event.BatchExecuteRequest;
import org.apache.ignite.client.proto.query.event.BatchExecuteResult;
import org.apache.ignite.client.proto.query.event.JdbcMetaColumnsRequest;
@@ -45,7 +46,7 @@ public interface JdbcQueryEventHandler {
* @param req Execute query request.
* @return Result.
*/
- QueryExecuteResult query(QueryExecuteRequest req);
+ CompletableFuture<QueryExecuteResult> query(QueryExecuteRequest req);
/**
* {@link QueryFetchRequest} command handler.
@@ -53,7 +54,7 @@ public interface JdbcQueryEventHandler {
* @param req Fetch query request.
* @return Result.
*/
- QueryFetchResult fetch(QueryFetchRequest req);
+ CompletableFuture<QueryFetchResult> fetch(QueryFetchRequest req);
/**
* {@link BatchExecuteRequest} command handler.
@@ -61,7 +62,7 @@ public interface JdbcQueryEventHandler {
* @param req Batch query request.
* @return Result.
*/
- BatchExecuteResult batch(BatchExecuteRequest req);
+ CompletableFuture<BatchExecuteResult> batch(BatchExecuteRequest req);
/**
* {@link QueryCloseRequest} command handler.
@@ -69,7 +70,7 @@ public interface JdbcQueryEventHandler {
* @param req Close query request.
* @return Result.
*/
- QueryCloseResult close(QueryCloseRequest req);
+ CompletableFuture<QueryCloseResult> close(QueryCloseRequest req);
/**
* {@link JdbcMetaTablesRequest} command handler.
@@ -77,7 +78,7 @@ public interface JdbcQueryEventHandler {
* @param req Jdbc tables metadata request.
* @return Result.
*/
- JdbcMetaTablesResult tablesMeta(JdbcMetaTablesRequest req);
+ CompletableFuture<JdbcMetaTablesResult> tablesMeta(JdbcMetaTablesRequest req);
/**
* {@link JdbcMetaColumnsRequest} command handler.
@@ -85,7 +86,7 @@ public interface JdbcQueryEventHandler {
* @param req Jdbc columns metadata request.
* @return Result.
*/
- JdbcMetaColumnsResult columnsMeta(JdbcMetaColumnsRequest req);
+ CompletableFuture<JdbcMetaColumnsResult> columnsMeta(JdbcMetaColumnsRequest req);
/**
* {@link JdbcMetaSchemasRequest} command handler.
@@ -93,7 +94,7 @@ public interface JdbcQueryEventHandler {
* @param req Jdbc schemas metadata request.
* @return Result.
*/
- JdbcMetaSchemasResult schemasMeta(JdbcMetaSchemasRequest req);
+ CompletableFuture<JdbcMetaSchemasResult> schemasMeta(JdbcMetaSchemasRequest req);
/**
* {@link JdbcMetaPrimaryKeysRequest} command handler.
@@ -101,7 +102,7 @@ public interface JdbcQueryEventHandler {
* @param req Jdbc primary keys metadata request.
* @return Result.
*/
- JdbcMetaPrimaryKeysResult primaryKeysMeta(JdbcMetaPrimaryKeysRequest req);
+ CompletableFuture<JdbcMetaPrimaryKeysResult> primaryKeysMeta(JdbcMetaPrimaryKeysRequest req);
/**
* {@link JdbcQueryMetadataRequest} command handler.
@@ -109,5 +110,5 @@ public interface JdbcQueryEventHandler {
* @param req Jdbc query metadata request.
* @return Result.
*/
- JdbcMetaColumnsResult queryMetadata(JdbcQueryMetadataRequest req);
+ CompletableFuture<JdbcMetaColumnsResult> queryMetadata(JdbcQueryMetadataRequest req);
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java
index 195b68a..a80f781 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java
@@ -22,8 +22,8 @@ import static org.apache.ignite.client.proto.query.IgniteQueryErrorCode.UNSUPPOR
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
@@ -40,9 +40,7 @@ import org.apache.ignite.client.proto.query.event.JdbcMetaSchemasRequest;
import org.apache.ignite.client.proto.query.event.JdbcMetaSchemasResult;
import org.apache.ignite.client.proto.query.event.JdbcMetaTablesRequest;
import org.apache.ignite.client.proto.query.event.JdbcMetaTablesResult;
-import org.apache.ignite.client.proto.query.event.JdbcPrimaryKeyMeta;
import org.apache.ignite.client.proto.query.event.JdbcQueryMetadataRequest;
-import org.apache.ignite.client.proto.query.event.JdbcTableMeta;
import org.apache.ignite.client.proto.query.event.QueryCloseRequest;
import org.apache.ignite.client.proto.query.event.QueryCloseResult;
import org.apache.ignite.client.proto.query.event.QueryExecuteRequest;
@@ -87,10 +85,10 @@ public class JdbcQueryEventHandlerImpl implements JdbcQueryEventHandler {
/** {@inheritDoc} */
@Override
- public QueryExecuteResult query(QueryExecuteRequest req) {
+ public CompletableFuture<QueryExecuteResult> query(QueryExecuteRequest req) {
if (req.pageSize() <= 0) {
- return new QueryExecuteResult(Response.STATUS_FAILED,
- "Invalid fetch size : [fetchSize=" + req.pageSize() + ']');
+ return CompletableFuture.completedFuture(new QueryExecuteResult(Response.STATUS_FAILED,
+ "Invalid fetch size : [fetchSize=" + req.pageSize() + ']'));
}
List<SqlCursor<List<?>>> cursors;
@@ -99,13 +97,13 @@ public class JdbcQueryEventHandlerImpl implements JdbcQueryEventHandler {
} catch (Exception e) {
StringWriter sw = getWriterWithStackTrace(e);
- return new QueryExecuteResult(Response.STATUS_FAILED,
- "Exception while executing query " + req.sqlQuery() + ". Error message: " + sw);
+ return CompletableFuture.completedFuture(new QueryExecuteResult(Response.STATUS_FAILED,
+ "Exception while executing query " + req.sqlQuery() + ". Error message: " + sw));
}
if (cursors.isEmpty()) {
- return new QueryExecuteResult(Response.STATUS_FAILED,
- "At least one cursor is expected for query " + req.sqlQuery());
+ return CompletableFuture.completedFuture(new QueryExecuteResult(Response.STATUS_FAILED,
+ "At least one cursor is expected for query " + req.sqlQuery()));
}
List<QuerySingleResult> results = new ArrayList<>();
@@ -118,26 +116,26 @@ public class JdbcQueryEventHandlerImpl implements JdbcQueryEventHandler {
} catch (Exception ex) {
StringWriter sw = getWriterWithStackTrace(ex);
- return new QueryExecuteResult(Response.STATUS_FAILED,
- "Failed to fetch results for query " + req.sqlQuery() + ". Error message: " + sw);
+ return CompletableFuture.completedFuture(new QueryExecuteResult(Response.STATUS_FAILED,
+ "Failed to fetch results for query " + req.sqlQuery() + ". Error message: " + sw));
}
- return new QueryExecuteResult(results);
+ return CompletableFuture.completedFuture(new QueryExecuteResult(results));
}
/** {@inheritDoc} */
@Override
- public QueryFetchResult fetch(QueryFetchRequest req) {
+ public CompletableFuture<QueryFetchResult> fetch(QueryFetchRequest req) {
Cursor<List<?>> cur = openCursors.get(req.cursorId());
if (cur == null) {
- return new QueryFetchResult(Response.STATUS_FAILED,
- "Failed to find query cursor with ID: " + req.cursorId());
+ return CompletableFuture.completedFuture(new QueryFetchResult(Response.STATUS_FAILED,
+ "Failed to find query cursor with ID: " + req.cursorId()));
}
if (req.pageSize() <= 0) {
- return new QueryFetchResult(Response.STATUS_FAILED,
- "Invalid fetch size : [fetchSize=" + req.pageSize() + ']');
+ return CompletableFuture.completedFuture(new QueryFetchResult(Response.STATUS_FAILED,
+ "Invalid fetch size : [fetchSize=" + req.pageSize() + ']'));
}
List<List<Object>> fetch;
@@ -149,28 +147,28 @@ public class JdbcQueryEventHandlerImpl implements JdbcQueryEventHandler {
} catch (Exception ex) {
StringWriter sw = getWriterWithStackTrace(ex);
- return new QueryFetchResult(Response.STATUS_FAILED,
- "Failed to fetch results for cursor id " + req.cursorId() + ". Error message: " + sw);
+ return CompletableFuture.completedFuture(new QueryFetchResult(Response.STATUS_FAILED,
+ "Failed to fetch results for cursor id " + req.cursorId() + ". Error message: " + sw));
}
- return new QueryFetchResult(fetch, hasNext);
+ return CompletableFuture.completedFuture(new QueryFetchResult(fetch, hasNext));
}
/** {@inheritDoc} */
@Override
- public BatchExecuteResult batch(BatchExecuteRequest req) {
- return new BatchExecuteResult(UNSUPPORTED_OPERATION,
- "ExecuteBatch operation is not implemented yet.");
+ public CompletableFuture<BatchExecuteResult> batch(BatchExecuteRequest req) {
+ return CompletableFuture.completedFuture(new BatchExecuteResult(UNSUPPORTED_OPERATION,
+ "ExecuteBatch operation is not implemented yet."));
}
/** {@inheritDoc} */
@Override
- public QueryCloseResult close(QueryCloseRequest req) {
+ public CompletableFuture<QueryCloseResult> close(QueryCloseRequest req) {
Cursor<List<?>> cur = openCursors.remove(req.cursorId());
if (cur == null) {
- return new QueryCloseResult(Response.STATUS_FAILED,
- "Failed to find query cursor with ID: " + req.cursorId());
+ return CompletableFuture.completedFuture(new QueryCloseResult(Response.STATUS_FAILED,
+ "Failed to find query cursor with ID: " + req.cursorId()));
}
try {
@@ -178,35 +176,35 @@ public class JdbcQueryEventHandlerImpl implements JdbcQueryEventHandler {
} catch (Exception ex) {
StringWriter sw = getWriterWithStackTrace(ex);
- return new QueryCloseResult(Response.STATUS_FAILED,
- "Failed to close SQL query [curId=" + req.cursorId() + "]. Error message: " + sw);
+ return CompletableFuture.completedFuture(new QueryCloseResult(Response.STATUS_FAILED,
+ "Failed to close SQL query [curId=" + req.cursorId() + "]. Error message: " + sw));
}
- return new QueryCloseResult();
+ return CompletableFuture.completedFuture(new QueryCloseResult());
}
/** {@inheritDoc} */
@Override
- public JdbcMetaColumnsResult queryMetadata(JdbcQueryMetadataRequest req) {
+ public CompletableFuture<JdbcMetaColumnsResult> queryMetadata(JdbcQueryMetadataRequest req) {
SqlCursor<List<?>> cur = openCursors.get(req.cursorId());
if (cur == null) {
- return new JdbcMetaColumnsResult(Response.STATUS_FAILED,
- "Failed to find query cursor with ID: " + req.cursorId());
+ return CompletableFuture.completedFuture(new JdbcMetaColumnsResult(Response.STATUS_FAILED,
+ "Failed to find query cursor with ID: " + req.cursorId()));
}
ResultSetMetadata metadata = cur.metadata();
if (metadata == null) {
- return new JdbcMetaColumnsResult(Response.STATUS_FAILED,
- "Failed to get query metadata for cursor with ID : " + req.cursorId());
+ return CompletableFuture.completedFuture(new JdbcMetaColumnsResult(Response.STATUS_FAILED,
+ "Failed to get query metadata for cursor with ID : " + req.cursorId()));
}
List<JdbcColumnMeta> meta = metadata.fields().stream()
.map(this::createColumnMetadata)
.collect(Collectors.toList());
- return new JdbcMetaColumnsResult(meta);
+ return CompletableFuture.completedFuture(new JdbcMetaColumnsResult(meta));
}
/**
@@ -236,34 +234,26 @@ public class JdbcQueryEventHandlerImpl implements JdbcQueryEventHandler {
/** {@inheritDoc} */
@Override
- public JdbcMetaTablesResult tablesMeta(JdbcMetaTablesRequest req) {
- List<JdbcTableMeta> tblsMeta = meta.getTablesMeta(req.schemaName(), req.tableName(), req.tableTypes());
-
- return new JdbcMetaTablesResult(tblsMeta);
+ public CompletableFuture<JdbcMetaTablesResult> tablesMeta(JdbcMetaTablesRequest req) {
+ return meta.getTablesMeta(req.schemaName(), req.tableName(), req.tableTypes()).thenApply(JdbcMetaTablesResult::new);
}
/** {@inheritDoc} */
@Override
- public JdbcMetaColumnsResult columnsMeta(JdbcMetaColumnsRequest req) {
- Collection<JdbcColumnMeta> tblsMeta = meta.getColumnsMeta(req.schemaName(), req.tableName(), req.columnName());
-
- return new JdbcMetaColumnsResult(tblsMeta);
+ public CompletableFuture<JdbcMetaColumnsResult> columnsMeta(JdbcMetaColumnsRequest req) {
+ return meta.getColumnsMeta(req.schemaName(), req.tableName(), req.columnName()).thenApply(JdbcMetaColumnsResult::new);
}
/** {@inheritDoc} */
@Override
- public JdbcMetaSchemasResult schemasMeta(JdbcMetaSchemasRequest req) {
- Collection<String> tblsMeta = meta.getSchemasMeta(req.schemaName());
-
- return new JdbcMetaSchemasResult(tblsMeta);
+ public CompletableFuture<JdbcMetaSchemasResult> schemasMeta(JdbcMetaSchemasRequest req) {
+ return meta.getSchemasMeta(req.schemaName()).thenApply(JdbcMetaSchemasResult::new);
}
/** {@inheritDoc} */
@Override
- public JdbcMetaPrimaryKeysResult primaryKeysMeta(JdbcMetaPrimaryKeysRequest req) {
- Collection<JdbcPrimaryKeyMeta> tblsMeta = meta.getPrimaryKeys(req.schemaName(), req.tableName());
-
- return new JdbcMetaPrimaryKeysResult(tblsMeta);
+ public CompletableFuture<JdbcMetaPrimaryKeysResult> primaryKeysMeta(JdbcMetaPrimaryKeysRequest req) {
+ return meta.getPrimaryKeys(req.schemaName(), req.tableName()).thenApply(JdbcMetaPrimaryKeysResult::new);
}
/**
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlCloseRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlCloseRequest.java
index a9e3dd1..94f4ce5 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlCloseRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlCloseRequest.java
@@ -45,10 +45,6 @@ public class ClientSqlCloseRequest {
req.readBinary(in);
- QueryCloseResult res = handler.close(req);
-
- res.writeBinary(out);
-
- return null;
+ return handler.close(req).thenAccept(res -> res.writeBinary(out));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlColumnMetadataRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlColumnMetadataRequest.java
index 9ffc615..7296190 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlColumnMetadataRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlColumnMetadataRequest.java
@@ -45,10 +45,6 @@ public class ClientSqlColumnMetadataRequest {
req.readBinary(in);
- JdbcMetaColumnsResult res = handler.columnsMeta(req);
-
- res.writeBinary(out);
-
- return null;
+ return handler.columnsMeta(req).thenAccept(res -> res.writeBinary(out));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteBatchRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteBatchRequest.java
index d9eff7e..bc68c57 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteBatchRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteBatchRequest.java
@@ -45,10 +45,6 @@ public class ClientSqlExecuteBatchRequest {
req.readBinary(in);
- BatchExecuteResult res = handler.batch(req);
-
- res.writeBinary(out);
-
- return null;
+ return handler.batch(req).thenAccept(res -> res.writeBinary(out));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java
index 8a3677d..280061e 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java
@@ -20,7 +20,6 @@ package org.apache.ignite.client.handler.requests.sql;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.proto.query.JdbcQueryEventHandler;
import org.apache.ignite.client.proto.query.event.QueryExecuteRequest;
-import org.apache.ignite.client.proto.query.event.QueryExecuteResult;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
@@ -45,10 +44,6 @@ public class ClientSqlExecuteRequest {
req.readBinary(in);
- QueryExecuteResult res = handler.query(req);
-
- res.writeBinary(out);
-
- return null;
+ return handler.query(req).thenAccept(res -> res.writeBinary(out));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlFetchRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlFetchRequest.java
index 6a08deb..5e628da 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlFetchRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlFetchRequest.java
@@ -45,10 +45,6 @@ public class ClientSqlFetchRequest {
req.readBinary(in);
- QueryFetchResult res = handler.fetch(req);
-
- res.writeBinary(out);
-
- return null;
+ return handler.fetch(req).thenAccept(res -> res.writeBinary(out));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlPrimaryKeyMetadataRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlPrimaryKeyMetadataRequest.java
index 3cbefad..22da23c 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlPrimaryKeyMetadataRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlPrimaryKeyMetadataRequest.java
@@ -45,10 +45,6 @@ public class ClientSqlPrimaryKeyMetadataRequest {
req.readBinary(in);
- JdbcMetaPrimaryKeysResult res = handler.primaryKeysMeta(req);
-
- res.writeBinary(out);
-
- return null;
+ return handler.primaryKeysMeta(req).thenAccept(res -> res.writeBinary(out));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlQueryMetadataRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlQueryMetadataRequest.java
index 1b46df7..a665afe 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlQueryMetadataRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlQueryMetadataRequest.java
@@ -45,10 +45,6 @@ public class ClientSqlQueryMetadataRequest {
req.readBinary(in);
- JdbcMetaColumnsResult res = handler.queryMetadata(req);
-
- res.writeBinary(out);
-
- return null;
+ return handler.queryMetadata(req).thenAccept(res -> res.writeBinary(out));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlSchemasMetadataRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlSchemasMetadataRequest.java
index a751ca2..22257ca 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlSchemasMetadataRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlSchemasMetadataRequest.java
@@ -45,10 +45,6 @@ public class ClientSqlSchemasMetadataRequest {
req.readBinary(in);
- JdbcMetaSchemasResult res = handler.schemasMeta(req);
-
- res.writeBinary(out);
-
- return null;
+ return handler.schemasMeta(req).thenAccept(res -> res.writeBinary(out));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlTableMetadataRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlTableMetadataRequest.java
index 3c9893f..d005b00 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlTableMetadataRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlTableMetadataRequest.java
@@ -45,10 +45,6 @@ public class ClientSqlTableMetadataRequest {
req.readBinary(in);
- JdbcMetaTablesResult res = handler.tablesMeta(req);
-
- res.writeBinary(out);
-
- return null;
+ return handler.tablesMeta(req).thenAccept(res -> res.writeBinary(out));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/JdbcMetadataCatalog.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/JdbcMetadataCatalog.java
index 64d2d87..fadc0b8 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/JdbcMetadataCatalog.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/JdbcMetadataCatalog.java
@@ -27,6 +27,7 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
+import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.ignite.client.proto.query.event.JdbcColumnMeta;
@@ -95,22 +96,15 @@ public class JdbcMetadataCatalog {
* @param tblNamePtrn Sql pattern for table name.
* @return Collection of primary keys information for tables that matches specified schema and table name patterns.
*/
- public Collection<JdbcPrimaryKeyMeta> getPrimaryKeys(String schemaNamePtrn, String tblNamePtrn) {
- Collection<JdbcPrimaryKeyMeta> metaSet = new HashSet<>();
-
+ public CompletableFuture<Collection<JdbcPrimaryKeyMeta>> getPrimaryKeys(String schemaNamePtrn, String tblNamePtrn) {
String schemaNameRegex = translateSqlWildcardsToRegex(schemaNamePtrn);
String tlbNameRegex = translateSqlWildcardsToRegex(tblNamePtrn);
- tables.tables().stream()
- .filter(t -> matches(getTblSchema(t.name()), schemaNameRegex))
- .filter(t -> matches(getTblName(t.name()), tlbNameRegex))
- .forEach(tbl -> {
- JdbcPrimaryKeyMeta meta = createPrimaryKeyMeta(tbl);
-
- metaSet.add(meta);
- });
-
- return metaSet;
+ return tables.tablesAsync().thenApply(tableList -> tableList.stream()
+ .filter(t -> matches(getTblSchema(t.name()), schemaNameRegex))
+ .filter(t -> matches(getTblName(t.name()), tlbNameRegex))
+ .map(this::createPrimaryKeyMeta)
+ .collect(Collectors.toSet()));
}
/**
@@ -126,19 +120,18 @@ public class JdbcMetadataCatalog {
* @param tblTypes Requested table types.
* @return List of metadatas of tables that matches.
*/
- public List<JdbcTableMeta> getTablesMeta(String schemaNamePtrn, String tblNamePtrn, String[] tblTypes) {
+ public CompletableFuture<List<JdbcTableMeta>> getTablesMeta(String schemaNamePtrn, String tblNamePtrn, String[] tblTypes) {
String schemaNameRegex = translateSqlWildcardsToRegex(schemaNamePtrn);
String tlbNameRegex = translateSqlWildcardsToRegex(tblNamePtrn);
- List<Table> tblsMeta = tables.tables().stream()
+ return tables.tablesAsync().thenApply(tablesList -> {
+ return tablesList.stream()
.filter(t -> matches(getTblSchema(t.name()), schemaNameRegex))
.filter(t -> matches(getTblName(t.name()), tlbNameRegex))
- .collect(Collectors.toList());
-
- return tblsMeta.stream()
.sorted(byTblTypeThenSchemaThenTblName)
.map(t -> new JdbcTableMeta(getTblSchema(t.name()), getTblName(t.name()), TBL_TYPE))
.collect(Collectors.toList());
+ });
}
/**
@@ -151,43 +144,36 @@ public class JdbcMetadataCatalog {
* @param colNamePtrn Column name java regex pattern.
* @return List of metadatas about columns that match specified schema/tablename/columnname criterias.
*/
- public Collection<JdbcColumnMeta> getColumnsMeta(String schemaNamePtrn, String tblNamePtrn, String colNamePtrn) {
- Collection<JdbcColumnMeta> metas = new LinkedHashSet<>();
-
+ public CompletableFuture<Collection<JdbcColumnMeta>> getColumnsMeta(String schemaNamePtrn, String tblNamePtrn, String colNamePtrn) {
String schemaNameRegex = translateSqlWildcardsToRegex(schemaNamePtrn);
String tlbNameRegex = translateSqlWildcardsToRegex(tblNamePtrn);
String colNameRegex = translateSqlWildcardsToRegex(colNamePtrn);
- tables.tables().stream()
+ return tables.tablesAsync().thenApply(tablesList -> {
+ return tablesList.stream()
.filter(t -> matches(getTblSchema(t.name()), schemaNameRegex))
.filter(t -> matches(getTblName(t.name()), tlbNameRegex))
.flatMap(
- tbl -> {
- SchemaDescriptor schema = ((TableImpl) tbl).schemaView().schema();
+ tbl -> {
+ SchemaDescriptor schema = ((TableImpl) tbl).schemaView().schema();
- List<Pair<String, Column>> tblColPairs = new ArrayList<>();
+ List<Pair<String, Column>> tblColPairs = new ArrayList<>();
- for (Column column : schema.keyColumns().columns()) {
- tblColPairs.add(new Pair<>(tbl.name(), column));
- }
+ for (Column column : schema.keyColumns().columns()) {
+ tblColPairs.add(new Pair<>(tbl.name(), column));
+ }
- for (Column column : schema.valueColumns().columns()) {
- tblColPairs.add(new Pair<>(tbl.name(), column));
- }
+ for (Column column : schema.valueColumns().columns()) {
+ tblColPairs.add(new Pair<>(tbl.name(), column));
+ }
- return tblColPairs.stream();
- })
+ return tblColPairs.stream();
+ })
.filter(e -> matches(e.getSecond().name(), colNameRegex))
.sorted(bySchemaThenTabNameThenColOrder)
- .forEachOrdered(pair -> {
- JdbcColumnMeta colMeta = createColumnMeta(pair.getFirst(), pair.getSecond());
-
- if (!metas.contains(colMeta)) {
- metas.add(colMeta);
- }
- });
-
- return metas;
+ .map(pair -> createColumnMeta(pair.getFirst(), pair.getSecond()))
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ });
}
/**
@@ -198,7 +184,7 @@ public class JdbcMetadataCatalog {
* @param schemaNamePtrn Sql pattern for schema name filter.
* @return schema names that matches provided pattern.
*/
- public Collection<String> getSchemasMeta(String schemaNamePtrn) {
+ public CompletableFuture<Collection<String>> getSchemasMeta(String schemaNamePtrn) {
SortedSet<String> schemas = new TreeSet<>(); // to have values sorted.
String schemaNameRegex = translateSqlWildcardsToRegex(schemaNamePtrn);
@@ -207,12 +193,12 @@ public class JdbcMetadataCatalog {
schemas.add(DEFAULT_SCHEMA_NAME);
}
- tables.tables().stream()
- .map(tbl -> getTblSchema(tbl.name()))
- .filter(schema -> matches(schema, schemaNameRegex))
- .forEach(schemas::add);
-
- return schemas;
+ return tables.tablesAsync().thenApply(tablesList ->
+ tablesList.stream()
+ .map(tbl -> getTblSchema(tbl.name()))
+ .filter(schema -> matches(schema, schemaNameRegex))
+ .collect(Collectors.toCollection(() -> schemas))
+ );
}
/**
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
index 86f4d74..dbf496e 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
@@ -135,15 +135,15 @@ public class TcpIgniteClient implements IgniteClient {
/**
* Send ClientMessage request to server size and reads ClientMessage result.
- *
* @param opCode Operation code.
* @param req ClientMessage request.
* @param res ClientMessage result.
+ * @return
*/
- public void sendRequest(int opCode, ClientMessage req, ClientMessage res) {
- ch.serviceAsync(opCode, w -> req.writeBinary(w.out()), p -> {
+ public <T extends ClientMessage> CompletableFuture<T> sendRequest(int opCode, ClientMessage req, T res) {
+ return ch.serviceAsync(opCode, w -> req.writeBinary(w.out()), p -> {
res.readBinary(p.in());
return res;
- }).join();
+ });
}
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/query/JdbcClientQueryEventHandler.java b/modules/client/src/main/java/org/apache/ignite/internal/client/query/JdbcClientQueryEventHandler.java
index 74099d2..4132c91 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/query/JdbcClientQueryEventHandler.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/query/JdbcClientQueryEventHandler.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.client.query;
+import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.proto.query.JdbcQueryEventHandler;
import org.apache.ignite.client.proto.query.event.BatchExecuteRequest;
import org.apache.ignite.client.proto.query.event.BatchExecuteResult;
@@ -56,91 +57,73 @@ public class JdbcClientQueryEventHandler implements JdbcQueryEventHandler {
/** {@inheritDoc} */
@Override
- public QueryExecuteResult query(QueryExecuteRequest req) {
+ public CompletableFuture<QueryExecuteResult> query(QueryExecuteRequest req) {
QueryExecuteResult res = new QueryExecuteResult();
- client.sendRequest(ClientOp.SQL_EXEC, req, res);
-
- return res;
+ return client.sendRequest(ClientOp.SQL_EXEC, req, res);
}
/** {@inheritDoc} */
@Override
- public QueryFetchResult fetch(QueryFetchRequest req) {
+ public CompletableFuture<QueryFetchResult> fetch(QueryFetchRequest req) {
QueryFetchResult res = new QueryFetchResult();
- client.sendRequest(ClientOp.SQL_NEXT, req, res);
-
- return res;
+ return client.sendRequest(ClientOp.SQL_NEXT, req, res);
}
/** {@inheritDoc} */
@Override
- public BatchExecuteResult batch(BatchExecuteRequest req) {
+ public CompletableFuture<BatchExecuteResult> batch(BatchExecuteRequest req) {
BatchExecuteResult res = new BatchExecuteResult();
- client.sendRequest(ClientOp.SQL_EXEC_BATCH, req, res);
-
- return res;
+ return client.sendRequest(ClientOp.SQL_EXEC_BATCH, req, res);
}
/** {@inheritDoc} */
@Override
- public QueryCloseResult close(QueryCloseRequest req) {
+ public CompletableFuture<QueryCloseResult> close(QueryCloseRequest req) {
QueryCloseResult res = new QueryCloseResult();
- client.sendRequest(ClientOp.SQL_CURSOR_CLOSE, req, res);
-
- return res;
+ return client.sendRequest(ClientOp.SQL_CURSOR_CLOSE, req, res);
}
/** {@inheritDoc} */
@Override
- public JdbcMetaTablesResult tablesMeta(JdbcMetaTablesRequest req) {
+ public CompletableFuture<JdbcMetaTablesResult> tablesMeta(JdbcMetaTablesRequest req) {
JdbcMetaTablesResult res = new JdbcMetaTablesResult();
- client.sendRequest(ClientOp.SQL_TABLE_META, req, res);
-
- return res;
+ return client.sendRequest(ClientOp.SQL_TABLE_META, req, res);
}
/** {@inheritDoc} */
@Override
- public JdbcMetaColumnsResult columnsMeta(JdbcMetaColumnsRequest req) {
+ public CompletableFuture<JdbcMetaColumnsResult> columnsMeta(JdbcMetaColumnsRequest req) {
JdbcMetaColumnsResult res = new JdbcMetaColumnsResult();
- client.sendRequest(ClientOp.SQL_COLUMN_META, req, res);
-
- return res;
+ return client.sendRequest(ClientOp.SQL_COLUMN_META, req, res);
}
/** {@inheritDoc} */
@Override
- public JdbcMetaSchemasResult schemasMeta(JdbcMetaSchemasRequest req) {
+ public CompletableFuture<JdbcMetaSchemasResult> schemasMeta(JdbcMetaSchemasRequest req) {
JdbcMetaSchemasResult res = new JdbcMetaSchemasResult();
- client.sendRequest(ClientOp.SQL_SCHEMAS_META, req, res);
-
- return res;
+ return client.sendRequest(ClientOp.SQL_SCHEMAS_META, req, res);
}
/** {@inheritDoc} */
@Override
- public JdbcMetaPrimaryKeysResult primaryKeysMeta(JdbcMetaPrimaryKeysRequest req) {
+ public CompletableFuture<JdbcMetaPrimaryKeysResult> primaryKeysMeta(JdbcMetaPrimaryKeysRequest req) {
JdbcMetaPrimaryKeysResult res = new JdbcMetaPrimaryKeysResult();
- client.sendRequest(ClientOp.SQL_PK_META, req, res);
-
- return res;
+ return client.sendRequest(ClientOp.SQL_PK_META, req, res);
}
/** {@inheritDoc} */
@Override
- public JdbcMetaColumnsResult queryMetadata(JdbcQueryMetadataRequest req) {
+ public CompletableFuture<JdbcMetaColumnsResult> queryMetadata(JdbcQueryMetadataRequest req) {
JdbcMetaColumnsResult res = new JdbcMetaColumnsResult();
- client.sendRequest(ClientOp.SQL_QUERY_META, req, res);
-
- return res;
+ return client.sendRequest(ClientOp.SQL_QUERY_META, req, res);
}
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcDatabaseMetadata.java b/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcDatabaseMetadata.java
index e1c21e8..7d8f064 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcDatabaseMetadata.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcDatabaseMetadata.java
@@ -863,7 +863,7 @@ public class JdbcDatabaseMetadata implements DatabaseMetaData {
}
JdbcMetaTablesResult res
- = conn.handler().tablesMeta(new JdbcMetaTablesRequest(schemaPtrn, tblNamePtrn, tblTypes));
+ = conn.handler().tablesMeta(new JdbcMetaTablesRequest(schemaPtrn, tblNamePtrn, tblTypes)).join();
if (!res.hasResults()) {
throw IgniteQueryErrorCode.createJdbcSqlException(res.err(), res.status());
@@ -898,7 +898,7 @@ public class JdbcDatabaseMetadata implements DatabaseMetaData {
return new JdbcResultSet(Collections.emptyList(), meta);
}
- JdbcMetaSchemasResult res = conn.handler().schemasMeta(new JdbcMetaSchemasRequest(schemaPtrn));
+ JdbcMetaSchemasResult res = conn.handler().schemasMeta(new JdbcMetaSchemasRequest(schemaPtrn)).join();
if (!res.hasResults()) {
throw IgniteQueryErrorCode.createJdbcSqlException(res.err(), res.status());
@@ -971,7 +971,7 @@ public class JdbcDatabaseMetadata implements DatabaseMetaData {
return new JdbcResultSet(Collections.emptyList(), meta);
}
- JdbcMetaColumnsResult res = conn.handler().columnsMeta(new JdbcMetaColumnsRequest(schemaPtrn, tblNamePtrn, colNamePtrn));
+ JdbcMetaColumnsResult res = conn.handler().columnsMeta(new JdbcMetaColumnsRequest(schemaPtrn, tblNamePtrn, colNamePtrn)).join();
if (!res.hasResults()) {
throw IgniteQueryErrorCode.createJdbcSqlException(res.err(), res.status());
@@ -1065,7 +1065,7 @@ public class JdbcDatabaseMetadata implements DatabaseMetaData {
return new JdbcResultSet(Collections.emptyList(), meta);
}
- JdbcMetaPrimaryKeysResult res = conn.handler().primaryKeysMeta(new JdbcMetaPrimaryKeysRequest(schema, tbl));
+ JdbcMetaPrimaryKeysResult res = conn.handler().primaryKeysMeta(new JdbcMetaPrimaryKeysRequest(schema, tbl)).join();
if (!res.hasResults()) {
throw IgniteQueryErrorCode.createJdbcSqlException(res.err(), res.status());
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcResultSet.java b/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcResultSet.java
index 690a81a..51e7bc4 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcResultSet.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcResultSet.java
@@ -207,7 +207,7 @@ public class JdbcResultSet implements ResultSet {
ensureNotClosed();
if ((rowsIter == null || !rowsIter.hasNext()) && !finished) {
- QueryFetchResult res = qryHandler.fetch(new QueryFetchRequest(cursorId, fetchSize));
+ QueryFetchResult res = qryHandler.fetch(new QueryFetchRequest(cursorId, fetchSize)).join();
if (!res.hasResults()) {
throw IgniteQueryErrorCode.createJdbcSqlException(res.err(), res.status());
@@ -259,7 +259,7 @@ public class JdbcResultSet implements ResultSet {
try {
if (stmt != null && (!finished || (isQuery && !autoClose))) {
- QueryCloseResult res = qryHandler.close(new QueryCloseRequest(cursorId));
+ QueryCloseResult res = qryHandler.close(new QueryCloseRequest(cursorId)).join();
if (!res.hasResults()) {
throw IgniteQueryErrorCode.createJdbcSqlException(res.err(), res.status());
@@ -2186,7 +2186,7 @@ public class JdbcResultSet implements ResultSet {
}
if (!metaInit) {
- JdbcMetaColumnsResult res = qryHandler.queryMetadata(new JdbcQueryMetadataRequest(cursorId));
+ JdbcMetaColumnsResult res = qryHandler.queryMetadata(new JdbcQueryMetadataRequest(cursorId)).join();
meta = res.meta();
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java b/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java
index e54ba62..6f7774a 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java
@@ -130,7 +130,7 @@ public class JdbcStatement implements Statement {
QueryExecuteRequest req = new QueryExecuteRequest(schema, pageSize, maxRows, sql,
args == null ? ArrayUtils.OBJECT_EMPTY_ARRAY : args.toArray());
- QueryExecuteResult res = conn.handler().query(req);
+ QueryExecuteResult res = conn.handler().query(req).join();
if (!res.hasResults()) {
throw IgniteQueryErrorCode.createJdbcSqlException(res.err(), res.status());
@@ -547,7 +547,7 @@ public class JdbcStatement implements Statement {
BatchExecuteRequest req = new BatchExecuteRequest(conn.getSchema(), batch, conn.getAutoCommit());
try {
- BatchExecuteResult res = conn.handler().batch(req);
+ BatchExecuteResult res = conn.handler().batch(req).join();
if (!res.hasResults()) {
throw new BatchUpdateException(res.err(),
diff --git a/modules/network/src/main/java/org/apache/ignite/network/scalecube/DelegatingTransportFactory.java b/modules/network/src/main/java/org/apache/ignite/network/scalecube/DelegatingTransportFactory.java
index d621964..4695f48 100644
--- a/modules/network/src/main/java/org/apache/ignite/network/scalecube/DelegatingTransportFactory.java
+++ b/modules/network/src/main/java/org/apache/ignite/network/scalecube/DelegatingTransportFactory.java
@@ -115,6 +115,10 @@ class DelegatingTransportFactory implements TransportFactory {
.filter(resp -> resp.correlationId() != null)
.filter(resp -> resp.correlationId().equals(request.correlationId()))
.take(1)
+ .filter(message -> {
+ System.out.println("thread = " + Thread.currentThread().getName());
+ return true;
+ })
.subscribe(sink::success, sink::error, sink::success);
// cancel the nested subscription if the client cancels the outer Mono
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index a7b7232..cd89b22 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -998,33 +998,34 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
* @return Future representing pending completion of the operation.
*/
private CompletableFuture<List<Table>> tablesAsyncInternal() {
- List<IgniteUuid> tableIds = directTableIds();
+ return CompletableFuture.supplyAsync(this::directTableIds)
+ .thenCompose(tableIds -> {
+ var tableFuts = new CompletableFuture[tableIds.size()];
- var tableFuts = new CompletableFuture[tableIds.size()];
+ var i = 0;
- var i = 0;
-
- for (IgniteUuid tblId : tableIds) {
- tableFuts[i++] = tableAsyncInternal(tblId, false);
- }
+ for (IgniteUuid tblId : tableIds) {
+ tableFuts[i++] = tableAsyncInternal(tblId, false);
+ }
- return CompletableFuture.allOf(tableFuts).thenApply(unused -> {
- var tables = new ArrayList<Table>(tableIds.size());
+ return CompletableFuture.allOf(tableFuts).thenApply(unused -> {
+ var tables = new ArrayList<Table>(tableIds.size());
- try {
- for (var fut : tableFuts) {
- var table = fut.get();
+ try {
+ for (var fut : tableFuts) {
+ var table = fut.get();
- if (table != null) {
- tables.add((Table) table);
- }
- }
- } catch (Throwable t) {
- throw new CompletionException(t);
- }
+ if (table != null) {
+ tables.add((Table) table);
+ }
+ }
+ } catch (Throwable t) {
+ throw new CompletionException(t);
+ }
- return tables;
- });
+ return tables;
+ });
+ });
}
/**