You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by am...@apache.org on 2021/08/02 12:06:03 UTC
[ignite-3] 02/05: WIP.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-15212
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 5797ee37dc64d45e103678848c68d1f1d81769ad
Author: Andrew Mashenkov <an...@gmail.com>
AuthorDate: Wed Jul 28 14:27:11 2021 +0300
WIP.
---
.../main/java/org/apache/ignite/app/Ignite.java | 8 +
.../org/apache/ignite/query/sql/IgniteSql.java | 109 ++++++++++++
.../Ignite.java => query/sql/SQLException.java} | 36 ++--
.../org/apache/ignite/query/sql/SqlColumnMeta.java | 60 +++++++
.../Ignite.java => query/sql/SqlResultSet.java} | 25 +--
.../apache/ignite/query/sql/SqlResultSetMeta.java | 63 +++++++
.../{app/Ignite.java => query/sql/SqlRow.java} | 26 +--
.../sql/reactive/ReactiveSqlResultSet.java} | 30 ++--
.../org/apache/ignite/internal/app/IgniteImpl.java | 12 +-
modules/sql/src/test/java/SqlTest.java | 194 +++++++++++++++++++++
modules/sql/src/test/java/TestRow.java | 181 +++++++++++++++++++
11 files changed, 677 insertions(+), 67 deletions(-)
diff --git a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java b/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
index 87e585a..aa4bd46 100644
--- a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
+++ b/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
@@ -17,6 +17,7 @@
package org.apache.ignite.app;
+import org.apache.ignite.query.sql.IgniteSql;
import org.apache.ignite.table.manager.IgniteTables;
import org.apache.ignite.tx.IgniteTransactions;
@@ -37,4 +38,11 @@ public interface Ignite extends AutoCloseable {
* @return Ignite transactions.
*/
IgniteTransactions transactions();
+
+ /**
+ * Returns a facade for SQL quering.
+ *
+ * @return Ignite SQL facade.
+ */
+ IgniteSql sql();
}
diff --git a/modules/api/src/main/java/org/apache/ignite/query/sql/IgniteSql.java b/modules/api/src/main/java/org/apache/ignite/query/sql/IgniteSql.java
new file mode 100644
index 0000000..a44ab14
--- /dev/null
+++ b/modules/api/src/main/java/org/apache/ignite/query/sql/IgniteSql.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.query.sql;
+
+import java.util.Collection;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.query.sql.reactive.ReactiveSqlResultSet;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Ignite SQL query facade.
+ */
+// TODO: Do we wand a separate IgniteQuery facade for non-sql (index/scan/full-text) queries?
+public interface IgniteSql {
+ ////////////// Query execution.
+ /**
+ * Shortcut method for running a query.
+ *
+ * @param sql SQL query template.
+ * @param args Arguments for template (optional)
+ * @return SQL query resultset.
+ * @throws SQLException If failed.
+ */
+ SqlResultSet execute(@NotNull String sql, @Nullable Object... args);
+
+ /**
+ * Shortcut method for running a query in asynchronously.
+ *
+ * @param sql SQL query template.
+ * @param args Arguments for template (optional)
+ * @return Query future.
+ * @throws SQLException If failed.
+ */
+ CompletableFuture<SqlResultSet> executeAsync(String sql, Object... args);
+ //TODO: May fut.cancel() cancels a query? If so, describe behavior in Javadoc.
+ //TODO: Cassandra API offers pagination API here, for manual fetching control. Is their AsyncResultSet ever usefull?
+
+ /**
+ * Shortcut method for running a query in asynchronously.
+ *
+ * @param sql SQL query template.
+ * @param args Arguments for template (optional)
+ * @return Reactive result.
+ * @throws SQLException If failed.
+ */
+ ReactiveSqlResultSet executeReactive(String sql, Object... args);
+
+ //TODO: Do we need a separate methods for DML/DDL that do not return the resultset? or extend a resultset?
+ //TODO: same methods for Statement.
+
+ ///////////////// Query monitoring and management.
+ /**
+ * Returns SQL views.
+ */
+ Object views(); //TODO: To be described.
+ //TODO: Add view for runnin queries.
+
+ /**
+ * Return info of the queries. //TODO: "running on the node locally"? or "started on the node"? or both?
+ *
+ * @return Running queries infos.
+ */
+ Collection<Object> runningQueries(); //TODO: Is it needed here or to be moved into the Views facade?
+
+ /**
+ * Kills query by its' id.
+ *
+ * @param queryID Query id.
+ */
+ void killQuery(UUID queryID);
+
+
+ ////////////// Statistics management.
+ Collection<Object> tableStatistics(String table); //TODO: Local or global? Ready or in-progress? TBD.
+
+ CompletableFuture<Void> gatherStatistics(String table, Object statisticsConfiguration);
+ //TODO: Creates new statistics in addition, or drop old and replaces with new?
+ //TODO: Should the existed one be refreshed?
+
+ CompletableFuture<Void> refreshStatistics(String table, @Nullable String... optionalStatisticName);
+
+ void dropStatistics(String table, @Nullable String... optionalStatisticName);
+
+ void refreshLocalStatistics(String table, @Nullable String... optionalStatisticName); //TODO: Actually, drops local statistics to be automatically refreshed.
+
+
+ //////////////////// DDL
+
+ //TODO: Do we expect any SQL DDL pragrammatical API here? Why we have tables().create(), but not here?
+ //TODO: alterTable()?
+}
+
diff --git a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java b/modules/api/src/main/java/org/apache/ignite/query/sql/SQLException.java
similarity index 60%
copy from modules/api/src/main/java/org/apache/ignite/app/Ignite.java
copy to modules/api/src/main/java/org/apache/ignite/query/sql/SQLException.java
index 87e585a..3e33d6e 100644
--- a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
+++ b/modules/api/src/main/java/org/apache/ignite/query/sql/SQLException.java
@@ -15,26 +15,28 @@
* limitations under the License.
*/
-package org.apache.ignite.app;
+package org.apache.ignite.query.sql;
-import org.apache.ignite.table.manager.IgniteTables;
-import org.apache.ignite.tx.IgniteTransactions;
+import org.apache.ignite.lang.IgniteException;
+import org.jetbrains.annotations.Nullable;
/**
- * Ignite node interface. Main entry-point for all Ignite APIs.
+ * SQL exception base class.
*/
-public interface Ignite extends AutoCloseable {
- /**
- * Gets an object for manipulate Ignite tables.
- *
- * @return Ignite tables.
- */
- IgniteTables tables();
+//TODO: Do we want to use this instead of java.sql.SQLException ?
+public class SQLException extends IgniteException {
+ public SQLException() {
+ }
- /**
- * Returns a transaction facade.
- *
- * @return Ignite transactions.
- */
- IgniteTransactions transactions();
+ public SQLException(String msg) {
+ super(msg);
+ }
+
+ public SQLException(Throwable cause) {
+ super(cause);
+ }
+
+ public SQLException(String msg, @Nullable Throwable cause) {
+ super(msg, cause);
+ }
}
diff --git a/modules/api/src/main/java/org/apache/ignite/query/sql/SqlColumnMeta.java b/modules/api/src/main/java/org/apache/ignite/query/sql/SqlColumnMeta.java
new file mode 100644
index 0000000..856c82f
--- /dev/null
+++ b/modules/api/src/main/java/org/apache/ignite/query/sql/SqlColumnMeta.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.query.sql;
+
+import org.apache.ignite.schema.ColumnType;
+
+/**
+ * Column metadata.
+ */
+public interface SqlColumnMeta {
+ /**
+ * Return column name in resultset.
+ *
+ * Note: If row column doess not represents any table column, then generated name will be used.
+ *
+ * @return Column name.
+ */
+ String name();
+ //TODO: do we expect a CanonicalName here?
+ //TODO: do we want a Table name for real column? Is Calcite supports this?
+
+ /**
+ * Returns column type.
+ *
+ * @return Column type.
+ */
+ ColumnType type();
+ //TODO: do ever we want to expose ColumnType (NativeType) here? Is it useful or drop this?
+
+ /**
+ * Returns column value type.
+ *
+ * @return Value type.
+ */
+ Class<?> valueClass();
+ //TODO: this maybe more useful in contraty to the previous one.
+
+ /**
+ * Row column nullability.
+ *
+ * @return {@code true} if column is nullable, {@code false} otherwise.
+ */
+ boolean nullable();
+ //TODO: AFAIK, Calcite can derive column type for us.
+}
diff --git a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java b/modules/api/src/main/java/org/apache/ignite/query/sql/SqlResultSet.java
similarity index 63%
copy from modules/api/src/main/java/org/apache/ignite/app/Ignite.java
copy to modules/api/src/main/java/org/apache/ignite/query/sql/SqlResultSet.java
index 87e585a..174c914 100644
--- a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
+++ b/modules/api/src/main/java/org/apache/ignite/query/sql/SqlResultSet.java
@@ -15,26 +15,19 @@
* limitations under the License.
*/
-package org.apache.ignite.app;
-
-import org.apache.ignite.table.manager.IgniteTables;
-import org.apache.ignite.tx.IgniteTransactions;
+package org.apache.ignite.query.sql;
/**
- * Ignite node interface. Main entry-point for all Ignite APIs.
+ * SQL result set provides methods to access SQL query resul represented as collection of {@link SqlRow}.
+ *
+ * All the rows in result set have the same structure described in {@link SqlResultSetMeta}.
+ * ResultSet must be closed after usage to free resources.
*/
-public interface Ignite extends AutoCloseable {
- /**
- * Gets an object for manipulate Ignite tables.
- *
- * @return Ignite tables.
- */
- IgniteTables tables();
-
+public interface SqlResultSet extends Iterable<SqlRow>, AutoCloseable {
/**
- * Returns a transaction facade.
+ * Returns metadata for the results.
*
- * @return Ignite transactions.
+ * @return ResultSet metadata.
*/
- IgniteTransactions transactions();
+ SqlResultSetMeta metadata();
}
diff --git a/modules/api/src/main/java/org/apache/ignite/query/sql/SqlResultSetMeta.java b/modules/api/src/main/java/org/apache/ignite/query/sql/SqlResultSetMeta.java
new file mode 100644
index 0000000..38c4a63
--- /dev/null
+++ b/modules/api/src/main/java/org/apache/ignite/query/sql/SqlResultSetMeta.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.query.sql;
+
+import java.util.List;
+
+/**
+ * ResultSet metadata.
+ */
+public interface SqlResultSetMeta {
+ /**
+ * Returns number of column that every row in a ResulSet contains.
+ *
+ * @return Number of columns.
+ */
+ int columnsCount();
+
+ /**
+ * Returns metadata with description for every column in resultset.
+ *
+ * @return Columns metadata.
+ */
+ List<SqlColumnMeta> columns();
+
+ /**
+ * Returns metadata for the requested column.
+ *
+ * @param columnId Column Id.
+ * @return Column metadata.
+ */
+ SqlColumnMeta column(int columnId);
+
+ /**
+ * Returns metadata for the requested column.
+ *
+ * @param columnName Column name.
+ * @return Column metadata.
+ */
+ SqlColumnMeta column(String columnName);
+
+ /**
+ * Returns column id in ResultSet by column name.
+ *
+ * @param columnName Columns name which id is resoliving.
+ * @return Column Id for requested column.
+ */
+ int indexOf(String columnName);
+}
diff --git a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java b/modules/api/src/main/java/org/apache/ignite/query/sql/SqlRow.java
similarity index 61%
copy from modules/api/src/main/java/org/apache/ignite/app/Ignite.java
copy to modules/api/src/main/java/org/apache/ignite/query/sql/SqlRow.java
index 87e585a..1938186 100644
--- a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
+++ b/modules/api/src/main/java/org/apache/ignite/query/sql/SqlRow.java
@@ -15,26 +15,16 @@
* limitations under the License.
*/
-package org.apache.ignite.app;
+package org.apache.ignite.query.sql;
-import org.apache.ignite.table.manager.IgniteTables;
-import org.apache.ignite.tx.IgniteTransactions;
+import org.apache.ignite.table.Tuple;
/**
- * Ignite node interface. Main entry-point for all Ignite APIs.
+ * SQL row provides methods to access row data by column name or id.
+ *
+ * Column description can be retrived from {@link SqlResultSet#metadata()}.
+ * @see SqlColumnMeta
*/
-public interface Ignite extends AutoCloseable {
- /**
- * Gets an object for manipulate Ignite tables.
- *
- * @return Ignite tables.
- */
- IgniteTables tables();
-
- /**
- * Returns a transaction facade.
- *
- * @return Ignite transactions.
- */
- IgniteTransactions transactions();
+public interface SqlRow extends Tuple {
+//TODO: We extends Tuple here just for short. Do we need smth else?
}
diff --git a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java b/modules/api/src/main/java/org/apache/ignite/query/sql/reactive/ReactiveSqlResultSet.java
similarity index 55%
copy from modules/api/src/main/java/org/apache/ignite/app/Ignite.java
copy to modules/api/src/main/java/org/apache/ignite/query/sql/reactive/ReactiveSqlResultSet.java
index 87e585a..fbb66e9 100644
--- a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
+++ b/modules/api/src/main/java/org/apache/ignite/query/sql/reactive/ReactiveSqlResultSet.java
@@ -15,26 +15,26 @@
* limitations under the License.
*/
-package org.apache.ignite.app;
+package org.apache.ignite.query.sql.reactive;
-import org.apache.ignite.table.manager.IgniteTables;
-import org.apache.ignite.tx.IgniteTransactions;
+import java.sql.ResultSetMetaData;
+import java.util.concurrent.Flow;
+import org.apache.ignite.query.sql.SqlRow;
/**
- * Ignite node interface. Main entry-point for all Ignite APIs.
+ * Reactive resultset provides methods to subscribe to the query results in reactive way.
+ *
+ * Note: It implies to be used with the reactive framework such as Reactor or R2DBC.
+ *
+ * @see reactor.core.publisher.Flux
+ * @see r2dbc
*/
-public interface Ignite extends AutoCloseable {
- /**
- * Gets an object for manipulate Ignite tables.
- *
- * @return Ignite tables.
- */
- IgniteTables tables();
-
+// TODO: add links to frameworks, code examples. See SqlTest.SqlRowSubsvriber.
+public interface ReactiveSqlResultSet extends Flow.Publisher<SqlRow> {
/**
- * Returns a transaction facade.
+ * Return publisher for the ResulSets' metadata.
*
- * @return Ignite transactions.
+ * @return Metadata publisher.
*/
- IgniteTransactions transactions();
+ Flow.Publisher<ResultSetMetaData> metadata();
}
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index 54a93b9..db88368 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.app;
import org.apache.ignite.app.Ignite;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.processors.query.calcite.SqlQueryProcessor;
+import org.apache.ignite.query.sql.IgniteSql;
import org.apache.ignite.table.manager.IgniteTables;
import org.apache.ignite.tx.IgniteTransactions;
@@ -30,9 +31,10 @@ public class IgniteImpl implements Ignite {
/** Distributed table manager. */
private final IgniteTables distributedTableManager;
- /** Vault manager */
+ /** Vault manager. */
private final VaultManager vaultManager;
+ /** Query manager. */
private final SqlQueryProcessor qryEngine;
/**
@@ -44,6 +46,8 @@ public class IgniteImpl implements Ignite {
this.distributedTableManager = tableManager;
this.vaultManager = vaultManager;
this.qryEngine = qryEngine;
+
+ // qryManager = new IgniteQueriesImpl(qryEngine);
}
/** {@inheritDoc} */
@@ -51,6 +55,7 @@ public class IgniteImpl implements Ignite {
return distributedTableManager;
}
+ // TODO: To be replaced with IgniteImpl#sql().
public SqlQueryProcessor queryEngine() {
return qryEngine;
}
@@ -61,6 +66,11 @@ public class IgniteImpl implements Ignite {
}
/** {@inheritDoc} */
+ @Override public IgniteSql sql() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
@Override public void close() throws Exception {
vaultManager.close();
}
diff --git a/modules/sql/src/test/java/SqlTest.java b/modules/sql/src/test/java/SqlTest.java
new file mode 100644
index 0000000..c49721a
--- /dev/null
+++ b/modules/sql/src/test/java/SqlTest.java
@@ -0,0 +1,194 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow;
+import java.util.function.Consumer;
+import org.apache.ignite.query.sql.IgniteSql;
+import org.apache.ignite.query.sql.SqlResultSet;
+import org.apache.ignite.query.sql.SqlResultSetMeta;
+import org.apache.ignite.query.sql.SqlRow;
+import org.apache.ignite.query.sql.reactive.ReactiveSqlResultSet;
+import org.apache.ignite.schema.ColumnType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Disabled // TODO: create a ticket to fix this.
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public class SqlTest {
+ @Mock
+ IgniteSql queryMgr;
+
+ @BeforeEach
+ void setUp() {
+ initMock();
+ }
+
+ @Test
+ public void testSynchronousSql() {
+ SqlResultSet rs = queryMgr.execute("SELECT id, val FROM table WHERE id < {} AND val LIKE {};", 10, "str%");
+
+ for (SqlRow r : rs) {
+ assertTrue(10 > r.longValue("id"));
+ assertTrue((r.stringValue("val")).startsWith("str"));
+ }
+ }
+
+ @Test
+ public void testAsyncSql() {
+ queryMgr.executeAsync("SELECT id, val FROM table WHERE id == {};", 10)
+ .thenCompose(rs -> {
+ String str = rs.iterator().next().stringValue("val");
+
+ return queryMgr.executeAsync("SELECT val FROM table where val LIKE {};", str);
+ })
+ .join();
+ }
+
+ @Test
+ public void testReactiveSql() {
+ SqlRowSubscriber subscriber = new SqlRowSubscriber(row -> {
+ assertTrue(10 > row.longValue("id"));
+ assertTrue(row.stringValue("val").startsWith("str"));
+ });
+
+ queryMgr.executeReactive("SELECT id, val FROM table WHERE id < {} AND val LIKE {};", 10, "str%")
+ .subscribe(subscriber);
+
+ subscriber.join();
+ }
+
+ @Disabled
+ @Test
+ public void testMetadata() {
+ SqlResultSet rs = queryMgr.execute("SELECT id, val FROM table WHERE id < {} AND val LIKE {}; ", 10, "str%");
+
+ SqlRow row = rs.iterator().next();
+
+ SqlResultSetMeta meta = rs.metadata();
+
+ assertEquals(rs.metadata().columnsCount(), row.columnCount());
+
+ assertEquals(0, meta.indexOf("id"));
+ assertEquals(1, meta.indexOf("val"));
+
+ assertEquals("id", meta.column(0).name());
+ assertEquals("val", meta.column(1).name());
+
+ assertEquals(ColumnType.INT64, meta.column(0).type());
+ assertEquals(ColumnType.string(), meta.column(1).type());
+
+ assertFalse(meta.column(0).nullable());
+ assertTrue(meta.column(1).nullable());
+ }
+
+ private void initMock() {
+ Mockito.when(queryMgr.execute(Mockito.eq("SELECT id, val FROM table WHERE id < {} AND val LIKE {};"), Mockito.any())).
+ thenAnswer(ans -> Mockito.when(Mockito.mock(SqlResultSet.class).iterator())
+ .thenReturn(List.of(
+ new TestRow().set("id", 1L).set("val", "string 1").build(),
+ new TestRow().set("id", 2L).set("val", "string 2").build(),
+ new TestRow().set("id", 5L).set("val", "string 3").build()
+ ).iterator()).getMock());
+
+ Mockito.when(queryMgr.executeAsync(Mockito.eq("SELECT id, val FROM table WHERE id == {};"), Mockito.any()))
+ .thenAnswer(ans -> {
+ Object mock = Mockito.when(Mockito.mock(SqlResultSet.class).iterator())
+ .thenReturn(List.of(new TestRow().set("id", 1L).set("val", "string 1").build()).iterator())
+ .getMock();
+
+ return CompletableFuture.completedFuture(mock);
+ });
+
+ Mockito.when(queryMgr.executeAsync(Mockito.eq("SELECT val FROM table where val LIKE {};"), Mockito.any()))
+ .thenAnswer(ans -> {
+ Object mock = Mockito.when(Mockito.mock(SqlResultSet.class).iterator())
+ .thenReturn(List.of(new TestRow().set("id", 10L).set("val", "string 10").build()).iterator())
+ .getMock();
+
+ return CompletableFuture.completedFuture(mock);
+ });
+
+ Mockito.when(queryMgr.executeReactive(Mockito.startsWith("SELECT id, val FROM table WHERE id < {} AND val LIKE {};"), Mockito.any()))
+ .thenAnswer(invocation -> {
+ ReactiveSqlResultSet mock = Mockito.mock(ReactiveSqlResultSet.class);
+
+ Mockito.doAnswer(ans -> {
+ Flow.Subscriber subscrber = ans.getArgument(0);
+
+ subscrber.onSubscribe(Mockito.mock(Flow.Subscription.class));
+
+ List.of(
+ new TestRow().set("id", 1L).set("val", "string 1").build(),
+ new TestRow().set("id", 2L).set("val", "string 2").build(),
+ new TestRow().set("id", 5L).set("val", "string 3").build()
+ ).forEach(i -> subscrber.onNext(i));
+
+ subscrber.onComplete();
+
+ return ans;
+ }).when(mock).subscribe(Mockito.any(Flow.Subscriber.class));
+
+ return mock;
+ });
+ }
+
+ /**
+ * Dummy subsctiber for test purposes.
+ */
+ static class SqlRowSubscriber extends CompletableFuture implements Flow.Subscriber<SqlRow> {
+ private Consumer<SqlRow> rowConsumer;
+
+ SqlRowSubscriber(Consumer<SqlRow> rowConsumer) {
+ this.rowConsumer = rowConsumer;
+ }
+
+ @Override public void onSubscribe(Flow.Subscription subscription) {
+ whenCompleteAsync((res, th) -> {
+ if (th != null)
+ subscription.cancel();
+ });
+
+ subscription.request(Long.MAX_VALUE); // Unbounded.
+ }
+
+ @Override public void onNext(SqlRow row) {
+ rowConsumer.accept(row);
+ }
+
+ @Override public void onError(Throwable throwable) {
+ completeExceptionally(throwable);
+ }
+
+ @Override public void onComplete() {
+ complete(null);
+ }
+ }
+}
diff --git a/modules/sql/src/test/java/TestRow.java b/modules/sql/src/test/java/TestRow.java
new file mode 100644
index 0000000..2105be9
--- /dev/null
+++ b/modules/sql/src/test/java/TestRow.java
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import java.util.BitSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.binary.BinaryObjects;
+import org.apache.ignite.query.sql.SqlRow;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Dummy table storage implementation.
+ */
+public class TestRow implements SqlRow {
+ /** Columns values. */
+ private final Map<String, Object> map = new HashMap<>();
+
+ public TestRow set(String columnName, Object value) {
+ map.put(columnName, value);
+
+ return this;
+ }
+
+ public SqlRow build() {
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> T valueOrDefault(String columnName, T def) {
+ return (T)map.getOrDefault(columnName, def);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> T value(String columnName) {
+ return (T)map.get(columnName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> T value(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int columnCount() {
+ return map.size();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String columnName(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Integer columnIndex(String columnName) {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public BinaryObject binaryObjectValue(String columnName) {
+ byte[] data = value(columnName);
+
+ return BinaryObjects.wrap(data);
+ }
+
+ /** {@inheritDoc} */
+ @Override public BinaryObject binaryObjectValue(int columnIndex) {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte byteValue(String columnName) {
+ return value(columnName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte byteValue(int columnIndex) {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short shortValue(String columnName) {
+ return value(columnName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short shortValue(int columnIndex) {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int intValue(String columnName) {
+ return value(columnName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int intValue(int columnIndex) {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long longValue(String columnName) {
+ return value(columnName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public long longValue(int columnIndex) {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public float floatValue(String columnName) {
+ return value(columnName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public float floatValue(int columnIndex) {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public double doubleValue(String columnName) {
+ return value(columnName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public double doubleValue(int columnIndex) {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String stringValue(String columnName) {
+ return value(columnName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String stringValue(int columnIndex) {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public UUID uuidValue(String columnName) {
+ return value(columnName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public UUID uuidValue(int columnIndex) {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public BitSet bitmaskValue(String columnName) {
+ return value(columnName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public BitSet bitmaskValue(int columnIndex) {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @NotNull @Override public Iterator<Object> iterator() {
+ throw new UnsupportedOperationException();
+ }
+}