Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

IGNITE-24215 Improve internal Catalog API #5112

Merged
merged 9 commits into from
Jan 27, 2025
Merged

IGNITE-24215 Improve internal Catalog API #5112

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
1e1b5c7
wip. Moved get-by-name methods for catalog objects from CatalogServic…
AMashenkov Jan 21, 2025
12c74b5
Improve javadocs.
AMashenkov Jan 23, 2025
a2fdcde
wip. Move get-by-timestamp methods from CatalogService to Catalog.
AMashenkov Jan 23, 2025
9d79823
wip. Move get-by-version methods from CatalogService to Catalog.
AMashenkov Jan 23, 2025
305c5ff
minor
AMashenkov Jan 24, 2025
4b25968
minor after review
AMashenkov Jan 27, 2025
4f2f82d
minor after review
AMashenkov Jan 27, 2025
274fbd2
fix javadoc
AMashenkov Jan 27, 2025
8c56384
Add TODOs
AMashenkov Jan 27, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
116 changes: 109 additions & 7 deletions modules/catalog/src/main/java/org/apache/ignite/internal/catalog/Catalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import java.util.Objects;
import java.util.stream.Collector;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.catalog.descriptors.CatalogObjectDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogSchemaDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
Expand All @@ -46,7 +47,9 @@
import org.jetbrains.annotations.Nullable;

