This is an automated email from the ASF dual-hosted git repository. amashenkov pushed a commit to branch ignite-18537 in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit f955a9d91a11713bceb53ccf57c7402d045011be Author: amashenkov <[email protected]> AuthorDate: Fri May 19 20:41:19 2023 +0300 Add add/drop distribution zone DDL commands. --- .../DistributionZoneAlreadyExistsException.java | 48 +++++++++ .../lang/DistributionZoneNotFoundException.java | 48 +++++++++ .../apache/ignite/internal/catalog/Catalog.java | 40 ++++--- .../ignite/internal/catalog/CatalogManager.java | 19 ++++ .../ignite/internal/catalog/CatalogService.java | 8 +- .../internal/catalog/CatalogServiceImpl.java | 97 +++++++++++++++-- .../commands/AbstractZoneCommandParams.java | 88 ++++++++++++++++ .../internal/catalog/commands/CatalogUtils.java | 15 +++ .../catalog/commands/CreateZoneParams.java | 36 +++++++ .../internal/catalog/commands/DropZoneParams.java | 36 +++++++ ...riptor.java => DistributionZoneDescriptor.java} | 40 +------ .../catalog/descriptors/ObjectDescriptor.java | 3 +- .../internal/catalog/storage/DropZoneEntry.java | 49 +++++++++ .../internal/catalog/storage/NewZoneEntry.java | 50 +++++++++ .../internal/catalog/CatalogServiceSelfTest.java | 115 ++++++++++++++++++++- .../engine/exec/ddl/DdlCommandHandlerWrapper.java | 18 ++++ .../exec/ddl/DdlToCatalogCommandConverter.java | 17 +++ 17 files changed, 667 insertions(+), 60 deletions(-) diff --git a/modules/api/src/main/java/org/apache/ignite/lang/DistributionZoneAlreadyExistsException.java b/modules/api/src/main/java/org/apache/ignite/lang/DistributionZoneAlreadyExistsException.java new file mode 100644 index 0000000000..8cd0685b81 --- /dev/null +++ b/modules/api/src/main/java/org/apache/ignite/lang/DistributionZoneAlreadyExistsException.java @@ -0,0 +1,48 @@ +/* + * 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.lang; + +import static org.apache.ignite.lang.ErrorGroups.DistributionZones.ZONE_ALREADY_EXISTS_ERR; + +import java.util.UUID; + +/** + * This exception is thrown when a distribution zone creation has failed because a zone with the specified name already existed. + */ +public class DistributionZoneAlreadyExistsException extends IgniteException { + /** + * Creates an exception with the given distribution zone name. + * + * @param zoneName Distribution zone name. + */ + public DistributionZoneAlreadyExistsException(String zoneName) { + super(ZONE_ALREADY_EXISTS_ERR, "zone already exists [name=" + zoneName + ']'); + } + + /** + * Creates an exception with the given trace IDS, error code, detailed message, and cause. + * + * @param traceId Unique identifier of the exception. + * @param code Full error code. + * @param message Detailed message. + * @param cause Optional nested exception (can be {@code null}). + */ + public DistributionZoneAlreadyExistsException(UUID traceId, int code, String message, Throwable cause) { + super(traceId, code, message, cause); + } +} diff --git a/modules/api/src/main/java/org/apache/ignite/lang/DistributionZoneNotFoundException.java b/modules/api/src/main/java/org/apache/ignite/lang/DistributionZoneNotFoundException.java new file mode 100644 index 0000000000..5c57d5fa60 --- /dev/null +++ b/modules/api/src/main/java/org/apache/ignite/lang/DistributionZoneNotFoundException.java @@ -0,0 +1,48 @@ +/* + * 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.lang; + +import static org.apache.ignite.lang.ErrorGroups.DistributionZones.ZONE_NOT_FOUND_ERR; + +import java.util.UUID; + +/** + * Exception is thrown when a specified distribution zone cannot be found. + */ +public class DistributionZoneNotFoundException extends IgniteException { + /** + * Creates an exception with the given distribution zone name. + * + * @param zoneName Distribution zone name. + */ + public DistributionZoneNotFoundException(String zoneName) { + super(ZONE_NOT_FOUND_ERR, "The distribution zone does not exist [name=" + zoneName + ']'); + } + + /** + * Creates an exception with the given trace ID, error code, detailed message, and cause. + * + * @param traceId Unique identifier of the exception. + * @param code Full error code. + * @param message Detailed message. + * @param cause Optional nested exception (can be {@code null}). + */ + public DistributionZoneNotFoundException(UUID traceId, int code, String message, Throwable cause) { + super(traceId, code, message, cause); + } +} diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/Catalog.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/Catalog.java index d4d58764e5..36c5f9570f 100644 --- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/Catalog.java +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/Catalog.java @@ -23,6 +23,7 @@ import java.util.Map; import java.util.Objects; import java.util.function.Function; import java.util.stream.Collectors; +import org.apache.ignite.internal.catalog.descriptors.DistributionZoneDescriptor; import org.apache.ignite.internal.catalog.descriptors.IndexDescriptor; import org.apache.ignite.internal.catalog.descriptors.ObjectDescriptor; import org.apache.ignite.internal.catalog.descriptors.SchemaDescriptor; @@ -38,41 +39,48 @@ public class Catalog { private final int objectIdGen; private final long activationTimestamp; private final Map<String, SchemaDescriptor> schemas; + private final Map<String, DistributionZoneDescriptor> zones; @IgniteToStringExclude private final Map<Integer, TableDescriptor> tablesMap; @IgniteToStringExclude private final Map<Integer, IndexDescriptor> indexesMap; + @IgniteToStringExclude + private final Map<Integer, DistributionZoneDescriptor> zonesMap; /** * Constructor. * * @param version A version of the catalog. * @param activationTimestamp A timestamp when this version becomes active (i.e. available for use). - * @param objectIdGen Current state of identifier generator. This value should be used to assign an - * id to a new object in the next version of the catalog. - * @param descriptors Enumeration of schemas available in the current version of catalog. + * @param objectIdGen Current state of identifier generator. This value should be used to assign an id to a new object in the + * next version of the catalog. + * @param zones Distribution zones descriptors. + * @param schemas Enumeration of schemas available in the current version of catalog. */ public Catalog( int version, long activationTimestamp, int objectIdGen, - SchemaDescriptor... descriptors + Collection<DistributionZoneDescriptor> zones, + SchemaDescriptor... schemas ) { this.version = version; this.activationTimestamp = activationTimestamp; this.objectIdGen = objectIdGen; - Objects.requireNonNull(descriptors, "schemas"); + Objects.requireNonNull(schemas, "schemas"); - assert descriptors.length > 0 : "No schemas found"; - assert Arrays.stream(descriptors).allMatch(t -> t.version() == version) : "Invalid schema version"; + assert schemas.length > 0 : "No schemas found"; + assert Arrays.stream(schemas).allMatch(t -> t.version() == version) : "Invalid schema version"; - schemas = Arrays.stream(descriptors).collect(Collectors.toUnmodifiableMap(SchemaDescriptor::name, t -> t)); + this.schemas = Arrays.stream(schemas).collect(Collectors.toUnmodifiableMap(SchemaDescriptor::name, t -> t)); + this.zones = zones.stream().collect(Collectors.toUnmodifiableMap(DistributionZoneDescriptor::name, t -> t)); + zonesMap = zones.stream().collect(Collectors.toUnmodifiableMap(DistributionZoneDescriptor::id, t -> t)); - tablesMap = schemas.values().stream().flatMap(s -> Arrays.stream(s.tables())) + tablesMap = this.schemas.values().stream().flatMap(s -> Arrays.stream(s.tables())) .collect(Collectors.toUnmodifiableMap(ObjectDescriptor::id, Function.identity())); - indexesMap = schemas.values().stream().flatMap(s -> Arrays.stream(s.indexes())) + indexesMap = this.schemas.values().stream().flatMap(s -> Arrays.stream(s.indexes())) .collect(Collectors.toUnmodifiableMap(ObjectDescriptor::id, Function.identity())); } @@ -100,8 +108,16 @@ public class Catalog { return indexesMap.get(indexId); } - public Collection<IndexDescriptor> tableIndexes(int tableId) { - return indexesMap.values().stream().filter(desc -> desc.tableId() == tableId).collect(Collectors.toList()); + public DistributionZoneDescriptor zone(String name) { + return zones.get(name); + } + + public DistributionZoneDescriptor zone(int zoneId) { + return zonesMap.get(zoneId); + } + + public Collection<DistributionZoneDescriptor> zones() { + return zones.values(); } /** {@inheritDoc} */ diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManager.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManager.java index 2724d5d965..ee9e08f136 100644 --- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManager.java +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManager.java @@ -21,7 +21,9 @@ import java.util.concurrent.CompletableFuture; import org.apache.ignite.internal.catalog.commands.AlterTableAddColumnParams; import org.apache.ignite.internal.catalog.commands.AlterTableDropColumnParams; import org.apache.ignite.internal.catalog.commands.CreateTableParams; +import org.apache.ignite.internal.catalog.commands.CreateZoneParams; import org.apache.ignite.internal.catalog.commands.DropTableParams; +import org.apache.ignite.internal.catalog.commands.DropZoneParams; import org.apache.ignite.internal.manager.IgniteComponent; /** @@ -59,4 +61,21 @@ public interface CatalogManager extends IgniteComponent, CatalogService { * @return Operation future. */ CompletableFuture<Void> dropColumn(AlterTableDropColumnParams params); + + /** + * Creates new distribution zone. + * + * @param params Parameters. + * @return Operation future. + */ + CompletableFuture<Void> createDistributionZone(CreateZoneParams params); + + /** + * Drops distribution zone. + * + * @param params Parameters. + * @return Operation future. + */ + CompletableFuture<Void> dropDistributionZone(DropZoneParams params); + } diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogService.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogService.java index 70e70aaf38..963d8e17e2 100644 --- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogService.java +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogService.java @@ -17,7 +17,7 @@ package org.apache.ignite.internal.catalog; -import java.util.Collection; +import org.apache.ignite.internal.catalog.descriptors.DistributionZoneDescriptor; import org.apache.ignite.internal.catalog.descriptors.IndexDescriptor; import org.apache.ignite.internal.catalog.descriptors.SchemaDescriptor; import org.apache.ignite.internal.catalog.descriptors.TableDescriptor; @@ -43,10 +43,12 @@ public interface CatalogService { IndexDescriptor index(int indexId, long timestamp); - Collection<IndexDescriptor> tableIndexes(int tableId, long timestamp); - SchemaDescriptor schema(int version); + DistributionZoneDescriptor zone(String zoneName, long timestamp); + + DistributionZoneDescriptor zone(int zoneId, long timestamp); + SchemaDescriptor activeSchema(long timestamp); void listen(CatalogEvent evt, EventListener<CatalogEventParameters> closure); diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogServiceImpl.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogServiceImpl.java index 4920647d4a..0a063833d1 100644 --- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogServiceImpl.java +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogServiceImpl.java @@ -22,18 +22,21 @@ import static java.util.concurrent.CompletableFuture.failedFuture; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collection; import java.util.List; import java.util.Map.Entry; import java.util.NavigableMap; import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentSkipListMap; +import java.util.stream.Collectors; import org.apache.ignite.internal.catalog.commands.AlterTableAddColumnParams; import org.apache.ignite.internal.catalog.commands.AlterTableDropColumnParams; import org.apache.ignite.internal.catalog.commands.CatalogUtils; import org.apache.ignite.internal.catalog.commands.CreateTableParams; +import org.apache.ignite.internal.catalog.commands.CreateZoneParams; import org.apache.ignite.internal.catalog.commands.DropTableParams; +import org.apache.ignite.internal.catalog.commands.DropZoneParams; +import org.apache.ignite.internal.catalog.descriptors.DistributionZoneDescriptor; import org.apache.ignite.internal.catalog.descriptors.IndexDescriptor; import org.apache.ignite.internal.catalog.descriptors.SchemaDescriptor; import org.apache.ignite.internal.catalog.descriptors.TableDescriptor; @@ -42,7 +45,9 @@ import org.apache.ignite.internal.catalog.events.CatalogEventParameters; import org.apache.ignite.internal.catalog.events.CreateTableEventParameters; import org.apache.ignite.internal.catalog.events.DropTableEventParameters; import org.apache.ignite.internal.catalog.storage.DropTableEntry; +import org.apache.ignite.internal.catalog.storage.DropZoneEntry; import org.apache.ignite.internal.catalog.storage.NewTableEntry; +import org.apache.ignite.internal.catalog.storage.NewZoneEntry; import org.apache.ignite.internal.catalog.storage.ObjectIdGenUpdateEntry; import org.apache.ignite.internal.catalog.storage.UpdateEntry; import org.apache.ignite.internal.catalog.storage.UpdateLog; @@ -52,7 +57,10 @@ import org.apache.ignite.internal.logger.IgniteLogger; import org.apache.ignite.internal.logger.Loggers; import org.apache.ignite.internal.manager.Producer; import org.apache.ignite.internal.util.ArrayUtils; +import org.apache.ignite.internal.util.CollectionUtils; import org.apache.ignite.internal.util.PendingComparableValuesTracker; +import org.apache.ignite.lang.DistributionZoneAlreadyExistsException; +import org.apache.ignite.lang.DistributionZoneNotFoundException; import org.apache.ignite.lang.ErrorGroups.Common; import org.apache.ignite.lang.IgniteInternalException; import org.apache.ignite.lang.TableAlreadyExistsException; @@ -92,7 +100,7 @@ public class CatalogServiceImpl extends Producer<CatalogEvent, CatalogEventParam int objectIdGen = 0; SchemaDescriptor schemaPublic = new SchemaDescriptor(objectIdGen++, "PUBLIC", 0, new TableDescriptor[0], new IndexDescriptor[0]); - registerCatalog(new Catalog(0, 0L, objectIdGen, schemaPublic)); + registerCatalog(new Catalog(0, 0L, objectIdGen, List.of(), schemaPublic)); updateLog.registerUpdateHandler(new OnUpdateHandlerImpl()); @@ -123,12 +131,6 @@ public class CatalogServiceImpl extends Producer<CatalogEvent, CatalogEventParam return catalogAt(timestamp).index(indexId); } - /** {@inheritDoc} */ - @Override - public Collection<IndexDescriptor> tableIndexes(int tableId, long timestamp) { - return catalogAt(timestamp).tableIndexes(tableId); - } - /** {@inheritDoc} */ @Override public @Nullable SchemaDescriptor schema(int version) { @@ -141,6 +143,18 @@ public class CatalogServiceImpl extends Producer<CatalogEvent, CatalogEventParam return catalog.schema(CatalogService.PUBLIC); } + /** {@inheritDoc} */ + @Override + public DistributionZoneDescriptor zone(String zoneName, long timestamp) { + return catalogAt(timestamp).zone(zoneName); + } + + /** {@inheritDoc} */ + @Override + public DistributionZoneDescriptor zone(int zoneId, long timestamp) { + return catalogAt(timestamp).zone(zoneId); + } + /** {@inheritDoc} */ @Override public @Nullable SchemaDescriptor activeSchema(long timestamp) { @@ -214,6 +228,41 @@ public class CatalogServiceImpl extends Producer<CatalogEvent, CatalogEventParam return failedFuture(new UnsupportedOperationException("Not implemented yet.")); } + @Override + public CompletableFuture<Void> createDistributionZone(CreateZoneParams params) { + return saveUpdate(catalog -> { + String zoneName = Objects.requireNonNull(params.zoneName(), "zone"); + + if (catalog.zone(params.zoneName()) != null) { + throw new DistributionZoneAlreadyExistsException(zoneName); + } + + DistributionZoneDescriptor zone = CatalogUtils.fromParams(catalog.objectIdGenState(), params); + + return List.of( + new NewZoneEntry(zone), + new ObjectIdGenUpdateEntry(1) + ); + }); + } + + @Override + public CompletableFuture<Void> dropDistributionZone(DropZoneParams params) { + return saveUpdate(catalog -> { + String zoneName = Objects.requireNonNull(params.zoneName(), "zone"); + + DistributionZoneDescriptor zone = catalog.zone(params.zoneName()); + + if (zone == null) { + throw new DistributionZoneNotFoundException(zoneName); + } + + return List.of( + new DropZoneEntry(zone.id()) + ); + }); + } + private void registerCatalog(Catalog newCatalog) { catalogByVer.put(newCatalog.version(), newCatalog); catalogByTs.put(newCatalog.time(), newCatalog); @@ -273,6 +322,7 @@ public class CatalogServiceImpl extends Producer<CatalogEvent, CatalogEventParam version, System.currentTimeMillis(), catalog.objectIdGenState(), + catalog.zones(), new SchemaDescriptor( schema.id(), schema.name(), @@ -294,6 +344,7 @@ public class CatalogServiceImpl extends Producer<CatalogEvent, CatalogEventParam version, System.currentTimeMillis(), catalog.objectIdGenState(), + catalog.zones(), new SchemaDescriptor( schema.id(), schema.name(), @@ -307,12 +358,42 @@ public class CatalogServiceImpl extends Producer<CatalogEvent, CatalogEventParam CatalogEvent.TABLE_DROP, new DropTableEventParameters(version, tableId) )); + } else if (entry instanceof NewZoneEntry) { + catalog = new Catalog( + version, + System.currentTimeMillis(), + catalog.objectIdGenState(), + CollectionUtils.concat(catalog.zones(), List.of(((NewZoneEntry) entry).descriptor())), + new SchemaDescriptor( + schema.id(), + schema.name(), + version, + schema.tables(), + schema.indexes() + ) + ); + } else if (entry instanceof DropZoneEntry) { + int zoneId = ((DropZoneEntry) entry).zoneId(); + catalog = new Catalog( + version, + System.currentTimeMillis(), + catalog.objectIdGenState(), + catalog.zones().stream().filter(z -> z.id() != zoneId).collect(Collectors.toList()), + new SchemaDescriptor( + schema.id(), + schema.name(), + version, + schema.tables(), + schema.indexes() + ) + ); } else if (entry instanceof ObjectIdGenUpdateEntry) { catalog = new Catalog( version, System.currentTimeMillis(), catalog.objectIdGenState() + ((ObjectIdGenUpdateEntry) entry).delta(), + catalog.zones(), new SchemaDescriptor( schema.id(), schema.name(), diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/AbstractZoneCommandParams.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/AbstractZoneCommandParams.java new file mode 100644 index 0000000000..6d5dc844b5 --- /dev/null +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/AbstractZoneCommandParams.java @@ -0,0 +1,88 @@ +/* + * 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.internal.catalog.commands; + +/** + * Abstract distribution zone ddl command. + */ +public class AbstractZoneCommandParams implements DdlCommandParams { + /** Distribution zone name. */ + protected String zoneName; + + /** Quietly ignore this command if distribution zone is not exists. */ + protected boolean ifZoneExists; + + /** + * Returns distribution zone name. + */ + public String zoneName() { + return zoneName; + } + + /** + * Quietly ignore if distribution zone is not exist. + */ + public boolean ifZoneExists() { + return ifZoneExists; + } + + /** + * Parameters builder. + */ + protected abstract static class AbstractBuilder<ParamT extends AbstractZoneCommandParams, BuilderT> { + protected ParamT params; + + AbstractBuilder(ParamT params) { + this.params = params; + } + + /** + * Sets distribution zone name. + * + * @param zoneName Distribution zone name. + * @return {@code this}. + */ + public BuilderT zoneName(String zoneName) { + params.zoneName = zoneName; + + return (BuilderT) this; + } + + /** + * Set quietly ignore flag. + * + * @param ifZoneExists Flag. + */ + public BuilderT ifZoneExists(boolean ifZoneExists) { + params.ifZoneExists = ifZoneExists; + + return (BuilderT) this; + } + + /** + * Builds parameters. + * + * @return Parameters. + */ + public ParamT build() { + ParamT params0 = params; + params = null; + return params0; + } + } +} diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java index 1ead944291..34bed3d4c5 100644 --- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java @@ -18,6 +18,7 @@ package org.apache.ignite.internal.catalog.commands; import java.util.stream.Collectors; +import org.apache.ignite.internal.catalog.descriptors.DistributionZoneDescriptor; import org.apache.ignite.internal.catalog.descriptors.TableColumnDescriptor; import org.apache.ignite.internal.catalog.descriptors.TableDescriptor; @@ -41,6 +42,20 @@ public class CatalogUtils { ); } + /** + * Converts DropZone command params to descriptor. + * + * @param id Distribution zone id. + * @param params Parameters. + * @return Table descriptor. + */ + public static DistributionZoneDescriptor fromParams(int id, CreateZoneParams params) { + return new DistributionZoneDescriptor( + id, + params.zoneName() + ); + } + private static TableColumnDescriptor fromParams(ColumnParams params) { return new TableColumnDescriptor(params.name(), params.type(), params.nullable(), params.defaultValueDefinition()); } diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CreateZoneParams.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CreateZoneParams.java new file mode 100644 index 0000000000..6b84f0ca6b --- /dev/null +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CreateZoneParams.java @@ -0,0 +1,36 @@ +/* + * 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.internal.catalog.commands; + +/** + * DROP ZONE statement. + */ +public class CreateZoneParams extends AbstractZoneCommandParams { + public static Builder builder() { + return new Builder(); + } + + /** + * Parameters builder. + */ + public static class Builder extends AbstractBuilder<CreateZoneParams, Builder> { + Builder() { + super(new CreateZoneParams()); + } + } +} diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/DropZoneParams.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/DropZoneParams.java new file mode 100644 index 0000000000..a2529e8b81 --- /dev/null +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/DropZoneParams.java @@ -0,0 +1,36 @@ +/* + * 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.internal.catalog.commands; + +/** + * DROP ZONE statement. + */ +public class DropZoneParams extends AbstractZoneCommandParams { + public static Builder builder() { + return new Builder(); + } + + /** + * Parameters builder. + */ + public static class Builder extends AbstractBuilder<DropZoneParams, Builder> { + Builder() { + super(new DropZoneParams()); + } + } +} diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/ObjectDescriptor.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/DistributionZoneDescriptor.java similarity index 51% copy from modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/ObjectDescriptor.java copy to modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/DistributionZoneDescriptor.java index e3ab1ff111..e060a7a161 100644 --- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/ObjectDescriptor.java +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/DistributionZoneDescriptor.java @@ -17,39 +17,16 @@ package org.apache.ignite.internal.catalog.descriptors; -import java.io.Serializable; -import java.util.Objects; import org.apache.ignite.internal.tostring.S; /** - * Base class for catalog objects. - * TODO: IGNITE-19082 Implement custom effective serialization instead. + * Distribution zone descriptor base class. */ -public abstract class ObjectDescriptor implements Serializable { - private static final long serialVersionUID = -6525237234280004860L; - private final int id; - private final String name; - private final Type type; +public class DistributionZoneDescriptor extends ObjectDescriptor { + private static final long serialVersionUID = 1093607327002694066L; - ObjectDescriptor(int id, Type type, String name) { - this.id = id; - this.type = Objects.requireNonNull(type, "type"); - this.name = Objects.requireNonNull(name, "name"); - } - - /** Returns id of the described object. */ - public int id() { - return id; - } - - /** Returns name of the described object. */ - public String name() { - return name; - } - - /** Return schema-object type. */ - public Type type() { - return type; + public DistributionZoneDescriptor(int id, String name) { + super(id, Type.ZONE, name); } /** {@inheritDoc} */ @@ -57,11 +34,4 @@ public abstract class ObjectDescriptor implements Serializable { public String toString() { return S.toString(this); } - - /** Catalog object type. */ - enum Type { - SCHEMA, - TABLE, - INDEX - } } diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/ObjectDescriptor.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/ObjectDescriptor.java index e3ab1ff111..46fa508e18 100644 --- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/ObjectDescriptor.java +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/ObjectDescriptor.java @@ -62,6 +62,7 @@ public abstract class ObjectDescriptor implements Serializable { enum Type { SCHEMA, TABLE, - INDEX + INDEX, + ZONE } } diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/DropZoneEntry.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/DropZoneEntry.java new file mode 100644 index 0000000000..2dc1816eb8 --- /dev/null +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/DropZoneEntry.java @@ -0,0 +1,49 @@ +/* + * 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.internal.catalog.storage; + +import org.apache.ignite.internal.tostring.S; + +/** + * Describes deletion of a zone. + */ +public class DropZoneEntry implements UpdateEntry { + private static final long serialVersionUID = 7727583734058987315L; + + private final int zoneId; + + /** + * Constructs the object. + * + * @param zoneId An id of a zone to drop. + */ + public DropZoneEntry(int zoneId) { + this.zoneId = zoneId; + } + + /** Returns an id of a zone to drop. */ + public int zoneId() { + return zoneId; + } + + /** {@inheritDoc} */ + @Override + public String toString() { + return S.toString(this); + } +} diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/NewZoneEntry.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/NewZoneEntry.java new file mode 100644 index 0000000000..55063f09d1 --- /dev/null +++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/NewZoneEntry.java @@ -0,0 +1,50 @@ +/* + * 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.internal.catalog.storage; + +import org.apache.ignite.internal.catalog.descriptors.DistributionZoneDescriptor; +import org.apache.ignite.internal.tostring.S; + +/** + * Describes addition of a new zone. + */ +public class NewZoneEntry implements UpdateEntry { + private static final long serialVersionUID = 2970125889493580121L; + + private final DistributionZoneDescriptor descriptor; + + /** + * Constructs the object. + * + * @param descriptor A descriptor of a zone to add. + */ + public NewZoneEntry(DistributionZoneDescriptor descriptor) { + this.descriptor = descriptor; + } + + /** Returns descriptor of a zone to add. */ + public DistributionZoneDescriptor descriptor() { + return descriptor; + } + + /** {@inheritDoc} */ + @Override + public String toString() { + return S.toString(this); + } +} diff --git a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogServiceSelfTest.java b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogServiceSelfTest.java index 003ab864f1..c309390909 100644 --- a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogServiceSelfTest.java +++ b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogServiceSelfTest.java @@ -39,8 +39,11 @@ import java.util.List; import java.util.concurrent.CompletableFuture; import org.apache.ignite.internal.catalog.commands.ColumnParams; import org.apache.ignite.internal.catalog.commands.CreateTableParams; +import org.apache.ignite.internal.catalog.commands.CreateZoneParams; import org.apache.ignite.internal.catalog.commands.DefaultValue; import org.apache.ignite.internal.catalog.commands.DropTableParams; +import org.apache.ignite.internal.catalog.commands.DropZoneParams; +import org.apache.ignite.internal.catalog.descriptors.DistributionZoneDescriptor; import org.apache.ignite.internal.catalog.descriptors.SchemaDescriptor; import org.apache.ignite.internal.catalog.descriptors.TableDescriptor; import org.apache.ignite.internal.catalog.events.CatalogEvent; @@ -58,6 +61,8 @@ import org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager; import org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage; import org.apache.ignite.internal.vault.VaultManager; import org.apache.ignite.internal.vault.inmemory.InMemoryVaultService; +import org.apache.ignite.lang.DistributionZoneAlreadyExistsException; +import org.apache.ignite.lang.DistributionZoneNotFoundException; import org.apache.ignite.lang.IgniteInternalException; import org.apache.ignite.lang.NodeStoppingException; import org.apache.ignite.lang.TableAlreadyExistsException; @@ -74,6 +79,7 @@ import org.mockito.Mockito; * Catalog service self test. */ public class CatalogServiceSelfTest { + private static final String ZONE_NAME = "myZone"; private static final String TABLE_NAME = "myTable"; private static final String TABLE_NAME_2 = "myTable2"; @@ -234,12 +240,14 @@ public class CatalogServiceSelfTest { } @Test - public void testDropTable() { + public void testDropTable() throws InterruptedException { assertThat(service.createTable(simpleTable(TABLE_NAME)), willBe((Object) null)); assertThat(service.createTable(simpleTable(TABLE_NAME_2)), willBe((Object) null)); long beforeDropTimestamp = System.currentTimeMillis(); + Thread.sleep(5); + DropTableParams dropTableParams = DropTableParams.builder().schemaName("PUBLIC").tableName(TABLE_NAME).build(); assertThat(service.dropTable(dropTableParams), willBe((Object) null)); @@ -393,6 +401,111 @@ public class CatalogServiceSelfTest { verifyNoMoreInteractions(eventListener); } + @Test + public void testCreateZone() { + CreateZoneParams params = CreateZoneParams.builder() + .zoneName(ZONE_NAME) + .build(); + + CompletableFuture<Void> fut = service.createDistributionZone(params); + + assertThat(fut, willBe((Object) null)); + + // Validate catalog version from the past. + assertNull(service.zone(ZONE_NAME, 0)); + assertNull(service.zone(1, 0)); + assertNull(service.zone(ZONE_NAME, 123L)); + assertNull(service.zone(1, 123L)); + + // Validate actual catalog + DistributionZoneDescriptor zone = service.zone(ZONE_NAME, System.currentTimeMillis()); + + assertNotNull(zone); + assertSame(zone, service.zone(1, System.currentTimeMillis())); + + // Validate newly created table + assertEquals(1L, zone.id()); + assertEquals(ZONE_NAME, zone.name()); + //TODO: fixme + } + + @Test + public void testDropZone() throws InterruptedException { + CreateZoneParams createZoneParams = CreateZoneParams.builder() + .zoneName(ZONE_NAME) + .build(); + + assertThat(service.createDistributionZone(createZoneParams), willBe((Object) null)); + + long beforeDropTimestamp = System.currentTimeMillis(); + + Thread.sleep(5); + + DropZoneParams params = DropZoneParams.builder() + .zoneName(ZONE_NAME) + .build(); + + CompletableFuture<Void> fut = service.dropDistributionZone(params); + + assertThat(fut, willBe((Object) null)); + + // Validate catalog version from the past. + DistributionZoneDescriptor zone = service.zone(ZONE_NAME, beforeDropTimestamp); + + assertNotNull(zone); + assertEquals(ZONE_NAME, zone.name()); + assertEquals(1, zone.id()); + + assertSame(zone, service.zone(1, beforeDropTimestamp)); + + // Validate actual catalog + assertNull(service.zone(ZONE_NAME, System.currentTimeMillis())); + assertNull(service.zone(1, System.currentTimeMillis())); + } + + @Test + public void testCreateZoneIfExistsFlag() { + CreateZoneParams params = CreateZoneParams.builder() + .zoneName(ZONE_NAME) + .ifZoneExists(true) + .build(); + + assertThat(service.createDistributionZone(params), willBe((Object) null)); + assertThat(service.createDistributionZone(params), willThrowFast(DistributionZoneAlreadyExistsException.class)); + + CompletableFuture<?> fut = service.createDistributionZone( + CreateZoneParams.builder() + .zoneName(ZONE_NAME) + .ifZoneExists(false) + .build()); + + assertThat(fut, willThrowFast(DistributionZoneAlreadyExistsException.class)); + } + + @Test + public void testDropZoneIfExistsFlag() { + CreateZoneParams createZoneParams = CreateZoneParams.builder() + .zoneName(ZONE_NAME) + .build(); + + assertThat(service.createDistributionZone(createZoneParams), willBe((Object) null)); + + DropZoneParams params = DropZoneParams.builder() + .zoneName(ZONE_NAME) + .ifZoneExists(true) + .build(); + + assertThat(service.dropDistributionZone(params), willBe((Object) null)); + assertThat(service.dropDistributionZone(params), willThrowFast(DistributionZoneNotFoundException.class)); + + params = DropZoneParams.builder() + .zoneName(ZONE_NAME) + .ifZoneExists(false) + .build(); + + assertThat(service.dropDistributionZone(params), willThrowFast(DistributionZoneNotFoundException.class)); + } + private static CreateTableParams simpleTable(String name) { return CreateTableParams.builder() .schemaName("PUBLIC") diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlCommandHandlerWrapper.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlCommandHandlerWrapper.java index 21f39ea169..5a78814d21 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlCommandHandlerWrapper.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlCommandHandlerWrapper.java @@ -21,10 +21,14 @@ import java.util.Objects; import java.util.concurrent.CompletableFuture; import org.apache.ignite.internal.catalog.CatalogManager; import org.apache.ignite.internal.distributionzones.DistributionZoneManager; +import org.apache.ignite.internal.distributionzones.exception.DistributionZoneAlreadyExistsException; +import org.apache.ignite.internal.distributionzones.exception.DistributionZoneNotFoundException; import org.apache.ignite.internal.index.IndexManager; import org.apache.ignite.internal.sql.engine.prepare.ddl.CreateTableCommand; +import org.apache.ignite.internal.sql.engine.prepare.ddl.CreateZoneCommand; import org.apache.ignite.internal.sql.engine.prepare.ddl.DdlCommand; import org.apache.ignite.internal.sql.engine.prepare.ddl.DropTableCommand; +import org.apache.ignite.internal.sql.engine.prepare.ddl.DropZoneCommand; import org.apache.ignite.internal.storage.DataStorageManager; import org.apache.ignite.internal.table.distributed.TableManager; import org.apache.ignite.lang.TableAlreadyExistsException; @@ -70,6 +74,20 @@ public class DdlCommandHandlerWrapper extends DdlCommandHandler { .thenCompose(res -> catalogManager.dropTable(DdlToCatalogCommandConverter.convert((DropTableCommand) cmd)) .handle(handleModificationResult(((DropTableCommand) cmd).ifTableExists(), TableNotFoundException.class)) ); + } else if (cmd instanceof CreateZoneCommand) { + CreateZoneCommand zoneCommand = (CreateZoneCommand) cmd; + + return ddlCommandFuture + .thenCompose(res -> catalogManager.createDistributionZone(DdlToCatalogCommandConverter.convert(zoneCommand)) + .handle(handleModificationResult(zoneCommand.ifNotExists(), DistributionZoneAlreadyExistsException.class)) + ); + } else if (cmd instanceof DropZoneCommand) { + DropZoneCommand zoneCommand = (DropZoneCommand) cmd; + + return ddlCommandFuture + .thenCompose(res -> catalogManager.dropDistributionZone(DdlToCatalogCommandConverter.convert(zoneCommand)) + .handle(handleModificationResult(zoneCommand.ifExists(), DistributionZoneNotFoundException.class)) + ); } return ddlCommandFuture; diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlToCatalogCommandConverter.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlToCatalogCommandConverter.java index e0e9f4f520..4868db6ab1 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlToCatalogCommandConverter.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlToCatalogCommandConverter.java @@ -21,12 +21,16 @@ import java.util.List; import java.util.stream.Collectors; import org.apache.ignite.internal.catalog.commands.ColumnParams; import org.apache.ignite.internal.catalog.commands.CreateTableParams; +import org.apache.ignite.internal.catalog.commands.CreateZoneParams; import org.apache.ignite.internal.catalog.commands.DefaultValue; import org.apache.ignite.internal.catalog.commands.DropTableParams; +import org.apache.ignite.internal.catalog.commands.DropZoneParams; import org.apache.ignite.internal.sql.engine.prepare.ddl.ColumnDefinition; import org.apache.ignite.internal.sql.engine.prepare.ddl.CreateTableCommand; +import org.apache.ignite.internal.sql.engine.prepare.ddl.CreateZoneCommand; import org.apache.ignite.internal.sql.engine.prepare.ddl.DefaultValueDefinition; import org.apache.ignite.internal.sql.engine.prepare.ddl.DropTableCommand; +import org.apache.ignite.internal.sql.engine.prepare.ddl.DropZoneCommand; import org.apache.ignite.internal.sql.engine.util.TypeUtils; /** @@ -56,6 +60,19 @@ class DdlToCatalogCommandConverter { .build(); } + static CreateZoneParams convert(CreateZoneCommand cmd) { + return CreateZoneParams.builder() + .zoneName(cmd.zoneName()) + //TOOD: fixme + .build(); + } + + static DropZoneParams convert(DropZoneCommand cmd) { + return DropZoneParams.builder() + .zoneName(cmd.zoneName()) + .build(); + } + private static ColumnParams convert(ColumnDefinition def) { return new ColumnParams(def.name(), TypeUtils.columnType(def.type()), convert(def.defaultValueDefinition()), def.nullable()); }
