rpuch commented on code in PR #2254:
URL: https://github.com/apache/ignite-3/pull/2254#discussion_r1279296488
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -2718,4 +2752,39 @@ private PartitionUpdateHandlers
createPartitionUpdateHandlers(
private CatalogZoneDescriptor getZoneDescriptor(int id) {
return toZoneDescriptor(getZoneById(zonesConfig, id).value());
}
+
+ private static @Nullable TableImpl
findTableImplByName(Collection<TableImpl> tables, String name) {
+ return tables.stream().filter(table ->
table.name().equals(name)).findAny().orElse(null);
+ }
+
+ /**
+ * Fires table creation events so that indexes can be correctly created at
IndexManager startup.
+ *
+ * <p>NOTE: This is a temporary solution that must be get
rid/remake/change.
+ */
+ // TODO: IGNITE-19499 Need to get rid/remake/change
+ private void fireCreateTablesOnStartManager() {
Review Comment:
```suggestion
private void fireCreateTablesOnManagerStart() {
```
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -96,414 +86,180 @@ public class IndexManager extends Producer<IndexEvent,
IndexEventParameters> imp
/** Table manager. */
private final TableManager tableManager;
+ /** Catalog manager. */
+ private final CatalogManager catalogManager;
+
+ /** Meta storage manager. */
+ private final MetaStorageManager metaStorageManager;
+
/** Busy lock to stop synchronously. */
private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
/** Prevents double stopping of the component. */
private final AtomicBoolean stopGuard = new AtomicBoolean();
+ /** Versioned value used only at the start of the manager. */
+ private final IncrementalVersionedValue<Void> startVv;
+
/**
* Constructor.
*
* @param tablesCfg Tables and indexes configuration.
* @param schemaManager Schema manager.
* @param tableManager Table manager.
+ * @param catalogManager Catalog manager.
*/
public IndexManager(
TablesConfiguration tablesCfg,
SchemaManager schemaManager,
- TableManager tableManager
+ TableManager tableManager,
+ CatalogManager catalogManager,
+ MetaStorageManager metaStorageManager,
+ Consumer<LongFunction<CompletableFuture<?>>> registry
) {
this.tablesCfg = Objects.requireNonNull(tablesCfg, "tablesCfg");
this.schemaManager = Objects.requireNonNull(schemaManager,
"schemaManager");
this.tableManager = tableManager;
+ this.catalogManager = catalogManager;
+ this.metaStorageManager = metaStorageManager;
+
+ startVv = new IncrementalVersionedValue<>(registry);
}
- /** {@inheritDoc} */
@Override
public void start() {
- LOG.debug("Index manager is about to start");
-
- tablesCfg.indexes().listenElements(new ConfigurationListener());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Index manager is about to start");
+ }
- tableManager.listen(TableEvent.CREATE, (param, ex) -> {
- if (ex != null) {
- return completedFuture(false);
- }
+ startIndexes();
- // We can't return this future as the listener's result, because a
deadlock can happen in the configuration component:
- // this listener is called inside a configuration notification
thread and all notifications are required to finish before
- // new configuration modifications can occur (i.e. we are creating
an index below). Therefore we create the index fully
- // asynchronously and rely on the underlying components to handle
PK index synchronisation.
- tableManager.tableAsync(param.causalityToken(), param.tableId())
- .thenCompose(table -> {
- String[] pkColumns =
Arrays.stream(table.schemaView().schema().keyColumns().columns())
- .map(Column::name)
- .toArray(String[]::new);
+ catalogManager.listen(INDEX_CREATE, (parameters, exception) -> {
+ assert exception == null : parameters;
- String pkName = table.name() + "_PK";
+ return onIndexCreate((CreateIndexEventParameters) parameters);
+ });
- return createIndexAsync("PUBLIC", pkName,
table.name(), false,
- change ->
change.changeUniq(true).convert(HashIndexChange.class)
- .changeColumnNames(pkColumns)
- );
- })
- .whenComplete((v, e) -> {
- if (e != null) {
- LOG.error("Error when creating index: " + e);
- }
- });
+ catalogManager.listen(INDEX_DROP, (parameters, exception) -> {
+ assert exception == null;
- return completedFuture(false);
+ return onIndexDrop((DropIndexEventParameters) parameters);
});
- LOG.info("Index manager started");
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Index manager started");
+ }
}
- /** {@inheritDoc} */
@Override
public void stop() throws Exception {
- LOG.debug("Index manager is about to stop");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Index manager is about to stop");
+ }
Review Comment:
This is a `stop()` method that is called once per JVM. It makes no sense to
save one object here, but a line of code might make a difference. Same thing
for `start()`: it's not on a hot path.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