/**
* Catalog descriptor represents database schema snapshot.
* Catalog descriptor represents a snapshot of the database schema.
*
* <p>It contains information about schemas, tables, indexes, and zones available in the current version of the catalog.
*/
public class Catalog {
private static <T extends CatalogObjectDescriptor> Collector<T, ?, Map<String, T>> toMapByName() {
Expand Down Expand Up @@ -128,79 +131,178 @@ public Catalog(
}
}

/**
* Returns the version of the catalog.
*
* @return The version of the catalog.
*/
public int version() {
return version;
}

/** Returns timestamp when this version becomes active (i.e. available for use). */
/**
* Returns the timestamp when this version becomes active (i.e., available for use).
*
* @return The activation timestamp.
*/
public long time() {
return activationTimestamp;
}

/**
* Returns the current state of the identifier generator. This value is used to generate an unique id for a new object in the next
* versions of the catalog.
*
* @return The current state of the identifier generator.
*/
public int objectIdGenState() {
return objectIdGen;
}

/**
* Returns the schema descriptor by schema name.
*
* @param name The name of the schema.
* @return The schema descriptor or {@code null} if the schema is not found.
*/
public @Nullable CatalogSchemaDescriptor schema(String name) {
return schemasByName.get(name);
}

/**
* Returns the schema descriptor by schema ID.
*
* @param schemaId The ID of the schema.
* @return The schema descriptor or {@code null} if the schema is not found.
*/
public @Nullable CatalogSchemaDescriptor schema(int schemaId) {
return schemasById.get(schemaId);
}

/**
* Returns all schemas in the catalog.
*
* @return A collection of all schema descriptors.
*/
public Collection<CatalogSchemaDescriptor> schemas() {
return schemasByName.values();
}

/**
* Returns the table descriptor by table ID.
*
* @param tableId The ID of the table.
* @return The table descriptor or {@code null} if the table is not found.
*/
public @Nullable CatalogTableDescriptor table(int tableId) {
return tablesById.get(tableId);
}

/**
* Returns table descriptor by table name and schema name. A shortcut method for {@code catalog.schema(schemaName).table(tableName)}.
* Returns the table descriptor by table name and schema name. Both names should be normalized.
*
* @param schemaName Schema name. Case-sensitive, without quotes.
* @param tableName Table name without schema. Case-sensitive, without quotes.
* @return Table descriptor or {@code null} if schema or table not found.
* */
* @param schemaName The name of the schema. Case-sensitive, without quotes.
* @param tableName The name of the table without schema. Case-sensitive, without quotes.
* @return The table descriptor or {@code null} if the schema or table is not found.
*/
public @Nullable CatalogTableDescriptor table(String schemaName, String tableName) {
CatalogSchemaDescriptor schema = schema(schemaName);
return schema == null ? null : schema.table(tableName);
}

/**
* Returns all tables in the catalog.
*
* @return A collection of all table descriptors.
*/
public Collection<CatalogTableDescriptor> tables() {
return tablesById.values();
}


/**
* Returns an index descriptor by the given index name and schema name, that is an index that has not been dropped yet.
*
* <p>This effectively means that the index must be present in the Catalog and not in the {@link CatalogIndexStatus#STOPPING}
* state.
*
* @param schemaName The name of the schema.
* @param indexName The name of the index.
* @return The index descriptor or {@code null} if the schema or index is not found.
*/
public @Nullable CatalogIndexDescriptor aliveIndex(String schemaName, String indexName) {
CatalogSchemaDescriptor schema = schema(schemaName);
return schema == null ? null : schema.aliveIndex(indexName);
}

/**
* Returns the index descriptor by index ID.
*
* @param indexId The ID of the index.
* @return The index descriptor or {@code null} if the index is not found.
*/
public @Nullable CatalogIndexDescriptor index(int indexId) {
return indexesById.get(indexId);
}

/**
* Returns all indexes in the catalog.
*
* @return A collection of all index descriptors.
*/
public Collection<CatalogIndexDescriptor> indexes() {
return indexesById.values();
}

/**
* Returns a list of index descriptors for a given table ID.
*
* @param tableId The ID of the table.
* @return A list of index descriptors or an empty list if no indexes are found.
*/
public List<CatalogIndexDescriptor> indexes(int tableId) {
return indexesByTableId.getOrDefault(tableId, List.of());
}

/**
* Returns the zone descriptor by zone name.
*
* @param name The name of the zone.
* @return The zone descriptor or {@code null} if the zone is not found.
*/
public @Nullable CatalogZoneDescriptor zone(String name) {
return zonesByName.get(name);
}

/**
* Returns the zone descriptor by zone ID.
*
* @param zoneId The ID of the zone.
* @return The zone descriptor or {@code null} if the zone is not found.
*/
public @Nullable CatalogZoneDescriptor zone(int zoneId) {
return zonesById.get(zoneId);
}

/**
* Returns all zones in the catalog.
*
* @return A collection of all zone descriptors.
*/
public Collection<CatalogZoneDescriptor> zones() {
return zonesByName.values();
}

/**
* Returns the default zone descriptor.
*
* @return The default zone descriptor or {@code null} if no default zone is set.
*/
public @Nullable CatalogZoneDescriptor defaultZone() {
return defaultZone;
}

/** {@inheritDoc} */
@Override
public String toString() {
return S.toString(this);
Expand Down
5 changes: 5 additions & 0 deletions modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,4 +41,9 @@ public interface CatalogManager extends IgniteComponent, CatalogService {
* @return Future representing result of execution (it will be completed with the created catalog version).
*/
CompletableFuture<Integer> execute(List<CatalogCommand> commands);

/**
* Returns a future, which completes when empty catalog is initialised. Otherwise this future completes upon startup.
*/
CompletableFuture<Void> catalogInitializationFuture();
}
118 changes: 5 additions & 113 deletions modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map.Entry;
import java.util.NavigableMap;
Expand All @@ -42,11 +41,7 @@
import org.apache.ignite.internal.catalog.commands.CreateSchemaCommand;
import org.apache.ignite.internal.catalog.commands.CreateZoneCommand;
import org.apache.ignite.internal.catalog.commands.StorageProfileParams;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogObjectDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogSchemaDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
import org.apache.ignite.internal.catalog.events.CatalogEvent;
import org.apache.ignite.internal.catalog.events.CatalogEventParameters;
import org.apache.ignite.internal.catalog.storage.Fireable;
Expand Down Expand Up @@ -86,8 +81,6 @@ public class CatalogManagerImpl extends AbstractEventProducer<CatalogEvent, Cata
/** Safe time to wait before new Catalog version activation. */
static final int DEFAULT_DELAY_DURATION = 0;

static final int DEFAULT_PARTITION_IDLE_SAFE_TIME_PROPAGATION_PERIOD = 0;

/**
* Initial update token for a catalog descriptor, this token is valid only before the first call of
* {@link UpdateEntry#applyUpdate(Catalog, long)}.
Expand Down Expand Up @@ -191,112 +184,6 @@ public CompletableFuture<Void> stopAsync(ComponentContext componentContext) {
return updateLog.stopAsync(componentContext);
}

@Override
public @Nullable CatalogTableDescriptor table(String schemaName, String tableName, long timestamp) {
CatalogSchemaDescriptor schema = catalogAt(timestamp).schema(schemaName);
if (schema == null) {
return null;
}
return schema.table(tableName);
}

@Override
public @Nullable CatalogTableDescriptor table(int tableId, long timestamp) {
return catalogAt(timestamp).table(tableId);
}

@Override
public @Nullable CatalogTableDescriptor table(int tableId, int catalogVersion) {
return catalog(catalogVersion).table(tableId);
}

@Override
public Collection<CatalogTableDescriptor> tables(int catalogVersion) {
return catalog(catalogVersion).tables();
}

@Override
public @Nullable CatalogIndexDescriptor aliveIndex(String schemaName, String indexName, long timestamp) {
CatalogSchemaDescriptor schema = catalogAt(timestamp).schema(schemaName);
if (schema == null) {
return null;
}
return schema.aliveIndex(indexName);
}

@Override
public @Nullable CatalogIndexDescriptor index(int indexId, long timestamp) {
return catalogAt(timestamp).index(indexId);
}

@Override
public @Nullable CatalogIndexDescriptor index(int indexId, int catalogVersion) {
return catalog(catalogVersion).index(indexId);
}

@Override
public Collection<CatalogIndexDescriptor> indexes(int catalogVersion) {
return catalog(catalogVersion).indexes();
}

@Override
public List<CatalogIndexDescriptor> indexes(int catalogVersion, int tableId) {
return catalog(catalogVersion).indexes(tableId);
}

@Override
public @Nullable CatalogSchemaDescriptor schema(int catalogVersion) {
return schema(SqlCommon.DEFAULT_SCHEMA_NAME, catalogVersion);
}

@Override
public @Nullable CatalogSchemaDescriptor schema(String schemaName, int catalogVersion) {
Catalog catalog = catalog(catalogVersion);

if (catalog == null) {
return null;
}

return catalog.schema(schemaName == null ? SqlCommon.DEFAULT_SCHEMA_NAME : schemaName);
}

@Override
public @Nullable CatalogSchemaDescriptor schema(int schemaId, int catalogVersion) {
Catalog catalog = catalog(catalogVersion);

return catalog == null ? null : catalog.schema(schemaId);
}

@Override
public @Nullable CatalogZoneDescriptor zone(String zoneName, long timestamp) {
return catalogAt(timestamp).zone(zoneName);
}

@Override
public @Nullable CatalogZoneDescriptor zone(int zoneId, long timestamp) {
return catalogAt(timestamp).zone(zoneId);
}

@Override
public @Nullable CatalogZoneDescriptor zone(int zoneId, int catalogVersion) {
return catalog(catalogVersion).zone(zoneId);
}

@Override
public Collection<CatalogZoneDescriptor> zones(int catalogVersion) {
return catalog(catalogVersion).zones();
}

@Override
public @Nullable CatalogSchemaDescriptor activeSchema(long timestamp) {
return catalogAt(timestamp).schema(SqlCommon.DEFAULT_SCHEMA_NAME);
}

@Override
public @Nullable CatalogSchemaDescriptor activeSchema(String schemaName, long timestamp) {
return catalogAt(timestamp).schema(schemaName == null ? SqlCommon.DEFAULT_SCHEMA_NAME : schemaName);
}

@Override
public int activeCatalogVersion(long timestamp) {
return catalogAt(timestamp).version();
Expand Down Expand Up @@ -327,6 +214,11 @@ public CompletableFuture<Void> catalogInitializationFuture() {
return catalogByVer.get(catalogVersion);
}

@Override
public Catalog activeCatalog(long timestamp) {
return catalogAt(timestamp);
}

private Catalog catalogAt(long timestamp) {
Entry<Long, Catalog> entry = catalogByTs.floorEntry(timestamp);

Expand Down
Loading