tkalkirill commented on code in PR #2518:
URL: https://github.com/apache/ignite-3/pull/2518#discussion_r1313815674
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -2734,6 +2709,19 @@ private PartitionUpdateHandlers
createPartitionUpdateHandlers(
return startedTables.get(tableId);
}
+ /**
+ * Returns a table instance if it exists, {@code null} otherwise.
+ *
+ * @param tableName Table name.
+ */
+ public @Nullable TableImpl getTable(String tableName) {
Review Comment:
Please add `@TestOnly`
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -2017,52 +1990,81 @@ public CompletableFuture<TableImpl>
tableImplAsync(String name) {
/**
* Gets a table by name, if it was created before. Doesn't parse canonical
name.
*
- * @param name Table name.
+ * @param tableName Table name.
* @return Future representing pending completion of the {@code
TableManager#tableAsyncInternal} operation.
*/
- public CompletableFuture<TableImpl> tableAsyncInternal(String name) {
+ public CompletableFuture<TableImpl> tableAsyncInternal(String tableName) {
if (!busyLock.enterBusy()) {
throw new IgniteException(new NodeStoppingException());
Review Comment:
Again =)
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -1864,38 +1848,27 @@ private CompletableFuture<List<Table>>
tablesAsyncInternal() {
}
/**
- * Collects a list of direct table ids.
+ * Return actual table id by given name or {@code null} if table doesn't
exist.
*
- * @return A list of direct table ids.
+ * @param tableName Table name.
*/
- private List<Integer> directTableIds() {
- return configuredTablesCache.configuredTableIds();
- }
-
- /**
- * Gets direct id of table with {@code tblName}.
- *
- * @param tblName Name of the table.
- * @return Direct id of the table, or {@code null} if the table with the
{@code tblName} has not been found.
- */
- @Nullable
- private Integer directTableId(String tblName) {
+ private @Nullable Integer tableNameToId(String tableName) {
try {
- TableConfiguration exTblCfg =
directProxy(tablesCfg.tables()).get(tblName);
+ TableImpl table =
findTableImplByName(tablesByIdVv.latest().values(), tableName);
- if (exTblCfg == null) {
+ if (table == null) {
return null;
} else {
- return exTblCfg.id().value();
+ return table.tableId();
}
} catch (NoSuchElementException e) {
Review Comment:
Can this exception happen now?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -2017,52 +1993,81 @@ public CompletableFuture<TableImpl>
tableImplAsync(String name) {
/**
* Gets a table by name, if it was created before. Doesn't parse canonical
name.
*
- * @param name Table name.
+ * @param tableName Table name.
* @return Future representing pending completion of the {@code
TableManager#tableAsyncInternal} operation.
*/
- public CompletableFuture<TableImpl> tableAsyncInternal(String name) {
+ public CompletableFuture<TableImpl> tableAsyncInternal(String tableName) {
if (!busyLock.enterBusy()) {
throw new IgniteException(new NodeStoppingException());
}
try {
- return supplyAsync(() -> inBusyLock(busyLock, () ->
directTableId(name)), ioExecutor)
- .thenCompose(tableId -> inBusyLock(busyLock, () -> {
+ HybridTimestamp now = clock.now();
+
+ return schemaSyncService.waitForMetadataCompleteness(now)
+ .thenComposeAsync(ignore -> {
+ Integer tableId = tableNameToId(tableName);
+
if (tableId == null) {
+ // Table isn't configured.
return completedFuture(null);
}
- return tableAsyncInternal(tableId, false);
- }));
+ TableImpl table = latestTablesById().get(tableId);
+
+ if (table != null) {
+ // Table was published.
+ return completedFuture(table);
+ }
+
+ // Wait for table initialization.
+ return tableReadyFuture(tableId);
+ }, ioExecutor);
} finally {
busyLock.leaveBusy();
}
}
/**
- * Internal method for getting table by id.
+ * Gets a table by id, if it was created before.
*
- * @param id Table id.
- * @param checkConfiguration {@code True} when the method checks a
configuration before trying to get a table, {@code false}
- * otherwise.
- * @return Future representing pending completion of the operation.
+ * @param tableId Table id.
+ * @return Future representing pending completion of the {@code
TableManager#tableAsyncInternal} operation.
*/
- public CompletableFuture<TableImpl> tableAsyncInternal(int id, boolean
checkConfiguration) {
- CompletableFuture<Boolean> tblCfgFut = checkConfiguration
- ? supplyAsync(() -> inBusyLock(busyLock, () ->
isTableConfigured(id)), ioExecutor)
- : completedFuture(true);
+ private CompletableFuture<TableImpl> tableAsyncInternal(int tableId) {
+ HybridTimestamp now = clock.now();
- return tblCfgFut.thenCompose(isCfg -> inBusyLock(busyLock, () -> {
- if (!isCfg) {
- return completedFuture(null);
- }
+ return schemaSyncService.waitForMetadataCompleteness(now)
+ .thenComposeAsync(ignore -> {
+ TableImpl table = latestTablesById().get(tableId);
- TableImpl tbl = latestTablesById().get(id);
+ if (table != null) {
+ // Table was published.
+ return completedFuture(table);
+ }
- if (tbl != null) {
- return completedFuture(tbl);
- }
+ if (tablesCfg.tables().value().stream().noneMatch(tbl ->
tbl.id() == tableId)) {
+ // Table isn't configured.
+ return completedFuture(null);
+ }
+
+ // Wait for table initialization.
+ return tableReadyFuture(tableId);
+ }, ioExecutor);
+ }
+ /**
+ * Internal method for getting a table future, which is completed when the
table will be published.
+ *
+ * @param id Table id.
+ * @return Future representing pending completion of the operation.
+ */
+ private CompletableFuture<TableImpl> tableReadyFuture(int id) {
+ if (!busyLock.enterBusy()) {
Review Comment:
You don't fix it (
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -1978,7 +1951,7 @@ public CompletableFuture<TableImpl> tableAsync(int id) {
throw new IgniteException(new NodeStoppingException());
Review Comment:
Again =)
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -1837,18 +1819,20 @@ public CompletableFuture<List<Table>> tablesAsync() {
* @return Future representing pending completion of the operation.
*/
private CompletableFuture<List<Table>> tablesAsyncInternal() {
- return supplyAsync(() -> inBusyLock(busyLock, this::directTableIds),
ioExecutor)
- .thenCompose(tableIds -> inBusyLock(busyLock, () -> {
- var tableFuts = new CompletableFuture[tableIds.size()];
+ return schemaSyncService.waitForMetadataCompleteness(clock.now())
+ .thenCompose(ignore -> inBusyLock(busyLock, () -> {
+ Collection<TableImpl> configuredTables =
tablesByIdVv.latest().values();
+
+ var tableFuts = new
CompletableFuture[configuredTables.size()];
var i = 0;
- for (int tblId : tableIds) {
- tableFuts[i++] = tableAsyncInternal(tblId, false);
+ for (TableImpl tbl : configuredTables) {
+ tableFuts[i++] = tableReadyFuture(tbl.tableId());
}
return allOf(tableFuts).thenApply(unused ->
inBusyLock(busyLock, () -> {
- var tables = new ArrayList<Table>(tableIds.size());
+ var tables = new
ArrayList<Table>(configuredTables.size());
for (var fut : tableFuts) {
var table = fut.join();
Review Comment:
Below code table can be null?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -1837,18 +1819,20 @@ public CompletableFuture<List<Table>> tablesAsync() {
* @return Future representing pending completion of the operation.
*/
private CompletableFuture<List<Table>> tablesAsyncInternal() {
- return supplyAsync(() -> inBusyLock(busyLock, this::directTableIds),
ioExecutor)
- .thenCompose(tableIds -> inBusyLock(busyLock, () -> {
- var tableFuts = new CompletableFuture[tableIds.size()];
+ return schemaSyncService.waitForMetadataCompleteness(clock.now())
+ .thenCompose(ignore -> inBusyLock(busyLock, () -> {
Review Comment:
thenComposeAsync is better
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
