tkalkirill commented on code in PR #2286:
URL: https://github.com/apache/ignite-3/pull/2286#discussion_r1266675352
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogServiceImpl.java:
##########
@@ -240,6 +241,17 @@ public CatalogZoneDescriptor zone(int zoneId, long
timestamp) {
return catalogAt(timestamp).zone(zoneId);
}
+ @Override
+ public CatalogZoneDescriptor zone(int zoneId, int version) {
Review Comment:
Missing `@Nullable`
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogServiceImpl.java:
##########
@@ -150,18 +151,18 @@ public CatalogServiceImpl(UpdateLog updateLog,
ClockWaiter clockWaiter, LongSupp
@Override
public void start() {
- int objectIdGen = 0;
+ int objectIdGen = 1;
// TODO: IGNITE-19082 Move default schema objects initialization to
cluster init procedure.
CatalogSchemaDescriptor schemaPublic = new CatalogSchemaDescriptor(
- objectIdGen++,
+ 0,
Review Comment:
Why 0 ?
##########
modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/BaseDistributionZoneManagerTest.java:
##########
@@ -126,6 +126,7 @@ void setUp() {
distributionZoneManager = new DistributionZoneManager(
zonesConfiguration,
tablesConfiguration,
+ DistributionZonesTestUtil.mockCatalog(),
Review Comment:
Perhaps it would be more correct to create this method in some
`CatalogTestUtils`.
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java:
##########
@@ -38,6 +38,31 @@
* Catalog utils.
*/
public class CatalogUtils {
+ /**
+ * Default TIMESTAMP type precision: microseconds.
+ *
+ * <p>SQL`16 part 2 section 6.1 syntax rule 36
+ */
+ public static final int DEFAULT_TIMESTAMP_PRECISION = 6;
Review Comment:
```suggestion
public static final int DEFAULT_TIMESTAMP_PRECISION = 6;
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/index/ItIndexManagerTest.java:
##########
@@ -90,7 +93,7 @@ public void eventsAreFiredWhenIndexesCreatedAndDropped() {
CompletableFuture<IndexEventParameters> indexDroppedFuture =
registerListener(indexManager, IndexEvent.DROP);
- await(indexManager.dropIndexAsync("PUBLIC", "INAME", true));
+
await(indexManager.dropIndexAsync(DropIndexParams.builder().schemaName("PUBLIC").indexName("INAME").build()));
Review Comment:
same
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -1462,32 +1472,53 @@ private Set<Assignment>
calculateAssignments(TablePartitionId tablePartitionId)
}
/**
- * Creates a new table with the given {@code name} asynchronously. If a
table with the same name already exists, a future will be
- * completed with {@link TableAlreadyExistsException}.
+ * Creates a new table from parameters.
*
- * @param name Table name.
- * @param zoneName Distribution zone name.
- * @param tableInitChange Table changer.
+ * @param parameters Create table parameters.
* @return Future representing pending completion of the operation.
- * @throws IgniteException If an unspecified platform exception has
happened internally. Is thrown when:
- * <ul>
- * <li>the node is stopping.</li>
- * </ul>
* @see TableAlreadyExistsException
*/
- public CompletableFuture<Table> createTableAsync(String name, String
zoneName, Consumer<TableChange> tableInitChange) {
+ public CompletableFuture<Table> createTableAsync(CreateTableParams
parameters) {
if (!busyLock.enterBusy()) {
throw new IgniteException(new NodeStoppingException());
}
try {
- return createTableAsyncInternal(name, zoneName, tableInitChange);
+ String tableName = parameters.tableName();
+ String zoneName = Objects.requireNonNullElse(parameters.zone(),
DEFAULT_ZONE_NAME);
+
+ // Copied from DdlCommandHandler
+ Consumer<TableChange> tblChanger = tableChange -> {
Review Comment:
Maybe we should get rid of the configuration?
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java:
##########
@@ -38,6 +38,31 @@
* Catalog utils.
*/
public class CatalogUtils {
+ /**
+ * Default TIMESTAMP type precision: microseconds.
+ *
+ * <p>SQL`16 part 2 section 6.1 syntax rule 36
+ */
+ public static final int DEFAULT_TIMESTAMP_PRECISION = 6;
+ /**
+ * Default TIME type precision: seconds.
+ *
+ * <p>SQL`16 part 2 section 6.1 syntax rule 36
+ */
+ public static final int DEFAULT_TIME_PRECISION = 0;
+ /**
+ * Default DECIMAL precision is implementation-defined.
+ *
+ * <p>SQL`16 part 2 section 6.1 syntax rule 25
+ */
+ public static final int DEFAULT_DECIMAL_PRECISION = 19;
Review Comment:
```suggestion
public static final int DEFAULT_DECIMAL_PRECISION = 19;
```
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java:
##########
@@ -38,6 +38,31 @@
* Catalog utils.
*/
public class CatalogUtils {
+ /**
+ * Default TIMESTAMP type precision: microseconds.
+ *
+ * <p>SQL`16 part 2 section 6.1 syntax rule 36
+ */
+ public static final int DEFAULT_TIMESTAMP_PRECISION = 6;
+ /**
+ * Default TIME type precision: seconds.
+ *
+ * <p>SQL`16 part 2 section 6.1 syntax rule 36
+ */
+ public static final int DEFAULT_TIME_PRECISION = 0;
Review Comment:
```suggestion
public static final int DEFAULT_TIME_PRECISION = 0;
```
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogServiceImpl.java:
##########
@@ -389,6 +401,13 @@ public CompletableFuture<Void>
alterColumn(AlterColumnParams params) {
@Override
public CompletableFuture<Void> createIndex(CreateHashIndexParams params) {
+ if (StringUtils.nullOrEmpty(params.indexName())) {
Review Comment:
It would be nice to add a test for this.
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogService.java:
##########
@@ -56,8 +58,11 @@ public interface CatalogService {
CatalogZoneDescriptor zone(String zoneName, long timestamp);
+ @Deprecated
CatalogZoneDescriptor zone(int zoneId, long timestamp);
+ CatalogZoneDescriptor zone(int zoneId, int version);
Review Comment:
Missing `@Nullable`
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -1462,32 +1472,53 @@ private Set<Assignment>
calculateAssignments(TablePartitionId tablePartitionId)
}
/**
- * Creates a new table with the given {@code name} asynchronously. If a
table with the same name already exists, a future will be
- * completed with {@link TableAlreadyExistsException}.
+ * Creates a new table from parameters.
*
- * @param name Table name.
- * @param zoneName Distribution zone name.
- * @param tableInitChange Table changer.
+ * @param parameters Create table parameters.
* @return Future representing pending completion of the operation.
- * @throws IgniteException If an unspecified platform exception has
happened internally. Is thrown when:
- * <ul>
- * <li>the node is stopping.</li>
- * </ul>
* @see TableAlreadyExistsException
*/
- public CompletableFuture<Table> createTableAsync(String name, String
zoneName, Consumer<TableChange> tableInitChange) {
+ public CompletableFuture<Table> createTableAsync(CreateTableParams
parameters) {
if (!busyLock.enterBusy()) {
throw new IgniteException(new NodeStoppingException());
}
try {
- return createTableAsyncInternal(name, zoneName, tableInitChange);
+ String tableName = parameters.tableName();
+ String zoneName = Objects.requireNonNullElse(parameters.zone(),
DEFAULT_ZONE_NAME);
+
+ // Copied from DdlCommandHandler
+ Consumer<TableChange> tblChanger = tableChange -> {
+ tableChange.changeColumns(columnsChange -> {
+ for (var col : parameters.columns()) {
+ columnsChange.create(col.name(), columnChange ->
CatalogDescriptorUtils.convertColumnDefinition(col, columnChange));
+ }
+ });
+
+ var colocationKeys = parameters.colocationColumns();
Review Comment:
Variable type not clear
##########
modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java:
##########
@@ -185,6 +191,9 @@ public class DistributionZoneManager implements
IgniteComponent {
/** Tables configuration. */
private final TablesConfiguration tablesConfiguration;
+ /** Catalog service. */
Review Comment:
```suggestion
/** Catalog manager. */
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/index/ItIndexManagerTest.java:
##########
@@ -69,11 +71,12 @@ public void eventsAreFiredWhenIndexesCreatedAndDropped() {
CompletableFuture<IndexEventParameters> indexCreatedFuture =
registerListener(indexManager, IndexEvent.CREATE);
await(indexManager.createIndexAsync(
- "PUBLIC",
- "INAME",
- "TNAME",
- true,
- tableIndexChange ->
tableIndexChange.convert(HashIndexChange.class).changeColumnNames("C3", "C2")
+ CreateHashIndexParams.builder()
+ .schemaName("PUBLIC")
Review Comment:
Les's use `CatalogService#DEFAULT_SCHEMA_NAME`
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogServiceImpl.java:
##########
@@ -150,18 +151,18 @@ public CatalogServiceImpl(UpdateLog updateLog,
ClockWaiter clockWaiter, LongSupp
@Override
public void start() {
- int objectIdGen = 0;
+ int objectIdGen = 1;
// TODO: IGNITE-19082 Move default schema objects initialization to
cluster init procedure.
CatalogSchemaDescriptor schemaPublic = new CatalogSchemaDescriptor(
- objectIdGen++,
+ 0,
DEFAULT_SCHEMA_NAME,
new CatalogTableDescriptor[0],
new CatalogIndexDescriptor[0]
);
CatalogZoneDescriptor defaultZone = new CatalogZoneDescriptor(
- objectIdGen++,
+ 0,
Review Comment:
Same
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -2758,4 +2826,78 @@ private static CatalogDataStorageDescriptor
toDataStorageDescriptor(DataStorageV
return new CatalogDataStorageDescriptor(config.name(), dataRegion);
}
+
+ // Copied from DdlCommandHandler
+ @Deprecated(forRemoval = true)
+ private CompletableFuture<Void> addColumnInternal(String fullName,
List<ColumnParams> columnParams) {
+ return alterTableAsyncInternal(
+ fullName,
+ chng -> {
+ if (columnParams.isEmpty()) {
+ return false;
+ }
+
+ chng.changeColumns(cols -> {
+ Set<String> colNamesToOrders = new
HashSet<>(cols.namedListKeys());
+
+ columnParams.stream()
+ .filter(k ->
colNamesToOrders.contains(k.name()))
+ .findAny()
+ .ifPresent(c -> {
+ throw new
ColumnAlreadyExistsException(c.name());
+ });
+
+ for (ColumnParams col : columnParams) {
+ cols.create(col.name(), colChg ->
CatalogDescriptorUtils.convertColumnDefinition(col, colChg));
+ }
+ });
+
+ return true;
+ }
+ );
+ }
+
+ // Copied from DdlCommandHandler
+ // TODO: IGNITE-19082 Drop unused temporary method.
+ @Deprecated(forRemoval = true)
+ private CompletableFuture<Void> dropColumnInternal(String tableName,
Set<String> colNames) {
+ AtomicBoolean ret = new AtomicBoolean(true);
+
+ return alterTableAsyncInternal(
+ tableName,
+ chng -> {
+ chng.changeColumns(cols -> {
Review Comment:
Shouldn't we get rid of the configuration?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -1462,32 +1472,53 @@ private Set<Assignment>
calculateAssignments(TablePartitionId tablePartitionId)
}
/**
- * Creates a new table with the given {@code name} asynchronously. If a
table with the same name already exists, a future will be
- * completed with {@link TableAlreadyExistsException}.
+ * Creates a new table from parameters.
*
- * @param name Table name.
- * @param zoneName Distribution zone name.
- * @param tableInitChange Table changer.
+ * @param parameters Create table parameters.
* @return Future representing pending completion of the operation.
- * @throws IgniteException If an unspecified platform exception has
happened internally. Is thrown when:
- * <ul>
- * <li>the node is stopping.</li>
- * </ul>
* @see TableAlreadyExistsException
*/
- public CompletableFuture<Table> createTableAsync(String name, String
zoneName, Consumer<TableChange> tableInitChange) {
+ public CompletableFuture<Table> createTableAsync(CreateTableParams
parameters) {
if (!busyLock.enterBusy()) {
throw new IgniteException(new NodeStoppingException());
}
try {
- return createTableAsyncInternal(name, zoneName, tableInitChange);
+ String tableName = parameters.tableName();
+ String zoneName = Objects.requireNonNullElse(parameters.zone(),
DEFAULT_ZONE_NAME);
Review Comment:
I think the check is not needed and the default zone should be set on the
buid of the `CreateTableParams`.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -2758,4 +2826,78 @@ private static CatalogDataStorageDescriptor
toDataStorageDescriptor(DataStorageV
return new CatalogDataStorageDescriptor(config.name(), dataRegion);
}
+
+ // Copied from DdlCommandHandler
+ @Deprecated(forRemoval = true)
+ private CompletableFuture<Void> addColumnInternal(String fullName,
List<ColumnParams> columnParams) {
+ return alterTableAsyncInternal(
+ fullName,
+ chng -> {
+ if (columnParams.isEmpty()) {
+ return false;
+ }
+
+ chng.changeColumns(cols -> {
Review Comment:
Shouldn't we get rid of the configuration?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java:
##########
@@ -338,24 +338,25 @@ private PartialNode startPartialNode(
GcConfiguration gcConfig =
clusterConfigRegistry.getConfiguration(GcConfiguration.KEY);
- SchemaManager schemaManager = new SchemaManager(registry,
tablesConfig, metaStorageMgr);
+ var clockWaiter = new ClockWaiter("test", hybridClock);
Review Comment:
Is the start and stop of `ClockWaiter` and `ClockWaiter` not forgotten here?
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -177,213 +134,50 @@ public void stop() throws Exception {
}
/**
- * Creates index from provided configuration changer.
- *
- * @param schemaName A name of the schema to create index in.
- * @param indexName A name of the index to create.
- * @param tableName A name of the table to create index for.
- * @param failIfExists Flag indicates whether exception be thrown if index
exists or not.
- * @param indexChange A consumer that suppose to change the configuration
in order to provide description of an index.
- * @return A future represented the result of creation.
+ * Creates index from provided parameters.
*/
- public CompletableFuture<Boolean> createIndexAsync(
- String schemaName,
- String indexName,
- String tableName,
- boolean failIfExists,
- Consumer<TableIndexChange> indexChange
- ) {
+ public CompletableFuture<Boolean>
createIndexAsync(AbstractIndexCommandParams params) {
if (!busyLock.enterBusy()) {
return failedFuture(new NodeStoppingException());
}
- LOG.debug("Going to create index [schema={}, table={}, index={}]",
schemaName, tableName, indexName);
-
try {
- validateName(indexName);
-
- CompletableFuture<Boolean> future = new CompletableFuture<>();
-
- // Check index existence flag, avoid usage of hasCause +
IndexAlreadyExistsException.
- AtomicBoolean idxExist = new AtomicBoolean(false);
-
- tablesCfg.change(tablesChange ->
tablesChange.changeIndexes(indexListChange -> {
- idxExist.set(false);
-
- if (indexListChange.get(indexName) != null) {
- idxExist.set(true);
-
- throw new IndexAlreadyExistsException(schemaName,
indexName);
- }
-
- TableView tableCfg = tablesChange.tables().get(tableName);
-
- if (tableCfg == null) {
- throw new TableNotFoundException(schemaName, tableName);
- }
-
- int tableId = tableCfg.id();
+ CompletableFuture<Void> indexCreateFuture = (params instanceof
CreateSortedIndexParams)
+ ? catalogManager.createIndex((CreateSortedIndexParams)
params)
+ : catalogManager.createIndex((CreateHashIndexParams)
params);
- int indexId = tablesChange.globalIdCounter() + 1;
-
- tablesChange.changeGlobalIdCounter(indexId);
-
- Consumer<TableIndexChange> chg = indexChange.andThen(c ->
c.changeTableId(tableId).changeId(indexId));
-
- indexListChange.create(indexName, chg);
- })).whenComplete((index, th) -> {
- if (th != null) {
- LOG.debug("Unable to create index [schema={}, table={},
index={}]",
- th, schemaName, tableName, indexName);
-
- if (!failIfExists && idxExist.get()) {
- future.complete(false);
- } else {
- future.completeExceptionally(th);
- }
- } else {
- TableIndexConfiguration idxCfg =
tablesCfg.indexes().get(indexName);
-
- if (idxCfg != null && idxCfg.value() != null) {
- LOG.info("Index created [schema={}, table={},
index={}, indexId={}]",
- schemaName, tableName, indexName,
idxCfg.id().value());
-
- future.complete(true);
- } else {
- var exception = new IgniteInternalException(
- Common.INTERNAL_ERR, "Looks like the index was
concurrently deleted");
-
- LOG.info("Unable to create index [schema={}, table={},
index={}]",
- exception, schemaName, tableName, indexName);
-
- future.completeExceptionally(exception);
- }
- }
- });
-
- return future;
- } catch (Exception ex) {
- return failedFuture(ex);
+ return indexCreateFuture.thenApply(ignore ->
catalogManager.index(params.indexName(), Long.MAX_VALUE))
Review Comment:
For what then you receive an index from the catalog?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -1276,18 +1282,22 @@ private CompletableFuture<?> createTableLocally(
pendingTables.put(tableId, table);
startedTables.put(tableId, table);
- tablesById(causalityToken).thenAccept(ignored -> inBusyLock(busyLock,
() -> {
+ createPartsFut.thenAccept(ignored -> inBusyLock(busyLock, () -> {
pendingTables.remove(tableId);
}));
- tablesById(causalityToken)
- .thenRun(() -> inBusyLock(busyLock, () ->
completeApiCreateFuture(table)));
+ createPartsFut
+ .thenRunAsync(() -> inBusyLock(busyLock, () ->
completeApiCreateFuture(table)));
Review Comment:
Missing thread pool
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -397,14 +191,20 @@ private CompletableFuture<?>
onIndexDrop(ConfigurationNotificationEvent<TableInd
try {
CompletableFuture<?> fireEventFuture = fireEvent(IndexEvent.DROP,
new IndexEventParameters(causalityToken, tableId, idxId));
- CompletableFuture<?> dropIndexFuture =
tableManager.tableAsync(causalityToken, tableId)
+ catalogManager.table(tableId, evt.catalogVersion());
+ CompletableFuture<?> dropIndexFuture =
tableManager.tableAsync(tableId)
Review Comment:
Why without `causalityToken`?
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogServiceImpl.java:
##########
@@ -409,6 +428,12 @@ public CompletableFuture<Void>
createIndex(CreateHashIndexParams params) {
@Override
public CompletableFuture<Void> createIndex(CreateSortedIndexParams params)
{
+ if (StringUtils.nullOrEmpty(params.indexName())) {
Review Comment:
Same
##########
modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java:
##########
@@ -79,56 +81,100 @@ public class SchemaManager extends Producer<SchemaEvent,
SchemaEventParameters>
/** Prevents double stopping of the component. */
private final AtomicBoolean stopGuard = new AtomicBoolean();
- /** Tables configuration. */
- private final TablesConfiguration tablesCfg;
-
/** Versioned store for tables by name. */
private final IncrementalVersionedValue<Map<Integer, SchemaRegistryImpl>>
registriesVv;
/** Meta storage manager. */
private final MetaStorageManager metastorageMgr;
+ /** Catalog manager. */
+ private final CatalogManager catalogManager;
+
/** Constructor. */
public SchemaManager(
Consumer<LongFunction<CompletableFuture<?>>> registry,
- TablesConfiguration tablesCfg,
- MetaStorageManager metastorageMgr
- ) {
+ MetaStorageManager metastorageMgr,
+ CatalogManager catalogManager) {
Review Comment:
```suggestion
CatalogManager catalogManager
) {
```
--
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]
