Skip to content
Permalink
Browse files
ATLAS-4447 : Update composite indexes status
Signed-off-by: Pinal Shah <pinal.shah@freestoneinfotech.com>
  • Loading branch information
pinal-shah committed Oct 12, 2021
1 parent 261331b commit c7c6f3699ab2ffe3fbebfbb2885e3dd1c997c345
Show file tree
Hide file tree
Showing 5 changed files with 128 additions and 1 deletion.
@@ -173,6 +173,11 @@ public interface AtlasGraphManagement {
*/
void updateUniqueIndexesForConsistencyLock();

/**
* update valid SchemaStatus for all vertex and edge indexes.
*/
void updateSchemaStatus();

/***
* Re-index elements.
* @param indexName: Name of the index that needs to be operated on.
@@ -45,10 +45,12 @@
import org.janusgraph.core.schema.Mapping;
import org.janusgraph.core.schema.Parameter;
import org.janusgraph.core.schema.PropertyKeyMaker;
import org.janusgraph.core.schema.SchemaStatus;
import org.janusgraph.diskstorage.BackendTransaction;
import org.janusgraph.diskstorage.indexing.IndexEntry;
import org.janusgraph.graphdb.database.IndexSerializer;
import org.janusgraph.graphdb.database.StandardJanusGraph;
import org.janusgraph.graphdb.database.management.GraphIndexStatusReport;
import org.janusgraph.graphdb.database.management.ManagementSystem;
import org.janusgraph.graphdb.internal.Token;
import org.janusgraph.graphdb.log.StandardTransactionLogProcessor;
@@ -64,6 +66,12 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;

import static org.janusgraph.core.schema.SchemaAction.ENABLE_INDEX;
import static org.janusgraph.core.schema.SchemaStatus.ENABLED;
import static org.janusgraph.core.schema.SchemaStatus.INSTALLED;
import static org.janusgraph.core.schema.SchemaStatus.REGISTERED;

/**
* Janus implementation of AtlasGraphManagement.
@@ -307,6 +315,57 @@ public void updateUniqueIndexesForConsistencyLock() {
}
}

@Override
public void updateSchemaStatus() {
updateSchemaStatus(this.management, this.graph.getGraph(), Vertex.class);
updateSchemaStatus(this.management, this.graph.getGraph(), Edge.class);
}

public static void updateSchemaStatus(JanusGraphManagement mgmt, JanusGraph graph, Class<? extends Element> elementType) {
LOG.info("updating SchemaStatus for {}: Starting...", elementType.getSimpleName());
int count = 0;

Iterable<JanusGraphIndex> iterable = mgmt.getGraphIndexes(elementType);

for (JanusGraphIndex index : iterable) {

if (index.isCompositeIndex()) {
PropertyKey[] propertyKeys = index.getFieldKeys();
SchemaStatus status = index.getIndexStatus(propertyKeys[0]);
String indexName = index.name();

try {
if (status == REGISTERED) {
JanusGraphManagement management = graph.openManagement();
JanusGraphIndex indexToUpdate = management.getGraphIndex(indexName);
management.updateIndex(indexToUpdate, ENABLE_INDEX).get();
management.commit();

GraphIndexStatusReport report = ManagementSystem.awaitGraphIndexStatus(graph, indexName).status(ENABLED).call();

if (!report.getConvergedKeys().isEmpty() && report.getConvergedKeys().containsKey(indexName)) {
LOG.info("SchemaStatus updated for index: {}, from {} to {}.", index.name(), REGISTERED, ENABLED);

count++;
} else if (!report.getNotConvergedKeys().isEmpty() && report.getNotConvergedKeys().containsKey(indexName)) {
LOG.error("SchemaStatus failed to update index: {}, from {} to {}.", index.name(), REGISTERED, ENABLED);
}

} else if (status == INSTALLED) {
LOG.warn("SchemaStatus {} found for index: {}", INSTALLED, indexName);

}
} catch (InterruptedException e) {
LOG.error("IllegalStateException for indexName : {}, Exception: ", indexName, e);
} catch (ExecutionException e) {
LOG.error("ExecutionException for indexName : {}, Exception: ", indexName, e);
}
}
}

LOG.info("updating SchemaStatus for {}: {}: Done!", elementType.getSimpleName(), count);
}

private static void setConsistency(JanusGraphManagement mgmt, Class<? extends Element> elementType) {
LOG.info("setConsistency: {}: Starting...", elementType.getSimpleName());
int count = 0;
@@ -81,7 +81,8 @@ public enum AtlasConfiguration {
DSL_CACHED_TRANSLATOR("atlas.dsl.cached.translator", true),
DEBUG_METRICS_ENABLED("atlas.debug.metrics.enabled", false),
TASKS_USE_ENABLED("atlas.tasks.enabled", true),
SESSION_TIMEOUT_SECS("atlas.session.timeout.secs", -1);
SESSION_TIMEOUT_SECS("atlas.session.timeout.secs", -1),
UPDATE_COMPOSITE_INDEX_STATUS("atlas.update.composite.index.status", true);


private static final Configuration APPLICATION_PROPERTIES;
@@ -95,6 +95,7 @@ private void init() {
handlers.add(new IndexConsistencyPatch(context));
handlers.add(new ReIndexPatch(context));
handlers.add(new ProcessNamePatch(context));
handlers.add(new UpdateCompositeIndexStatusPatch(context));

LOG.info("<== AtlasPatchManager.init()");
}
@@ -0,0 +1,61 @@
/**
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.atlas.repository.patches;

import org.apache.atlas.AtlasConfiguration;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.repository.graphdb.AtlasGraph;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.atlas.model.patches.AtlasPatch.PatchStatus.UNKNOWN;

public class UpdateCompositeIndexStatusPatch extends AtlasPatchHandler {
private static final Logger LOG = LoggerFactory.getLogger(UpdateCompositeIndexStatusPatch.class);

private static final String PATCH_ID = "JAVA_PATCH_0000_010";
private static final String PATCH_DESCRIPTION = "updates schema status of composite indexes which are in REGISTERED to ENABLED.";

private final PatchContext context;

public UpdateCompositeIndexStatusPatch(PatchContext context) {
super(context.getPatchRegistry(), PATCH_ID, PATCH_DESCRIPTION);
this.context = context;
}

@Override
public void apply() throws AtlasBaseException {
if (!AtlasConfiguration.UPDATE_COMPOSITE_INDEX_STATUS.getBoolean()) {
LOG.info("UpdateCompositeIndexStatusPatch: Skipped, since not enabled!");
return;
}

AtlasGraph graph = context.getGraph();

try {
LOG.info("UpdateCompositeIndexStatusPatch: Starting...");
graph.getManagementSystem().updateSchemaStatus();
} finally {
LOG.info("UpdateCompositeIndexStatusPatch: Done!");
}

setStatus(UNKNOWN);

LOG.info("UpdateCompositeIndexStatusPatch.apply(): patchId={}, status={}", getPatchId(), getStatus());
}
}

0 comments on commit c7c6f36

Please sign in to comment.