diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fb8d60ab270..3cc75b5cc45 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -106,16 +106,11 @@ jobs: export DOCKER_BUILDKIT=1 export COMPOSE_DOCKER_CLI_BUILD=1 docker compose -f docker-compose.atlas-base.yml build - docker compose \ - -f docker-compose.atlas.yml \ - -f docker-compose.atlas-hadoop.yml \ - -f docker-compose.atlas-hbase.yml \ - -f docker-compose.atlas-kafka.yml \ - -f docker-compose.atlas-hive.yml up -d --wait + docker compose -f docker-compose.atlas.yml -f docker-compose.atlas-hadoop.yml -f docker-compose.atlas-hive.yml up -d --wait - name: Check status of containers and remove them run: | - containers=(atlas atlas-hadoop atlas-hbase atlas-kafka atlas-hive); + containers=(atlas-zk atlas-solr atlas-kafka atlas-db atlas-hadoop atlas-hbase atlas-hive atlas); flag=true; for container in "${containers[@]}"; do if [[ $(docker inspect -f '{{.State.Running}}' $container 2>/dev/null) == "true" ]]; then diff --git a/dev-support/atlas-docker/.env b/dev-support/atlas-docker/.env index 8e608f7956a..a4f208f6291 100644 --- a/dev-support/atlas-docker/.env +++ b/dev-support/atlas-docker/.env @@ -24,3 +24,5 @@ HBASE_VERSION=2.5.0 KAFKA_VERSION=2.8.2 HIVE_VERSION=3.1.3 HIVE_HADOOP_VERSION=3.1.1 + +ATLAS_BACKEND=hbase diff --git a/dev-support/atlas-docker/Dockerfile.atlas b/dev-support/atlas-docker/Dockerfile.atlas index 95a34168b55..b46b8eb2d44 100644 --- a/dev-support/atlas-docker/Dockerfile.atlas +++ b/dev-support/atlas-docker/Dockerfile.atlas @@ -16,6 +16,7 @@ FROM atlas-base:latest +ARG ATLAS_BACKEND ARG ATLAS_SERVER_JAVA_VERSION ARG ATLAS_VERSION ARG TARGETARCH diff --git a/dev-support/atlas-docker/Dockerfile.atlas-db b/dev-support/atlas-docker/Dockerfile.atlas-db index 95e67675bec..8fa5cdd4806 100644 --- a/dev-support/atlas-docker/Dockerfile.atlas-db +++ b/dev-support/atlas-docker/Dockerfile.atlas-db @@ -14,7 +14,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -FROM postgres:12 +FROM postgres:13.21 # Copy DB init script USER 0 diff --git a/dev-support/atlas-docker/README.md b/dev-support/atlas-docker/README.md index e557c23bc2a..949228ea911 100644 --- a/dev-support/atlas-docker/README.md +++ b/dev-support/atlas-docker/README.md @@ -42,17 +42,24 @@ Docker files in this folder create docker images and run them to build Apache At export COMPOSE_DOCKER_CLI_BUILD=1 ~~~ -6. Build and deploy Apache Atlas in containers using docker-compose +6. Build and deploy Apache Atlas in containers using docker compose - 6.1. Ensure that the `${HOME}/.m2` directory exists and Execute following command to build Apache Atlas: + 6.1. Build atlas-base image with the following command: + docker compose -f docker-compose.atlas-base.yml build + + 6.2. Ensure that the `${HOME}/.m2` directory exists and execute following command to build Apache Atlas: mkdir -p ${HOME}/.m2 - docker-compose -f docker-compose.atlas-base.yml -f docker-compose.atlas-build.yml up + docker compose -f docker-compose.atlas-build.yml up Time taken to complete the build might vary (upto an hour), depending on status of ${HOME}/.m2 directory cache. - 6.2. Execute following command to install and start Atlas and dependent services (Solr, HBase, Kafka) in containers: + 6.3. To install and start Atlas using Postgres as backend store, execute following commands: + export ATLAS_BACKEND=postgres + docker compose -f docker-compose.atlas.yml up -d --wait - docker-compose -f docker-compose.atlas-base.yml -f docker-compose.atlas.yml -f docker-compose.atlas-hadoop.yml -f docker-compose.atlas-hbase.yml -f docker-compose.atlas-kafka.yml -f docker-compose.atlas-hive.yml up -d + 6.4. To install and start Atlas using HBase as backend store, execute following commands: + export ATLAS_BACKEND=hbase + docker compose -f docker-compose.atlas.yml -f docker-compose.atlas-hadoop.yml up -d --wait Apache Atlas will be installed at /opt/atlas/, and logs are at /var/logs/atlas directory. diff --git a/dev-support/atlas-docker/config/init_postgres.sh b/dev-support/atlas-docker/config/init_postgres.sh index 06028789fbf..bb7b9c66514 100644 --- a/dev-support/atlas-docker/config/init_postgres.sh +++ b/dev-support/atlas-docker/config/init_postgres.sh @@ -23,4 +23,14 @@ psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" <<-E CREATE USER hive WITH PASSWORD 'atlasR0cks!'; CREATE DATABASE hive; GRANT ALL PRIVILEGES ON DATABASE hive TO hive; + + CREATE USER atlas WITH PASSWORD 'atlasR0cks!'; + CREATE DATABASE atlas; + GRANT ALL PRIVILEGES ON DATABASE atlas TO atlas; + + \c hive + GRANT ALL ON SCHEMA public TO public; + + \c atlas + GRANT ALL ON SCHEMA public TO public; EOSQL diff --git a/dev-support/atlas-docker/docker-compose.atlas-backend.yml b/dev-support/atlas-docker/docker-compose.atlas-backend.yml new file mode 100644 index 00000000000..04072d771b8 --- /dev/null +++ b/dev-support/atlas-docker/docker-compose.atlas-backend.yml @@ -0,0 +1,10 @@ +services: + hbase: + extends: + file: docker-compose.atlas-hbase.yml + service: atlas-hbase + + postgres: + extends: + file: docker-compose.atlas-common.yml + service: atlas-db diff --git a/dev-support/atlas-docker/docker-compose.atlas-build.yml b/dev-support/atlas-docker/docker-compose.atlas-build.yml index 468c1b62966..8d41c521263 100644 --- a/dev-support/atlas-docker/docker-compose.atlas-build.yml +++ b/dev-support/atlas-docker/docker-compose.atlas-build.yml @@ -18,8 +18,6 @@ services: - ./patches:/home/atlas/patches - ./dist:/home/atlas/dist - ./../..:/home/atlas/src:delegated - depends_on: - - atlas-base environment: - ATLAS_VERSION - BRANCH diff --git a/dev-support/atlas-docker/docker-compose.atlas-common.yml b/dev-support/atlas-docker/docker-compose.atlas-common.yml new file mode 100644 index 00000000000..c1d25ceaf38 --- /dev/null +++ b/dev-support/atlas-docker/docker-compose.atlas-common.yml @@ -0,0 +1,63 @@ +services: + atlas-solr: + build: + context: . + dockerfile: Dockerfile.atlas-solr + image: atlas-solr + container_name: atlas-solr + hostname: atlas-solr.example.com + networks: + - atlas + ports: + - "8983:8983" + + atlas-kafka: + build: + context: . + dockerfile: Dockerfile.atlas-kafka + args: + - KAFKA_VERSION=${KAFKA_VERSION} + - ATLAS_VERSION=${ATLAS_VERSION} + image: atlas-kafka + container_name: atlas-kafka + hostname: atlas-kafka.example.com + stdin_open: true + tty: true + networks: + - atlas + ports: + - "9092:9092" + depends_on: + atlas-zk: + condition: service_started + environment: + - KAFKA_VERSION + - ATLAS_VERSION + + atlas-zk: + build: + context: . + dockerfile: Dockerfile.atlas-zk + image: atlas-zk + container_name: atlas-zk + hostname: atlas-zk.example.com + networks: + - atlas + ports: + - "2181:2181" + + atlas-db: + build: + context: . + dockerfile: Dockerfile.atlas-db + image: atlas-db + container_name: atlas-db + hostname: atlas-db.example.com + networks: + - atlas + healthcheck: + test: 'su -c "pg_isready -q" postgres' + interval: 30s + timeout: 10s + retries: 30 + start_period: 40s diff --git a/dev-support/atlas-docker/docker-compose.atlas-hive.yml b/dev-support/atlas-docker/docker-compose.atlas-hive.yml index 9934a9d12de..7e9bcc7df66 100644 --- a/dev-support/atlas-docker/docker-compose.atlas-hive.yml +++ b/dev-support/atlas-docker/docker-compose.atlas-hive.yml @@ -24,7 +24,7 @@ services: - "10000:10000" depends_on: atlas-db: - condition: service_started + condition: service_healthy atlas-hadoop: condition: service_healthy atlas-zk: @@ -37,19 +37,19 @@ services: - ATLAS_VERSION atlas-db: - build: - context: . - dockerfile: Dockerfile.atlas-db - image: atlas-db - container_name: atlas-db - hostname: atlas-db.example.com - networks: - - atlas - healthcheck: - test: 'su -c "pg_isready -q" postgres' - interval: 10s - timeout: 2s - retries: 30 + extends: + service: atlas-db + file: docker-compose.atlas-common.yml + + atlas-zk: + extends: + service: atlas-zk + file: docker-compose.atlas-common.yml + + atlas-kafka: + extends: + service: atlas-kafka + file: docker-compose.atlas-common.yml networks: atlas: diff --git a/dev-support/atlas-docker/docker-compose.atlas-kafka.yml b/dev-support/atlas-docker/docker-compose.atlas-kafka.yml deleted file mode 100644 index 55e09a0c312..00000000000 --- a/dev-support/atlas-docker/docker-compose.atlas-kafka.yml +++ /dev/null @@ -1,27 +0,0 @@ -services: - atlas-kafka: - build: - context: . - dockerfile: Dockerfile.atlas-kafka - args: - - KAFKA_VERSION=${KAFKA_VERSION} - - ATLAS_VERSION=${ATLAS_VERSION} - image: atlas-kafka - container_name: atlas-kafka - hostname: atlas-kafka.example.com - stdin_open: true - tty: true - networks: - - atlas - ports: - - "9092:9092" - depends_on: - atlas-zk: - condition: service_started - environment: - - KAFKA_VERSION - - ATLAS_VERSION - -networks: - atlas: - name: atlasnw diff --git a/dev-support/atlas-docker/docker-compose.atlas.yml b/dev-support/atlas-docker/docker-compose.atlas.yml index 3b4d7d58e1c..54bb7c164a0 100644 --- a/dev-support/atlas-docker/docker-compose.atlas.yml +++ b/dev-support/atlas-docker/docker-compose.atlas.yml @@ -4,6 +4,7 @@ services: context: . dockerfile: Dockerfile.atlas args: + - ATLAS_BACKEND=${ATLAS_BACKEND} - ATLAS_SERVER_JAVA_VERSION=${ATLAS_SERVER_JAVA_VERSION} - ATLAS_VERSION=${ATLAS_VERSION} image: atlas:latest @@ -18,7 +19,7 @@ services: ports: - "21000:21000" depends_on: - atlas-hbase: + atlas-backend: condition: service_healthy atlas-kafka: condition: service_started @@ -27,34 +28,31 @@ services: atlas-zk: condition: service_started environment: + - ATLAS_BACKEND - ATLAS_SERVER_JAVA_VERSION - ATLAS_VERSION command: - /home/atlas/scripts/atlas.sh - atlas-zk: - build: - context: . - dockerfile: Dockerfile.atlas-zk - image: atlas-zk - container_name: atlas-zk - hostname: atlas-zk.example.com - networks: - - atlas - ports: - - "2181:2181" + atlas-backend: + extends: + service: ${ATLAS_BACKEND} + file: docker-compose.atlas-backend.yml + + atlas-kafka: + extends: + service: atlas-kafka + file: docker-compose.atlas-common.yml atlas-solr: - build: - context: . - dockerfile: Dockerfile.atlas-solr - image: atlas-solr - container_name: atlas-solr - hostname: atlas-solr.example.com - networks: - - atlas - ports: - - "8983:8983" + extends: + service: atlas-solr + file: docker-compose.atlas-common.yml + + atlas-zk: + extends: + service: atlas-zk + file: docker-compose.atlas-common.yml networks: atlas: diff --git a/dev-support/atlas-docker/scripts/atlas.sh b/dev-support/atlas-docker/scripts/atlas.sh index 965bd713776..f0879538337 100755 --- a/dev-support/atlas-docker/scripts/atlas.sh +++ b/dev-support/atlas-docker/scripts/atlas.sh @@ -44,6 +44,35 @@ then echo "" >> /opt/atlas/conf/atlas-application.properties echo "atlas.graph.storage.hbase.compression-algorithm=NONE" >> /opt/atlas/conf/atlas-application.properties + echo "atlas.graph.graph.replace-instance-if-exists=true" >> /opt/atlas/conf/atlas-application.properties + + if [ "${ATLAS_BACKEND}" == "postgres" ] + then + # set RDBMS as backend and entity-audit store + sed -i "s/^atlas.graph.storage.backend=hbase2/# atlas.graph.storage.backend=hbase2/" /opt/atlas/conf/atlas-application.properties + sed -i "s/atlas.EntityAuditRepository.impl=.*$/# atlas.EntityAuditRepository.impl=org.apache.atlas.repository.audit.HBaseBasedAuditRepository/" /opt/atlas/conf/atlas-application.properties + + cat <> /opt/atlas/conf/atlas-application.properties + +atlas.graph.storage.backend=rdbms +atlas.graph.storage.rdbms.jpa.hikari.driverClassName=org.postgresql.Driver +atlas.graph.storage.rdbms.jpa.hikari.jdbcUrl=jdbc:postgresql://atlas-db/atlas +atlas.graph.storage.rdbms.jpa.hikari.username=atlas +atlas.graph.storage.rdbms.jpa.hikari.password=atlasR0cks! +atlas.graph.storage.rdbms.jpa.hikari.maximumPoolSize=40 +atlas.graph.storage.rdbms.jpa.hikari.minimumIdle=5 +atlas.graph.storage.rdbms.jpa.hikari.idleTimeout=300000 +atlas.graph.storage.rdbms.jpa.hikari.connectionTestQuery=select 1 +atlas.graph.storage.rdbms.jpa.hikari.maxLifetime=1800000 +atlas.graph.storage.rdbms.jpa.hikari.connectionTimeout=30000 +atlas.graph.storage.rdbms.jpa.javax.persistence.jdbc.dialect=org.eclipse.persistence.platform.database.PostgreSQLPlatform +atlas.graph.storage.rdbms.jpa.javax.persistence.schema-generation.database.action=create +atlas.graph.storage.rdbms.jpa.javax.persistence.schema-generation.create-database-schemas=true +atlas.graph.storage.rdbms.jpa.javax.persistence.schema-generation.create-source=script +atlas.graph.storage.rdbms.jpa.javax.persistence.schema-generation.create-script-source=META-INF/postgres/create_schema.sql +atlas.EntityAuditRepository.impl=org.apache.atlas.repository.audit.rdbms.RdbmsBasedAuditRepository +EOF + fi chown -R atlas:atlas ${ATLAS_HOME}/ diff --git a/graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasGraph.java b/graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasGraph.java index 1c3e94924af..6bbb077c29b 100644 --- a/graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasGraph.java +++ b/graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasGraph.java @@ -367,4 +367,13 @@ public interface AtlasGraph { * @return */ List getAllEdgesVertices(AtlasVertex vertex); + + /** + * Get the unique key handler for this graph. + * + * @return AtlasUniqueKeyHandler + */ + default AtlasUniqueKeyHandler getUniqueKeyHandler() { + return null; // Default implementation returns null, subclasses can override to provide specific implementation + } } diff --git a/graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasUniqueKeyHandler.java b/graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasUniqueKeyHandler.java new file mode 100644 index 00000000000..a23f19b583f --- /dev/null +++ b/graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasUniqueKeyHandler.java @@ -0,0 +1,33 @@ +/** + * 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.atlas.repository.graphdb; + +public abstract class AtlasUniqueKeyHandler { + public abstract void addUniqueKey(String keyName, Object value, Object elementId, boolean isVertex); + + public abstract void removeUniqueKey(String keyName, Object value, boolean isVertex); + + public abstract void addTypeUniqueKey(String typeName, String keyName, Object value, Object elementId, boolean isVertex); + + public abstract void removeTypeUniqueKey(String typeName, String keyName, Object value, boolean isVertex); + + public abstract void removeUniqueKeysForVertexId(Object vertexId); + + public abstract void removeUniqueKeysForEdgeId(Object edgeId); +} diff --git a/graphdb/janus/pom.xml b/graphdb/janus/pom.xml index 4cd9b2b247c..5360363813e 100644 --- a/graphdb/janus/pom.xml +++ b/graphdb/janus/pom.xml @@ -51,6 +51,17 @@ atlas-graphdb-common ${project.version} + + org.apache.atlas + janusgraph-rdbms + ${project.version} + + + ch.qos.logback + * + + + org.apache.commons diff --git a/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusGraph.java b/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusGraph.java index 7153448a540..e5016eb703a 100644 --- a/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusGraph.java +++ b/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusGraph.java @@ -22,6 +22,7 @@ import com.google.common.collect.Lists; import com.google.common.collect.Maps; import org.apache.atlas.ApplicationProperties; +import org.apache.atlas.AtlasConfiguration; import org.apache.atlas.AtlasErrorCode; import org.apache.atlas.AtlasException; import org.apache.atlas.exception.AtlasBaseException; @@ -37,6 +38,7 @@ import org.apache.atlas.repository.graphdb.AtlasIndexQueryParameter; import org.apache.atlas.repository.graphdb.AtlasPropertyKey; import org.apache.atlas.repository.graphdb.AtlasSchemaViolationException; +import org.apache.atlas.repository.graphdb.AtlasUniqueKeyHandler; import org.apache.atlas.repository.graphdb.AtlasVertex; import org.apache.atlas.repository.graphdb.GraphIndexQueryParameters; import org.apache.atlas.repository.graphdb.GremlinVersion; @@ -44,6 +46,7 @@ import org.apache.atlas.repository.graphdb.utils.IteratorToIterableAdapter; import org.apache.atlas.type.AtlasType; import org.apache.commons.configuration.Configuration; +import org.apache.hadoop.util.StringUtils; import org.apache.tinkerpop.gremlin.groovy.jsr223.GremlinGroovyScriptEngine; import org.apache.tinkerpop.gremlin.jsr223.DefaultImportCustomizer; import org.apache.tinkerpop.gremlin.process.traversal.P; @@ -104,6 +107,7 @@ public class AtlasJanusGraph implements AtlasGraph multiProperties = new HashSet<>(); private final StandardJanusGraph janusGraph; + private final AtlasUniqueKeyHandler uniqueKeyHandler; private final ThreadLocal scriptEngine = ThreadLocal.withInitial(() -> { DefaultImportCustomizer.Builder builder = DefaultImportCustomizer.build() @@ -131,6 +135,12 @@ public AtlasJanusGraph(JanusGraph graphInstance) { multiProperties.add(key.name()); } } + + if (StringUtils.equalsIgnoreCase(AtlasConfiguration.STORAGE_BACKEND_TYPE.getString(), "rdbms")) { + uniqueKeyHandler = new AtlasJanusRdbmsUniqueKeyHandler(); + } else { + uniqueKeyHandler = null; + } } finally { if (mgmt != null) { mgmt.rollback(); @@ -431,6 +441,12 @@ public List getAllEdgesVertices(AtlasVertex vertex) { return resultList; } + @Override + public AtlasUniqueKeyHandler getUniqueKeyHandler() { + return uniqueKeyHandler; + } + + public Iterable> wrapVertices(Iterable it) { return Iterables.transform(it, (Function>) input -> GraphDbObjectFactory.createVertex(AtlasJanusGraph.this, input)); } diff --git a/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusGraphDatabase.java b/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusGraphDatabase.java index e4b4f0ff7d4..8648fb6c137 100644 --- a/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusGraphDatabase.java +++ b/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusGraphDatabase.java @@ -40,6 +40,7 @@ import org.janusgraph.diskstorage.StandardStoreManager; import org.janusgraph.diskstorage.es.ElasticSearch7Index; import org.janusgraph.diskstorage.hbase.HBaseStoreManager; +import org.janusgraph.diskstorage.rdbms.RdbmsStoreManager; import org.janusgraph.diskstorage.solr.Solr6Index; import org.janusgraph.graphdb.database.serialize.attribute.SerializableSerializer; import org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry; @@ -336,6 +337,31 @@ private static void addHBase2Support() { } } + private static void addRdbmsSupport() { + try { + Field field = StandardStoreManager.class.getDeclaredField("ALL_MANAGER_CLASSES"); + + field.setAccessible(true); + + Field modifiersField = Field.class.getDeclaredField("modifiers"); + + modifiersField.setAccessible(true); + modifiersField.setInt(field, field.getModifiers() & ~Modifier.FINAL); + + Map customMap = new HashMap<>(StandardStoreManager.getAllManagerClasses()); + + customMap.put("rdbms", RdbmsStoreManager.class.getName()); + + ImmutableMap immap = ImmutableMap.copyOf(customMap); + + field.set(null, immap); + + LOG.debug("Injected RDBMS support - {}", RdbmsStoreManager.class.getName()); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + private static void addSolr6Index() { try { Field field = StandardIndexProvider.class.getDeclaredField("ALL_MANAGER_CLASSES"); @@ -454,6 +480,7 @@ private static void stopEmbeddedSolr() throws AtlasException { static { addHBase2Support(); + addRdbmsSupport(); addSolr6Index(); diff --git a/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusRdbmsUniqueKeyHandler.java b/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusRdbmsUniqueKeyHandler.java new file mode 100644 index 00000000000..103cd9311f1 --- /dev/null +++ b/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusRdbmsUniqueKeyHandler.java @@ -0,0 +1,59 @@ +/** + * 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.atlas.repository.graphdb.janus; + +import org.apache.atlas.repository.graphdb.AtlasUniqueKeyHandler; +import org.janusgraph.diskstorage.rdbms.RdbmsUniqueKeyHandler; + +public class AtlasJanusRdbmsUniqueKeyHandler extends AtlasUniqueKeyHandler { + private final RdbmsUniqueKeyHandler uniqueKeyHandler; + + public AtlasJanusRdbmsUniqueKeyHandler() { + uniqueKeyHandler = new RdbmsUniqueKeyHandler(); + } + + @Override + public void addUniqueKey(String keyName, Object value, Object elementId, boolean isVertex) { + uniqueKeyHandler.addUniqueKey(keyName, value, elementId, isVertex); + } + + @Override + public void removeUniqueKey(String keyName, Object value, boolean isVertex) { + uniqueKeyHandler.removeUniqueKey(keyName, value, isVertex); + } + + @Override + public void addTypeUniqueKey(String typeName, String keyName, Object value, Object elementId, boolean isVertex) { + uniqueKeyHandler.addTypeUniqueKey(typeName, keyName, value, elementId, isVertex); + } + + @Override + public void removeTypeUniqueKey(String typeName, String keyName, Object value, boolean isVertex) { + uniqueKeyHandler.removeTypeUniqueKey(typeName, keyName, value, isVertex); + } + + @Override + public void removeUniqueKeysForVertexId(Object vertexId) { + uniqueKeyHandler.removeUniqueKeysForVertexId(vertexId); + } + + @Override + public void removeUniqueKeysForEdgeId(Object edgeId) { + uniqueKeyHandler.removeUniqueKeysForEdgeId(edgeId); + } +} diff --git a/graphdb/janusgraph-rdbms/pom.xml b/graphdb/janusgraph-rdbms/pom.xml new file mode 100644 index 00000000000..663f09cd0c1 --- /dev/null +++ b/graphdb/janusgraph-rdbms/pom.xml @@ -0,0 +1,91 @@ + + + + 4.0.0 + + + org.apache.atlas + atlas-graphdb + 3.0.0-SNAPSHOT + + + janusgraph-rdbms + JanusGraph RDBMS backend store + + + 4.0.3 + + + + + com.zaxxer + HikariCP + ${HikariCP.version} + + + org.eclipse.persistence + eclipselink + ${eclipse.jpa.version} + + + + org.eclipse.persistence + javax.persistence + ${javax.persistence.version} + + + + org.janusgraph + janusgraph-core + ${janusgraph.version} + + + com.rabbitmq + amqp-client + + + org.noggit + noggit + + + + + + + + + true + ${basedir}/src/main/resources + + + + + ${basedir}/src/test/resources + + + + + + + postgres + Postgres Driver + https://mvnrepository.com/artifact/org.postgresql/postgresql + + + diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/RdbmsTransaction.java b/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/RdbmsTransaction.java new file mode 100644 index 00000000000..b5943740846 --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/RdbmsTransaction.java @@ -0,0 +1,54 @@ +/** + * 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.atlas.repository.audit.rdbms; + +import org.janusgraph.diskstorage.rdbms.RdbmsStoreManager; + +import javax.persistence.EntityManager; +import javax.persistence.EntityTransaction; + +public class RdbmsTransaction implements AutoCloseable { + private final EntityManager em; + private final EntityTransaction trx; + + public RdbmsTransaction() { + this.em = RdbmsStoreManager.getInstance().getDaoManager().createEntityManager(); + this.trx = em.getTransaction(); + + this.trx.begin(); + } + + public EntityManager getEntityManager() { + return em; + } + + public void commit() { + if (trx.isActive()) { + trx.commit(); + } + } + + @Override + public void close() { + if (trx.isActive()) { + trx.rollback(); + } + + em.close(); + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/dao/BaseDao.java b/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/dao/BaseDao.java new file mode 100644 index 00000000000..d3a28869fc7 --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/dao/BaseDao.java @@ -0,0 +1,35 @@ +/* + * 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.atlas.repository.audit.rdbms.dao; + +import javax.persistence.EntityManager; + +public abstract class BaseDao { + protected final EntityManager em; + + protected BaseDao(EntityManager em) { + this.em = em; + } + + public T create(T obj) { + em.persist(obj); + + return obj; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/dao/DbEntityAuditDao.java b/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/dao/DbEntityAuditDao.java new file mode 100644 index 00000000000..4f2bebbb48f --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/dao/DbEntityAuditDao.java @@ -0,0 +1,71 @@ +/** + * 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.atlas.repository.audit.rdbms.dao; + +import org.apache.atlas.repository.audit.rdbms.entity.DbEntityAudit; + +import javax.persistence.EntityManager; +import javax.persistence.NoResultException; + +import java.util.Collections; +import java.util.List; + +public class DbEntityAuditDao extends BaseDao { + public DbEntityAuditDao(EntityManager em) { + super(em); + } + + public List getByEntityIdActionStartTimeStartIdx(String entityId, int action, long eventTimeStart, int eventIdxStart, int maxResults) { + try { + return em.createNamedQuery("DbEntityAudit.getByEntityIdActionStartTimeStartIdx", DbEntityAudit.class) + .setParameter("entityId", entityId) + .setParameter("action", action) + .setParameter("eventTimeStart", eventTimeStart) + .setParameter("eventIdxStart", eventIdxStart) + .setMaxResults(maxResults) + .getResultList(); + } catch (NoResultException excp) { + // ignore + } + + return Collections.emptyList(); + } + + public List getByEntityIdAction(String entityId, Integer action, int startIdx, int maxResults) { + try { + if (action == null) { + return em.createNamedQuery("DbEntityAudit.getByEntityId", DbEntityAudit.class) + .setParameter("entityId", entityId) + .setFirstResult(startIdx) + .setMaxResults(maxResults) + .getResultList(); + } else { + return em.createNamedQuery("DbEntityAudit.getByEntityIdAction", DbEntityAudit.class) + .setParameter("entityId", entityId) + .setParameter("action", action) + .setFirstResult(startIdx) + .setMaxResults(maxResults) + .getResultList(); + } + } catch (NoResultException excp) { + // ignore + } + + return Collections.emptyList(); + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/entity/DbEntityAudit.java b/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/entity/DbEntityAudit.java new file mode 100644 index 00000000000..e5c9005e550 --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/entity/DbEntityAudit.java @@ -0,0 +1,180 @@ +/** + * 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.atlas.repository.audit.rdbms.entity; + +import javax.persistence.Cacheable; +import javax.persistence.Column; +import javax.persistence.Entity; +import javax.persistence.GeneratedValue; +import javax.persistence.GenerationType; +import javax.persistence.Id; +import javax.persistence.Index; +import javax.persistence.Lob; +import javax.persistence.SequenceGenerator; +import javax.persistence.Table; + +import java.util.Objects; + +/** + * RDBMS representation of a JanusGraph Column - name/value pair in a JanusGraph key + * + * @author Madhan Neethiraj <madhan@apache.org> + */ +@Entity +@Cacheable(false) +@Table(name = "atlas_entity_audit", + indexes = {@Index(name = "atlas_entity_audit_idx_entity_id", columnList = "entity_id"), + @Index(name = "atlas_entity_audit_idx_event_time", columnList = "event_time"), + @Index(name = "atlas_entity_audit_idx_entity_id_event_time", columnList = "entity_id,event_time"), + @Index(name = "atlas_entity_audit_idx_user_name", columnList = "user_name")}) +public class DbEntityAudit implements java.io.Serializable { + @Id + @SequenceGenerator(name = "atlas_entity_audit_seq", sequenceName = "atlas_entity_audit_seq", allocationSize = 1) + @GeneratedValue(strategy = GenerationType.AUTO, generator = "atlas_entity_audit_seq") + @Column(name = "id") + protected Long id; + + @Column(name = "entity_id", nullable = false, length = 64) + protected String entityId; + + @Column(name = "event_time", nullable = false) + protected long eventTime; + + @Column(name = "event_idx", nullable = false) + protected int eventIndex; + + @Column(name = "user_name", nullable = false, length = 64) + protected String user; + + @Column(name = "operation", nullable = false) + protected int action; + + @Column(name = "details") + @Lob + protected String details; + + @Column(name = "entity") + @Lob + protected String entity; + + @Column(name = "audit_type") + protected int auditType; + + public Long getId() { + return id; + } + + public void setId(Long id) { + this.id = id; + } + + public String getEntityId() { + return entityId; + } + + public void setEntityId(String entityId) { + this.entityId = entityId; + } + + public long getEventTime() { + return eventTime; + } + + public void setEventTime(long eventTime) { + this.eventTime = eventTime; + } + + public int getEventIndex() { + return eventIndex; + } + + public void setEventIndex(int eventIndex) { + this.eventIndex = eventIndex; + } + + public String getUser() { + return user; + } + + public void setUser(String user) { + this.user = user; + } + + public int getAction() { + return action; + } + + public void setAction(int action) { + this.action = action; + } + + public String getDetails() { + return details; + } + + public void setDetails(String details) { + this.details = details; + } + + public String getEntity() { + return entity; + } + + public void setEntity(String entity) { + this.entity = entity; + } + + public int getAuditType() { + return auditType; + } + + public void setAuditType(int auditType) { + this.auditType = auditType; + } + + @Override + public int hashCode() { + return Objects.hash(id, entityId, eventTime, eventIndex, user, action, details, entity, auditType); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } else if (obj instanceof DbEntityAudit && getClass() == obj.getClass()) { + DbEntityAudit other = (DbEntityAudit) obj; + + return Objects.equals(id, other.id) && + Objects.equals(entityId, other.entityId) && + eventTime == other.eventTime && + eventIndex == other.eventIndex && + Objects.equals(user, other.user) && + action == other.action && + Objects.equals(details, other.details) && + Objects.equals(entity, other.entity) && + auditType == other.auditType; + } + + return false; + } + + @Override + public String toString() { + return "DbEntityAudit(id=" + id + ", entityId=" + entityId + ", eventTime=" + eventTime + ", eventIndex=" + eventIndex + ", user=" + user + ", action=" + action + ", details=" + details + ", entity=" + entity + ", auditType=" + auditType + ")"; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/JanusColumnValue.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/JanusColumnValue.java new file mode 100644 index 00000000000..bd9b0f8233a --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/JanusColumnValue.java @@ -0,0 +1,51 @@ +/** + * 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.janusgraph.diskstorage.rdbms; + +import org.janusgraph.diskstorage.StaticBuffer; +import org.janusgraph.diskstorage.util.StaticArrayBuffer; + +/** + * ColumnValue stored in RDBMS + * + */ +public class JanusColumnValue { + private final byte[] column; + private final byte[] value; + + public JanusColumnValue(byte[] column, byte[] value) { + this.column = column; + this.value = value; + } + + public byte[] getColumn() { + return column; + } + + public byte[] getValue() { + return value; + } + + public StaticBuffer getColumnAsStaticBuffer() { + return StaticArrayBuffer.of(column); + } + + public StaticBuffer getValueAsStaticBuffer() { + return StaticArrayBuffer.of(value); + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsStore.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsStore.java new file mode 100644 index 00000000000..c3df6851e8f --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsStore.java @@ -0,0 +1,329 @@ +/** + * 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.janusgraph.diskstorage.rdbms; + +import org.janusgraph.diskstorage.BackendException; +import org.janusgraph.diskstorage.Entry; +import org.janusgraph.diskstorage.EntryList; +import org.janusgraph.diskstorage.EntryMetaData; +import org.janusgraph.diskstorage.StaticBuffer; +import org.janusgraph.diskstorage.keycolumnvalue.KeyColumnValueStore; +import org.janusgraph.diskstorage.keycolumnvalue.KeyIterator; +import org.janusgraph.diskstorage.keycolumnvalue.KeyRangeQuery; +import org.janusgraph.diskstorage.keycolumnvalue.KeySliceQuery; +import org.janusgraph.diskstorage.keycolumnvalue.KeySlicesIterator; +import org.janusgraph.diskstorage.keycolumnvalue.MultiSlicesQuery; +import org.janusgraph.diskstorage.keycolumnvalue.SliceQuery; +import org.janusgraph.diskstorage.keycolumnvalue.StoreTransaction; +import org.janusgraph.diskstorage.rdbms.dao.DaoManager; +import org.janusgraph.diskstorage.rdbms.dao.JanusColumnDao; +import org.janusgraph.diskstorage.rdbms.dao.JanusKeyDao; +import org.janusgraph.diskstorage.rdbms.dao.JanusStoreDao; +import org.janusgraph.diskstorage.rdbms.entity.JanusKey; +import org.janusgraph.diskstorage.rdbms.entity.JanusStore; +import org.janusgraph.diskstorage.util.StaticArrayEntry; +import org.janusgraph.diskstorage.util.StaticArrayEntryList; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; + +/** + * KeyColumnValue store backed by RDBMS + * + */ +public class RdbmsStore implements KeyColumnValueStore { + private static final Logger LOG = LoggerFactory.getLogger(RdbmsStore.class); + + private static final int STORE_CREATE_MAX_ATTEMPTS = 10; + private static final int STORE_CREATE_RETRY_DELAY_MS = 100; + private static final int KEY_CREATE_MAX_ATTEMPTS = 10; + private static final int KEY_CREATE_RETRY_DELAY_MS = 100; + + private final String name; + private final DaoManager daoManager; + private final EntryMetaData[] entryMetaData; + private Long storeId; + + public RdbmsStore(String name, RdbmsStoreManager storeManager) { + LOG.info("RdbmsStore(name={})", name); + + this.name = name; + this.daoManager = storeManager.getDaoManager(); + this.entryMetaData = storeManager.getMetaDataSchema(name); + this.storeId = null; + } + + @Override + public EntryList getSlice(KeySliceQuery query, StoreTransaction trx) { + LOG.debug("==> RdbmsStore.getSlice(name={}, query={}, trx={})", name, query, trx); + + final EntryList ret; + + if (isStorePresent(trx)) { + JanusColumnDao dao = new JanusColumnDao((RdbmsTransaction) trx, this); + Long keyId = getKeyIdOrCreate(toBytes(query.getKey()), trx); + byte[] sliceStart = toBytes(query.getSliceStart()); + byte[] sliceEnd = toBytes(query.getSliceEnd()); + List entries = dao.getColumns(keyId, sliceStart, sliceEnd, query.getLimit()); + + if (entries != null && !entries.isEmpty()) { + ret = StaticArrayEntryList.ofStaticBuffer(entries, toEntry); + } else { + ret = EntryList.EMPTY_LIST; + } + } else { + ret = EntryList.EMPTY_LIST; + } + + LOG.debug("<== RdbmsStore.getSlice(name={}, query={}, trx={}): ret={}", name, query, trx, ret.size()); + + return ret; + } + + @Override + public Map getSlice(List keys, SliceQuery query, StoreTransaction trx) { + LOG.debug("==> RdbmsStore.getSlice(name={}, len(keys)={}, query={}, trx={})", name, keys.size(), query, trx); + + final Map ret; + + if (isStorePresent(trx)) { + ret = new TreeMap<>(); + + for (StaticBuffer key : keys) { + ret.put(key, getSlice(new KeySliceQuery(key, query), trx)); + } + } else { + ret = Collections.emptyMap(); + } + + LOG.debug("<== RdbmsStore.getSlice(name={}, len(keys)={}, query={}, trx={}): ret={}", name, keys.size(), query, trx, ret); + + return ret; + } + + @Override + public void mutate(StaticBuffer key, List additions, List deletions, StoreTransaction trx) { + LOG.debug("==> RdbmsStore.mutate(name={}, key={}, additions={}, deletions={}, trx={})", name, key, additions, deletions, trx); + + byte[] keyName = toBytes(key); + long keyId = getKeyIdOrCreate(keyName, trx); + JanusColumnDao columnDao = new JanusColumnDao((RdbmsTransaction) trx, this); + + for (StaticBuffer column : deletions) { + byte[] columnName = toBytes(column); + + columnDao.remove(keyId, columnName); + } + + for (Entry entry : additions) { + columnDao.addOrUpdate(keyId, toBytes(entry.getColumn()), toBytes(entry.getValue())); + } + + LOG.debug("<== RdbmsStore.mutate(name={}, key={}, additions={}, deletions={}, trx={})", name, key, additions, deletions, trx); + } + + @Override + public void acquireLock(StaticBuffer key, StaticBuffer column, StaticBuffer expectedValue, StoreTransaction trx) { + LOG.debug("RdbmsStore.acquireLock(key={}, column={}, expectedValue={}, trx={}): UnsupportedOperation", key, column, expectedValue, trx); + + throw new UnsupportedOperationException(); + } + + @Override + public KeyIterator getKeys(KeyRangeQuery query, StoreTransaction trx) { + LOG.debug("==> RdbmsStore.getKeys(name={}, query={}, trx={})", name, query, trx); + + final KeyIterator ret; + + if (isStorePresent(trx)) { + JanusColumnDao dao = new JanusColumnDao((RdbmsTransaction) trx, this); + + ret = dao.getKeysByKeyAndColumnRange(this.storeId, toBytes(query.getKeyStart()), toBytes(query.getKeyEnd()), toBytes(query.getSliceStart()), toBytes(query.getSliceEnd()), query.getLimit()); + } else { + ret = JanusColumnDao.EMPTY_KEY_ITERATOR; + } + + LOG.debug("<== RdbmsStore.debug(name={}, query={}, trx={}): ret={}", name, query, trx, ret); + + return ret; + } + + @Override + public KeyIterator getKeys(SliceQuery query, StoreTransaction trx) { + LOG.debug("==> RdbmsStore.getKeys(name={}, query={}, trx={})", name, query, trx); + + final KeyIterator ret; + + if (isStorePresent(trx)) { + JanusColumnDao dao = new JanusColumnDao((RdbmsTransaction) trx, this); + + ret = dao.getKeysByColumnRange(this.storeId, toBytes(query.getSliceStart()), toBytes(query.getSliceEnd()), query.getLimit()); + } else { + ret = JanusColumnDao.EMPTY_KEY_ITERATOR; + } + + LOG.debug("<== RdbmsStore.debug(name={}, query={}, trx={}): ret={}", name, query, trx, ret); + + return ret; + } + + @Override + public KeySlicesIterator getKeys(MultiSlicesQuery query, StoreTransaction trx) { + LOG.debug("RdbmsStore.getKeys(query={}, trx={}): UnsupportedOperation", query, trx); + + throw new UnsupportedOperationException(); + } + + @Override + public String getName() { + return name; + } + + @Override + public void close() throws BackendException { + LOG.debug("RdbmsStore.close(name={})", name); + } + + private boolean isStorePresent(StoreTransaction trx) { + Long storeId = this.storeId; + + if (storeId == null) { + JanusStoreDao storeDao = new JanusStoreDao((RdbmsTransaction) trx); + + storeId = storeDao.getIdByName(name); + + if (storeId != null) { + this.storeId = storeId; + } + } + + return storeId != null; + } + + private static byte[] toBytes(StaticBuffer val) { + return val == null ? null : val.as(StaticBuffer.ARRAY_FACTORY); + } + + private Long getStoreIdOrCreate(StoreTransaction trx) { + Long ret = this.storeId; + + if (ret == null) { + JanusStoreDao dao = new JanusStoreDao((RdbmsTransaction) trx); + + ret = dao.getIdByName(name); + + for (int attempt = 1; ret == null; attempt++) { + try (RdbmsTransaction trx2 = new RdbmsTransaction(trx.getConfiguration(), daoManager)) { + JanusStoreDao dao2 = new JanusStoreDao(trx2); + JanusStore store = dao2.create(new JanusStore(name)); + + trx2.commit(); + + ret = store != null ? store.getId() : null; + + LOG.debug("attempt #{}: created store(name={}): id={}", attempt, name, ret); + } catch (IOException excp) { + LOG.error("attempt #{}: failed to create store(name={})", attempt, name, excp); + } + + if (ret != null || attempt >= STORE_CREATE_MAX_ATTEMPTS) { + break; + } + + try { + Thread.sleep(STORE_CREATE_RETRY_DELAY_MS); + } catch (InterruptedException excp) { + LOG.error("Thread interrupted while waiting to retry store creation(name={})", name, excp); + Thread.currentThread().interrupt(); + } + } + + if (ret != null) { + this.storeId = ret; + } else { + LOG.error("Failed to create store(name={}) after {} attempts", name, STORE_CREATE_MAX_ATTEMPTS); + } + } + + return ret; + } + + private Long getKeyIdOrCreate(byte[] key, StoreTransaction trx) { + Long storeId = getStoreIdOrCreate(trx); + JanusKeyDao dao = new JanusKeyDao((RdbmsTransaction) trx); + Long ret = dao.getIdByStoreIdAndName(storeId, key); + + for (int attempt = 1; ret == null; attempt++) { + try (RdbmsTransaction trx2 = new RdbmsTransaction(trx.getConfiguration(), daoManager)) { + JanusKeyDao dao2 = new JanusKeyDao(trx2); + JanusKey createdKey = dao2.create(new JanusKey(storeId, key)); + + trx2.commit(); + + ret = createdKey != null ? createdKey.getId() : null; + + LOG.debug("attempt #{}: created key(storeId={}, key={}): id={}", attempt, storeId, key, ret); + } catch (Throwable t) { + LOG.error("attempt #{}: failed to create key(storeId={}, key.length={}, key={}): {}", attempt, storeId, key.length, new String(key), t); + } + + if (ret != null || attempt >= KEY_CREATE_MAX_ATTEMPTS) { + break; + } + + try { + Thread.sleep(KEY_CREATE_RETRY_DELAY_MS); + } catch (InterruptedException excp) { + LOG.error("Thread interrupted while waiting to retry key creation(storeId={}, key={})", storeId, key, excp); + Thread.currentThread().interrupt(); + } + } + + return ret; + } + + public final StaticArrayEntry.GetColVal toEntry = + new StaticArrayEntry.GetColVal() { + @Override + public StaticBuffer getColumn(JanusColumnValue columnValue) { + return columnValue.getColumnAsStaticBuffer(); + } + + @Override + public StaticBuffer getValue(JanusColumnValue columnValue) { + return columnValue.getValueAsStaticBuffer(); + } + + @Override + public EntryMetaData[] getMetaSchema(JanusColumnValue janusColumnValue) { + return entryMetaData; + } + + @Override + public Object getMetaData(JanusColumnValue janusColumnValue, EntryMetaData entryMetaData) { + LOG.debug("RdbmsStore.getMetaData(janusColumnValue={}, entryMetaData={}): UnsupportedOperation", janusColumnValue, entryMetaData); + + return new UnsupportedOperationException(); + } + }; +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsStoreManager.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsStoreManager.java new file mode 100644 index 00000000000..95117a1700f --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsStoreManager.java @@ -0,0 +1,184 @@ +/** + * 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.janusgraph.diskstorage.rdbms; + +import org.janusgraph.diskstorage.BackendException; +import org.janusgraph.diskstorage.BaseTransactionConfig; +import org.janusgraph.diskstorage.StaticBuffer; +import org.janusgraph.diskstorage.StoreMetaData; +import org.janusgraph.diskstorage.common.AbstractStoreManager; +import org.janusgraph.diskstorage.configuration.ConfigNamespace; +import org.janusgraph.diskstorage.configuration.Configuration; +import org.janusgraph.diskstorage.keycolumnvalue.KCVMutation; +import org.janusgraph.diskstorage.keycolumnvalue.KeyColumnValueStore; +import org.janusgraph.diskstorage.keycolumnvalue.KeyColumnValueStoreManager; +import org.janusgraph.diskstorage.keycolumnvalue.KeyRange; +import org.janusgraph.diskstorage.keycolumnvalue.StandardStoreFeatures; +import org.janusgraph.diskstorage.keycolumnvalue.StoreFeatures; +import org.janusgraph.diskstorage.keycolumnvalue.StoreTransaction; +import org.janusgraph.diskstorage.rdbms.dao.DaoManager; +import org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration; +import org.janusgraph.graphdb.configuration.PreInitializeConfigOptions; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Storage Manager for RDBMS + * + */ +@PreInitializeConfigOptions +public class RdbmsStoreManager extends AbstractStoreManager implements KeyColumnValueStoreManager { + private static final Logger LOG = LoggerFactory.getLogger(RdbmsStoreManager.class); + + private static final String NAME = "rdbms"; + + public static final ConfigNamespace RDBMS_NS = new ConfigNamespace(GraphDatabaseConfiguration.STORAGE_NS, NAME, "RDBMS configuration options"); + public static final ConfigNamespace JPA_CONFIG_NS = new ConfigNamespace(RDBMS_NS, "jpa", "JPA configurations", true); + + private static RdbmsStoreManager sInstance; + + private final StandardStoreFeatures features; + private final Map stores; + private final DaoManager daoManager; + + public RdbmsStoreManager(Configuration config) { + super(config); + + features = new StandardStoreFeatures.Builder() + .orderedScan(true) + .unorderedScan(true) + .multiQuery(true) + .transactional(true) + .keyConsistent(GraphDatabaseConfiguration.buildGraphConfiguration()) + .keyOrdered(true) + .batchMutation(true) + .build(); + stores = new HashMap<>(); + daoManager = new DaoManager(config.getSubset(JPA_CONFIG_NS)); + + sInstance = this; + + LOG.info("RdbmsStoreManager()"); + } + + public static RdbmsStoreManager getInstance() { + return sInstance; + } + + public DaoManager getDaoManager() { + return daoManager; + } + + @Override + public KeyColumnValueStore openDatabase(String name, StoreMetaData.Container container) throws BackendException { + LOG.debug("==> RdbmsStoreManager.openDatabase(name={})", name); + + RdbmsStore ret = stores.get(name); + + if (ret == null) { + synchronized (this) { + ret = stores.computeIfAbsent(name, k -> new RdbmsStore(name, this)); + } + } + + LOG.debug("<== RdbmsStoreManager.openDatabase(name={})", name); + + return ret; + } + + @Override + public void mutateMany(Map> storeMutations, StoreTransaction trx) throws BackendException { + LOG.debug("==> RdbmsStoreManager.mutateMany(numStores={}, trx={})", storeMutations.size(), trx); + + int numMutations = 0; + int numAdditions = 0; + int numDeletions = 0; + + for (Map.Entry> storeEntry : storeMutations.entrySet()) { + String storeName = storeEntry.getKey(); + KeyColumnValueStore store = this.openDatabase(storeName); + + for (Map.Entry mutationEntry : storeEntry.getValue().entrySet()) { + StaticBuffer key = mutationEntry.getKey(); + KCVMutation mutation = mutationEntry.getValue(); + + numMutations++; + numAdditions += mutation.getAdditions().size(); + numDeletions += mutation.getDeletions().size(); + + store.mutate(key, mutation.getAdditions(), mutation.getDeletions(), trx); + } + } + + LOG.debug("<== RdbmsStoreManager.mutateMany(numStores={}, numMutations={}, numAdditions={}, numDeletions={})", storeMutations.size(), numMutations, numAdditions, numDeletions); + } + + @Override + public StoreTransaction beginTransaction(BaseTransactionConfig baseTransactionConfig) throws BackendException { + LOG.debug("RdbmsStoreManager.beginTransaction()"); + + return new RdbmsTransaction(baseTransactionConfig, this.daoManager); + } + + @Override + public void close() throws BackendException { + LOG.debug("RdbmsStoreManager.close()"); + + synchronized (this) { + for (RdbmsStore store : stores.values()) { + store.close(); + } + + stores.clear(); + daoManager.close(); + } + } + + @Override + public void clearStorage() throws BackendException { + LOG.debug("RdbmsStoreManager.clearStorage(): UnsupportedOperation"); + + throw new UnsupportedOperationException(); + } + + @Override + public boolean exists() throws BackendException { + return true; + } + + @Override + public StoreFeatures getFeatures() { + return features; + } + + @Override + public String getName() { + return NAME; + } + + @Override + public List getLocalKeyPartition() throws BackendException { + LOG.debug("RdbmsStoreManager.getLocalKeyPartition(): UnsupportedOperation"); + + throw new UnsupportedOperationException(); + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsTransaction.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsTransaction.java new file mode 100644 index 00000000000..b692677d3bb --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsTransaction.java @@ -0,0 +1,151 @@ +/** + * 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.janusgraph.diskstorage.rdbms; + +import org.janusgraph.diskstorage.BaseTransactionConfig; +import org.janusgraph.diskstorage.common.AbstractStoreTransaction; +import org.janusgraph.diskstorage.rdbms.dao.DaoManager; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.persistence.EntityManager; +import javax.persistence.EntityTransaction; + +import java.io.Closeable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Store transaction implementation for RDBMS + * + */ +public class RdbmsTransaction extends AbstractStoreTransaction implements Closeable { + private static final Logger LOG = LoggerFactory.getLogger(RdbmsTransaction.class); + + private static final ThreadLocal> ACTIVE_TRANSACTIONS = ThreadLocal.withInitial(ArrayList::new); + private static final ThreadLocal TRANSACTION_COUNTS = ThreadLocal.withInitial(AtomicLong::new); + + private final EntityManager em; + private final EntityTransaction trx; + + public RdbmsTransaction(BaseTransactionConfig trxConfig, DaoManager daoManager) { + super(trxConfig); + + em = daoManager.createEntityManager(); + trx = em.getTransaction(); + + trx.begin(); + + long count = TRANSACTION_COUNTS.get().incrementAndGet(); + + if (count % 100 == 0) { + LOG.debug("RDBMS transactions count for thread[{}]: {}", Thread.currentThread().getName(), count); + } + + addToActiveTransactions(); + } + + public EntityManager getEntityManager() { + return em; + } + + @Override + public void commit() { + LOG.trace("==> RdbmsTransaction.commit()"); + + try { + if (trx.isActive()) { + trx.commit(); + } + } finally { + removeFromActiveTransactions(); + em.close(); + } + + LOG.trace("<== RdbmsTransaction.commit()"); + } + + @Override + public void rollback() { + LOG.trace("==> RdbmsTransaction.rollback()"); + + try { + if (trx.isActive()) { + trx.rollback(); + } + } finally { + removeFromActiveTransactions(); + em.close(); + } + + LOG.trace("<== RdbmsTransaction.rollback()"); + } + + @Override + public void close() throws IOException { + LOG.trace("==> RdbmsTransaction.close()"); + + IOException ret = null; + + if (trx.isActive()) { + try { + trx.rollback(); + } catch (Exception excp) { + ret = new IOException(excp); + } + } + + if (em.isOpen()) { + try { + em.close(); + } catch (Exception excp) { + if (ret != null) { + ret = new IOException(excp); + } + } + } + + removeFromActiveTransactions(); + + if (ret != null) { + throw ret; + } + + LOG.trace("<== RdbmsTransaction.close()"); + } + + static RdbmsTransaction getActiveTransaction() { + List trxList = ACTIVE_TRANSACTIONS.get(); + + return trxList.isEmpty() ? null : trxList.get(trxList.size() - 1); + } + + private void addToActiveTransactions() { + List trxList = ACTIVE_TRANSACTIONS.get(); + + if (!trxList.contains(this)) { + trxList.add(this); + } + } + + private void removeFromActiveTransactions() { + ACTIVE_TRANSACTIONS.get().remove(this); + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsUniqueKeyHandler.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsUniqueKeyHandler.java new file mode 100644 index 00000000000..3dce8457dfb --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/RdbmsUniqueKeyHandler.java @@ -0,0 +1,150 @@ +/** + * 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.janusgraph.diskstorage.rdbms; + +import org.janusgraph.graphdb.relations.RelationIdentifier; + +public class RdbmsUniqueKeyHandler { + private static final String SQL_INSERT_UNIQUE_VERTEX_KEY = "INSERT INTO janus_unique_vertex_key (vertex_id, key_name, val) VALUES (?, ?, ?)"; + private static final String SQL_DELETE_UNIQUE_VERTEX_KEY = "DELETE FROM janus_unique_vertex_key WHERE key_name = ? AND val = ?"; + private static final String SQL_INSERT_UNIQUE_VERTEX_TYPE_KEY = "INSERT INTO janus_unique_vertex_type_key (vertex_id, type_name, key_name, val) VALUES (?, ?, ?, ?)"; + private static final String SQL_DELETE_UNIQUE_VERTEX_TYPE_KEY = "DELETE FROM janus_unique_vertex_type_key WHERE type_name = ? AND key_name = ? AND val = ?"; + private static final String SQL_DELETE_UNIQUE_VERTEX_KEY_BY_VERTEX_ID = "DELETE FROM janus_unique_vertex_key WHERE vertex_id = ?"; + private static final String SQL_DELETE_UNIQUE_VERTEX_TYPE_KEY_BY_VERTEX_ID = "DELETE FROM janus_unique_vertex_type_key WHERE vertex_id = ?"; + + private static final String SQL_INSERT_UNIQUE_EDGE_KEY = "INSERT INTO janus_unique_edge_key (edge_id, key_name, val) VALUES (?, ?, ?)"; + private static final String SQL_DELETE_UNIQUE_EDGE_KEY = "DELETE FROM janus_unique_edge_key WHERE key_name = ? AND val = ?"; + private static final String SQL_INSERT_UNIQUE_EDGE_TYPE_KEY = "INSERT INTO janus_unique_edge_type_key (edge_id, type_name, key_name, val) VALUES (?, ?, ?, ?)"; + private static final String SQL_DELETE_UNIQUE_EDGE_TYPE_KEY = "DELETE FROM janus_unique_edge_type_key WHERE type_name = ? AND key_name = ? AND val = ?"; + private static final String SQL_DELETE_UNIQUE_EDGE_KEY_BY_EDGE_ID = "DELETE FROM janus_unique_edge_key WHERE edge_id = ?"; + private static final String SQL_DELETE_UNIQUE_EDGE_TYPE_KEY_BY_EDGE_ID = "DELETE FROM janus_unique_edge_type_key WHERE edge_id = ?"; + + public void addUniqueKey(String keyName, Object value, Object elementId, boolean isVertex) { + RdbmsTransaction trx = RdbmsTransaction.getActiveTransaction(); + + if (trx != null) { + trx.getEntityManager().createNativeQuery(isVertex ? SQL_INSERT_UNIQUE_VERTEX_KEY : SQL_INSERT_UNIQUE_EDGE_KEY) + .setParameter(1, getNumberId(elementId)) + .setParameter(2, keyName) + .setParameter(3, getStringValue(value)) + .executeUpdate(); + } else { + throw new IllegalStateException("No active transaction found to add unique key: keyName=" + keyName + ", value=" + value); + } + } + + public void removeUniqueKey(String keyName, Object value, boolean isVertex) { + RdbmsTransaction trx = RdbmsTransaction.getActiveTransaction(); + + if (trx != null) { + trx.getEntityManager().createNativeQuery(isVertex ? SQL_DELETE_UNIQUE_VERTEX_KEY : SQL_DELETE_UNIQUE_EDGE_KEY) + .setParameter(1, keyName) + .setParameter(2, getStringValue(value)) + .executeUpdate(); + } else { + throw new IllegalStateException("No active transaction found to remove unique key: keyName=" + keyName + ", value=" + value); + } + } + + public void addTypeUniqueKey(String typeName, String keyName, Object value, Object elementId, boolean isVertex) { + RdbmsTransaction trx = RdbmsTransaction.getActiveTransaction(); + + if (trx != null) { + trx.getEntityManager().createNativeQuery(isVertex ? SQL_INSERT_UNIQUE_VERTEX_TYPE_KEY : SQL_INSERT_UNIQUE_EDGE_TYPE_KEY) + .setParameter(1, getNumberId(elementId)) + .setParameter(2, typeName) + .setParameter(3, keyName) + .setParameter(4, getStringValue(value)) + .executeUpdate(); + } else { + throw new IllegalStateException("No active transaction found to add type unique key: typeName=" + typeName + ", keyName=" + keyName + ", value=" + value); + } + } + + public void removeTypeUniqueKey(String typeName, String keyName, Object value, boolean isVertex) { + RdbmsTransaction trx = RdbmsTransaction.getActiveTransaction(); + + if (trx != null) { + trx.getEntityManager().createNativeQuery(isVertex ? SQL_DELETE_UNIQUE_VERTEX_TYPE_KEY : SQL_DELETE_UNIQUE_EDGE_TYPE_KEY) + .setParameter(1, typeName) + .setParameter(2, keyName) + .setParameter(3, getStringValue(value)) + .executeUpdate(); + } else { + throw new IllegalStateException("No active transaction found to remove type unique key: typeName=" + typeName + ", keyName=" + keyName + ", value=" + value); + } + } + + public void removeUniqueKeysForVertexId(Object vertexId) { + RdbmsTransaction trx = RdbmsTransaction.getActiveTransaction(); + + if (trx != null) { + final Number id = getNumberId(vertexId); + + trx.getEntityManager().createNativeQuery(SQL_DELETE_UNIQUE_VERTEX_KEY_BY_VERTEX_ID) + .setParameter(1, id) + .executeUpdate(); + + trx.getEntityManager().createNativeQuery(SQL_DELETE_UNIQUE_VERTEX_TYPE_KEY_BY_VERTEX_ID) + .setParameter(1, id) + .executeUpdate(); + } else { + throw new IllegalStateException("No active transaction found to remove unique keys for vertex: vertexId=" + vertexId); + } + } + + public void removeUniqueKeysForEdgeId(Object edgeId) { + RdbmsTransaction trx = RdbmsTransaction.getActiveTransaction(); + + if (trx != null) { + final Number id = getNumberId(edgeId); + + trx.getEntityManager().createNativeQuery(SQL_DELETE_UNIQUE_EDGE_KEY_BY_EDGE_ID) + .setParameter(1, id) + .executeUpdate(); + + trx.getEntityManager().createNativeQuery(SQL_DELETE_UNIQUE_EDGE_TYPE_KEY_BY_EDGE_ID) + .setParameter(1, id) + .executeUpdate(); + } else { + throw new IllegalStateException("No active transaction found to remove unique keys for edge: edgeId=" + edgeId); + } + } + + private Number getNumberId(Object elementId) { + if (elementId instanceof Number) { + return (Number) elementId; + } else if (elementId instanceof RelationIdentifier) { + return ((RelationIdentifier) elementId).getRelationId(); + } else if (elementId == null) { + throw new IllegalArgumentException("Invalid elementId: null"); + } else { + throw new IllegalArgumentException("Invalid elementId type: " + elementId.getClass().getName()); + } + } + + private String getStringValue(Object value) { + if (value instanceof String) { + return (String) value; + } else if (value == null) { + return null; + } else { + return value.toString(); + } + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/BaseDao.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/BaseDao.java new file mode 100644 index 00000000000..ce4530c014a --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/BaseDao.java @@ -0,0 +1,53 @@ +/* + * 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.janusgraph.diskstorage.rdbms.dao; + +import org.janusgraph.diskstorage.rdbms.RdbmsTransaction; + +import javax.persistence.EntityManager; + +/** + * Base DAO to access entities stored in RDBMS + * + */ +public abstract class BaseDao { + protected final EntityManager em; + + protected BaseDao(RdbmsTransaction trx) { + this.em = trx.getEntityManager(); + } + + protected BaseDao(EntityManager em) { + this.em = em; + } + + public T create(T obj) { + em.persist(obj); + + return obj; + } + + protected Long toLong(Object obj) { + return (obj instanceof Long) ? (Long) obj : null; + } + + protected byte[] toByteArray(Object obj) { + return (obj instanceof byte[]) ? (byte[]) obj : null; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/DaoManager.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/DaoManager.java new file mode 100644 index 00000000000..ae434c65772 --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/DaoManager.java @@ -0,0 +1,121 @@ +/* + * 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.janusgraph.diskstorage.rdbms.dao; + +import com.zaxxer.hikari.HikariConfig; +import com.zaxxer.hikari.HikariDataSource; +import org.eclipse.persistence.config.PersistenceUnitProperties; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.persistence.EntityManager; +import javax.persistence.EntityManagerFactory; +import javax.persistence.spi.PersistenceProvider; +import javax.persistence.spi.PersistenceProviderResolver; +import javax.persistence.spi.PersistenceProviderResolverHolder; +import javax.sql.DataSource; + +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; + +/** + * DAO manager that initializes JPA layer + * + * Sample properties to be set in atlas-application.properties for RDBMS storage backend: + * atlas.graph.storage.backend=rdbms + * atlas.graph.storage.rdbms.jpa.hikari.driverClassName=org.postgresql.Driver + * atlas.graph.storage.rdbms.jpa.hikari.jdbcUrl=jdbc:postgresql://atlas-db/atlas + * atlas.graph.storage.rdbms.jpa.hikari.username=atlas + * atlas.graph.storage.rdbms.jpa.hikari.password=atlasR0cks! + * atlas.graph.storage.rdbms.jpa.hikari.maximumPoolSize=40 + * atlas.graph.storage.rdbms.jpa.hikari.minimumIdle=5 + * atlas.graph.storage.rdbms.jpa.hikari.idleTimeout=300000 + * atlas.graph.storage.rdbms.jpa.hikari.connectionTestQuery=select 1 + * atlas.graph.storage.rdbms.jpa.hikari.maxLifetime=1800000 + * atlas.graph.storage.rdbms.jpa.hikari.connectionTimeout=30000 + * atlas.graph.storage.rdbms.jpa.javax.persistence.jdbc.dialect=org.eclipse.persistence.platform.database.PostgreSQLPlatform + * atlas.graph.storage.rdbms.jpa.javax.persistence.schema-generation.database.action=create + * atlas.graph.storage.rdbms.jpa.javax.persistence.schema-generation.create-database-schemas=true + * atlas.graph.storage.rdbms.jpa.javax.persistence.schema-generation.create-source=script + * atlas.graph.storage.rdbms.jpa.javax.persistence.schema-generation.create-script-source=META-INF/postgres/create_schema.sql + * atlas.EntityAuditRepository.impl=org.apache.atlas.repository.audit.rdbms.RdbmsBasedAuditRepository + * + */ +public class DaoManager { + private static final Logger LOG = LoggerFactory.getLogger(DaoManager.class); + + private final EntityManagerFactory emFactory; + + /** + * + * @param jpaConfig + */ + public DaoManager(Map jpaConfig) { + Map config = new HashMap<>(); + Properties hikariConfig = new Properties(); + + if (jpaConfig != null) { + for (Map.Entry entry : jpaConfig.entrySet()) { + String key = entry.getKey(); + Object value = entry.getValue(); + + if (value != null) { + if (key.startsWith("hikari.")) { + hikariConfig.put(key.substring("hikari".length() + 1), value.toString()); + } else { + config.put(key, value.toString()); + } + } + } + } + + DataSource dataSource = new HikariDataSource(new HikariConfig(hikariConfig)); + + config.put(PersistenceUnitProperties.ECLIPSELINK_PERSISTENCE_XML, "META-INF/janus-persistence.xml"); + config.put(PersistenceUnitProperties.NON_JTA_DATASOURCE, dataSource); + + PersistenceProviderResolver resolver = PersistenceProviderResolverHolder.getPersistenceProviderResolver(); + EntityManagerFactory emf = null; + + for (PersistenceProvider provider : resolver.getPersistenceProviders()) { + LOG.debug("PersistenceProvider: {}", provider); + + emf = provider.createEntityManagerFactory("janusPU", config); + + if (emf != null) { + break; + } + } + + emFactory = emf; + } + + public EntityManager createEntityManager() { + return emFactory.createEntityManager(); + } + + public void close() { + LOG.info("DaoManager.close()"); + + if (this.emFactory.isOpen()) { + this.emFactory.close(); + } + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/JanusColumnDao.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/JanusColumnDao.java new file mode 100644 index 00000000000..78d3f6affcf --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/JanusColumnDao.java @@ -0,0 +1,308 @@ +/* + * 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.janusgraph.diskstorage.rdbms.dao; + +import org.eclipse.persistence.queries.ScrollableCursor; +import org.janusgraph.diskstorage.Entry; +import org.janusgraph.diskstorage.StaticBuffer; +import org.janusgraph.diskstorage.keycolumnvalue.KeyIterator; +import org.janusgraph.diskstorage.rdbms.JanusColumnValue; +import org.janusgraph.diskstorage.rdbms.RdbmsStore; +import org.janusgraph.diskstorage.rdbms.RdbmsTransaction; +import org.janusgraph.diskstorage.rdbms.entity.JanusColumn; +import org.janusgraph.diskstorage.util.RecordIterator; +import org.janusgraph.diskstorage.util.StaticArrayBuffer; +import org.janusgraph.diskstorage.util.StaticArrayEntry; + +import javax.persistence.NoResultException; + +import java.util.ArrayList; +import java.util.List; + +/** + * DAO to access Column entities stored in RDBMS + * + */ +public class JanusColumnDao extends BaseDao { + private final RdbmsStore store; + + public JanusColumnDao(RdbmsTransaction trx, RdbmsStore store) { + super(trx); + + this.store = store; + } + + public void addOrUpdate(long keyId, byte[] name, byte[] val) { + try { + em.createNativeQuery("INSERT INTO janus_column (key_id, name, val) VALUES (?, ?, ?) ON CONFLICT (key_id, name) DO UPDATE SET val = EXCLUDED.val") + .setParameter(1, keyId) + .setParameter(2, name) + .setParameter(3, val) + .executeUpdate(); + } catch (NoResultException excp) { + // ignore + } + } + + public int remove(long keyId, byte[] name) { + return em.createNamedQuery("JanusColumn.deleteByKeyIdAndName") + .setParameter("keyId", keyId) + .setParameter("name", name) + .executeUpdate(); + } + + public List getColumns(long keyId, byte[] startColumn, byte[] endColumn, int limit) { + List result = em.createNamedQuery("JanusColumn.getColumnsByKeyIdStartNameEndName", Object[].class) + .setParameter("keyId", keyId) + .setParameter("startName", startColumn) + .setParameter("endName", endColumn) + .setMaxResults(limit) + .getResultList(); + + return toColumnList(result); + } + + public KeyIterator getKeysByColumnRange(long storeId, byte[] startColumn, byte[] endColumn, int limit) { + ScrollableCursor result = (ScrollableCursor) em.createNamedQuery("JanusColumn.getKeysByStoreIdColumnRange") + .setParameter("storeId", storeId) + .setParameter("startName", startColumn) + .setParameter("endName", endColumn) + .setHint("eclipselink.cursor.scrollable", true) + .getResultList(); + + return toKeyColumns(result, limit); + } + + public KeyIterator getKeysByKeyAndColumnRange(long storeId, byte[] startKey, byte[] endKey, byte[] startColumn, byte[] endColumn, int limit) { + ScrollableCursor result = (ScrollableCursor) em.createNamedQuery("JanusColumn.getKeysByStoreIdKeyRangeColumnRange") + .setParameter("storeId", storeId) + .setParameter("startKey", startKey) + .setParameter("endKey", endKey) + .setParameter("startName", startColumn) + .setParameter("endName", endColumn) + .setHint("eclipselink.cursor.scrollable", true) + .getSingleResult(); + + return toKeyColumns(result, limit); + } + + private List toColumnList(List result) { + List ret = null; + + if (result != null && !result.isEmpty()) { + ret = new ArrayList<>(result.size()); + + for (Object[] row : result) { + byte[] name = toByteArray(row[0]); + byte[] val = toByteArray(row[1]); + + ret.add(new JanusColumnValue(name, val)); + } + } + + return ret; + } + + private KeyIterator toKeyColumns(ScrollableCursor keysResult, int limit) { + final KeyIterator ret; + + if (keysResult != null && keysResult.hasNext()) { + ret = new RdbmsKeyIterator(keysResult, limit); + } else { + ret = EMPTY_KEY_ITERATOR; + } + + return ret; + } + + private class RdbmsKeyIterator implements KeyIterator { + private final ScrollableCursor rows; + private final int limit; + private final Row currKey = new Row(); + private final Row nextKey = new Row(); + private Long prevKeyId; + private boolean isClosed; + + public RdbmsKeyIterator(ScrollableCursor rows, int limit) { + this.rows = rows; + this.limit = limit; + } + + @Override + public boolean hasNext() { + ensureOpen(); + + if (nextKey.keyId == null) { + while (rows.hasNext()) { + Object[] nextRow = (Object[]) rows.next(); + Long keyId = toLong(nextRow[0]); + + if (prevKeyId != null && prevKeyId.equals(keyId)) { // ignore additional columns for this key + continue; + } + + nextKey.set(keyId, StaticArrayBuffer.of(toByteArray(nextRow[1])), new JanusColumnValue(toByteArray(nextRow[2]), toByteArray(nextRow[3]))); + + break; + } + } + + return nextKey.keyId != null; + } + + @Override + public StaticBuffer next() { + ensureOpen(); + + prevKeyId = currKey.keyId; + + if (nextKey.keyId == null) { + hasNext(); + } + + currKey.copyFrom(nextKey); + + nextKey.reset(); + + return currKey.key; + } + + @Override + public RecordIterator getEntries() { + ensureOpen(); + + return new RecordIterator() { + private boolean isClosed; + private int colCount; + + @Override + public boolean hasNext() { + ensureOpen(); + + if (currKey.column == null) { + while (rows.hasNext()) { + Object[] nextRow = (Object[]) rows.next(); + Long keyId = toLong(nextRow[0]); + + if (!keyId.equals(currKey.keyId)) { + nextKey.set(keyId, StaticArrayBuffer.of(toByteArray(nextRow[1])), new JanusColumnValue(toByteArray(nextRow[2]), toByteArray(nextRow[3]))); + currKey.reset(); + + break; + } else if (colCount < limit) { // ignore additional columns for this key + currKey.column = new JanusColumnValue(toByteArray(nextRow[2]), toByteArray(nextRow[3])); + + break; + } + } + } + + return currKey.column != null; + } + + @Override + public Entry next() { + JanusColumnValue ret = currKey.column; + + currKey.column = null; + colCount++; + + return StaticArrayEntry.ofStaticBuffer(ret, store.toEntry); + } + + @Override + public void close() { + this.isClosed = true; + } + + private void ensureOpen() { + if (isClosed) { + throw new IllegalStateException("Iterator has been closed."); + } + } + }; + } + + @Override + public void close() { + isClosed = true; + + rows.close(); + } + + private void ensureOpen() { + if (isClosed) { + throw new IllegalStateException("Iterator has been closed."); + } + } + + private class Row { + private Long keyId; + private StaticBuffer key; + private JanusColumnValue column; + + Row() { } + + Row(Long keyId, StaticBuffer key, JanusColumnValue column) { + this.keyId = keyId; + this.key = key; + this.column = column; + } + + void copyFrom(Row other) { + this.keyId = other.keyId; + this.key = other.key; + this.column = other.column; + } + + void set(Long keyId, StaticBuffer key, JanusColumnValue column) { + this.keyId = keyId; + this.key = key; + this.column = column; + } + + void reset() { + this.keyId = null; + this.key = null; + this.column = null; + } + } + } + + public static final KeyIterator EMPTY_KEY_ITERATOR = new KeyIterator() { + @Override + public RecordIterator getEntries() { + return null; + } + + @Override + public void close() { + } + + @Override + public boolean hasNext() { + return false; + } + + @Override + public StaticBuffer next() { + return null; + } + }; +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/JanusKeyDao.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/JanusKeyDao.java new file mode 100644 index 00000000000..1d8b73d286c --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/JanusKeyDao.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.janusgraph.diskstorage.rdbms.dao; + +import org.janusgraph.diskstorage.rdbms.RdbmsTransaction; +import org.janusgraph.diskstorage.rdbms.entity.JanusKey; + +import javax.persistence.NoResultException; + +/** + * DAO to access Key entities stored in RDBMS + * + */ +public class JanusKeyDao extends BaseDao { + public JanusKeyDao(RdbmsTransaction trx) { + super(trx); + } + + public Long getIdByStoreIdAndName(long storeId, byte[] name) { + try { + Object result = em.createNamedQuery("JanusKey.getIdByStoreIdAndName") + .setParameter("storeId", storeId) + .setParameter("name", name) + .getSingleResult(); + + return toLong(result); + } catch (NoResultException excp) { + // ignore + } + + return null; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/JanusStoreDao.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/JanusStoreDao.java new file mode 100644 index 00000000000..dc4c8958d7c --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/dao/JanusStoreDao.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.janusgraph.diskstorage.rdbms.dao; + +import org.janusgraph.diskstorage.rdbms.RdbmsTransaction; +import org.janusgraph.diskstorage.rdbms.entity.JanusStore; + +import javax.persistence.NoResultException; + +/** + * DAO to access Store entities stored in RDBMS + * + */ +public class JanusStoreDao extends BaseDao { + public JanusStoreDao(RdbmsTransaction trx) { + super(trx); + } + + public Long getIdByName(String name) { + try { + Object result = em.createNamedQuery("JanusStore.getIdByName") + .setParameter("name", name) + .getSingleResult(); + + return toLong(result); + } catch (NoResultException excp) { + // ignore + } + + return null; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusColumn.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusColumn.java new file mode 100644 index 00000000000..c2b0e5359fe --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusColumn.java @@ -0,0 +1,126 @@ +/* + * 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.janusgraph.diskstorage.rdbms.entity; + +import javax.persistence.Cacheable; +import javax.persistence.Column; +import javax.persistence.Entity; +import javax.persistence.GeneratedValue; +import javax.persistence.GenerationType; +import javax.persistence.Id; +import javax.persistence.Index; +import javax.persistence.Lob; +import javax.persistence.SequenceGenerator; +import javax.persistence.Table; +import javax.persistence.UniqueConstraint; + +import java.util.Arrays; +import java.util.Objects; + +@Entity +@Cacheable(false) +@Table(name = "janus_column", + indexes = {@Index(name = "janus_column_idx_key_id", columnList = "key_id")}, + uniqueConstraints = {@UniqueConstraint(name = "janus_column_uk_key_name", columnNames = {"key_id", "name"})}) +public class JanusColumn implements java.io.Serializable { + private static final long serialVersionUID = 1L; + + @Id + @SequenceGenerator(name = "janus_column_seq", sequenceName = "janus_column_seq", allocationSize = 1) + @GeneratedValue(strategy = GenerationType.AUTO, generator = "janus_column_seq") + @Column(name = "id") + protected Long id; + + @Column(name = "key_id", nullable = false) + protected Long keyId; + + @Lob + @Column(name = "name", nullable = false) + protected byte[] name; + + @Lob + @Column(name = "val") + protected byte[] val; + + public JanusColumn() { } + + public JanusColumn(Long keyId, byte[] name, byte[] val) { + this.keyId = keyId; + this.name = name; + this.val = val; + } + + public void setId(Long id) { + this.id = id; + } + + public Long getId() { + return id; + } + + public void setKeyId(Long keyId) { + this.keyId = keyId; + } + + public Long getKeyId() { + return keyId; + } + + public void setName(byte[] name) { + this.name = name; + } + + public byte[] getName() { + return name; + } + + public void setVal(byte[] val) { + this.val = val; + } + + public byte[] getVal() { + return val; + } + + @Override + public int hashCode() { + return Objects.hash(id, keyId, name, val); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } else if (obj instanceof JanusColumn && getClass() == obj.getClass()) { + JanusColumn other = (JanusColumn) obj; + + return Objects.equals(id, other.id) && + Objects.equals(keyId, other.keyId) && + Arrays.equals(name, other.name) && + Arrays.equals(val, other.val); + } + + return false; + } + + @Override + public String toString() { + return "JanusColumn(id=" + id + ", keyId=" + keyId + ", name=" + name + ", val=" + val + ")"; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusKey.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusKey.java new file mode 100644 index 00000000000..a127de63b37 --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusKey.java @@ -0,0 +1,112 @@ +/* + * 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.janusgraph.diskstorage.rdbms.entity; + +import javax.persistence.Cacheable; +import javax.persistence.Column; +import javax.persistence.Entity; +import javax.persistence.GeneratedValue; +import javax.persistence.GenerationType; +import javax.persistence.Id; +import javax.persistence.Index; +import javax.persistence.Lob; +import javax.persistence.SequenceGenerator; +import javax.persistence.Table; +import javax.persistence.UniqueConstraint; + +import java.util.Arrays; +import java.util.Objects; + +@Entity +@Cacheable(false) +@Table(name = "janus_key", + indexes = {@Index(name = "janus_key_idx_store_id", columnList = "store_id")}, + uniqueConstraints = {@UniqueConstraint(name = "janus_key_uk_store_name", columnNames = {"store_id", "name"})}) +public class JanusKey implements java.io.Serializable { + private static final long serialVersionUID = 1L; + + @Id + @SequenceGenerator(name = "janus_key_seq", sequenceName = "janus_key_seq", allocationSize = 1) + @GeneratedValue(strategy = GenerationType.AUTO, generator = "janus_key_seq") + @Column(name = "id") + protected Long id; + + @Column(name = "store_id", nullable = false) + protected Long storeId; + + @Lob + @Column(name = "name", nullable = false) + protected byte[] name; + + public JanusKey() { } + + public JanusKey(Long storeId, byte[] name) { + this.storeId = storeId; + this.name = name; + } + + public void setId(Long id) { + this.id = id; + } + + public Long getId() { + return id; + } + + public void setStoreId(Long storeId) { + this.storeId = storeId; + } + + public Long getStoreId() { + return storeId; + } + + public void setName(byte[] name) { + this.name = name; + } + + public byte[] getName() { + return name; + } + + @Override + public int hashCode() { + return Objects.hash(id, storeId, name); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } else if (obj instanceof JanusKey && getClass() == obj.getClass()) { + JanusKey other = (JanusKey) obj; + + return Objects.equals(id, other.id) && + Objects.equals(storeId, other.storeId) && + Arrays.equals(name, other.name); + } + + return false; + } + + @Override + public String toString() { + return "JanusKey(id=" + id + ", storeId=" + storeId + ", name=" + name + ")"; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusStore.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusStore.java new file mode 100644 index 00000000000..4a75f62885a --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusStore.java @@ -0,0 +1,94 @@ +/* + * 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.janusgraph.diskstorage.rdbms.entity; + +import javax.persistence.Cacheable; +import javax.persistence.Column; +import javax.persistence.Entity; +import javax.persistence.GeneratedValue; +import javax.persistence.GenerationType; +import javax.persistence.Id; +import javax.persistence.SequenceGenerator; +import javax.persistence.Table; +import javax.persistence.UniqueConstraint; + +import java.util.Objects; + +@Entity +@Cacheable(false) +@Table(name = "janus_store", + uniqueConstraints = @UniqueConstraint(name = "janus_store_uk_name", columnNames = "name")) +public class JanusStore implements java.io.Serializable { + private static final long serialVersionUID = 1L; + + @Id + @SequenceGenerator(name = "janus_store_seq", sequenceName = "janus_store_seq", allocationSize = 1) + @GeneratedValue(strategy = GenerationType.AUTO, generator = "janus_store_seq") + @Column(name = "id") + protected Long id; + + @Column(name = "name", nullable = false) + protected String name; + + public JanusStore() { } + + public JanusStore(String name) { + this.name = name; + } + + public void setId(Long id) { + this.id = id; + } + + public Long getId() { + return id; + } + + public void setName(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + @Override + public int hashCode() { + return Objects.hash(id, name); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } else if (obj instanceof JanusStore && getClass() == obj.getClass()) { + JanusStore other = (JanusStore) obj; + + return Objects.equals(id, other.id) && + Objects.equals(name, other.name); + } + + return false; + } + + @Override + public String toString() { + return "JanusStore(id=" + id + ", name=" + name + ")"; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueEdgeKey.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueEdgeKey.java new file mode 100644 index 00000000000..36af654498a --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueEdgeKey.java @@ -0,0 +1,124 @@ +/* + * 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.janusgraph.diskstorage.rdbms.entity; + +import org.eclipse.persistence.annotations.Index; + +import javax.persistence.Cacheable; +import javax.persistence.Column; +import javax.persistence.Entity; +import javax.persistence.GeneratedValue; +import javax.persistence.GenerationType; +import javax.persistence.Id; +import javax.persistence.Lob; +import javax.persistence.SequenceGenerator; +import javax.persistence.Table; +import javax.persistence.UniqueConstraint; + +import java.util.Objects; + +@Entity +@Cacheable(false) +@Table(name = "janus_unique_edge_key", + uniqueConstraints = {@UniqueConstraint(name = "janus_unique_edge_key_uk", columnNames = {"key_name", "val"})}) +public class JanusUniqueEdgeKey implements java.io.Serializable { + private static final long serialVersionUID = 1L; + + @Id + @SequenceGenerator(name = "janus_unique_edge_key_seq", sequenceName = "janus_unique_edge_key_seq", allocationSize = 1) + @GeneratedValue(strategy = GenerationType.AUTO, generator = "janus_unique_edge_key_seq") + @Column(name = "id") + protected Long id; + + @Column(name = "edge_id", nullable = false) + @Index + protected Long edgeId; + + @Column(name = "key_name", nullable = false) + protected String keyName; + + @Lob + @Column(name = "val", nullable = false) + protected String val; + + public JanusUniqueEdgeKey() { } + + public JanusUniqueEdgeKey(String keyName, String val) { + this.keyName = keyName; + this.val = val; + } + + public void setId(Long id) { + this.id = id; + } + + public Long getId() { + return id; + } + + public void setEdgeId(Long edgeId) { + this.edgeId = edgeId; + } + + public Long getEdgeId() { + return edgeId; + } + + public void setKeyName(String keyName) { + this.keyName = keyName; + } + + public String getKeyName() { + return keyName; + } + + public void setVal(String val) { + this.val = val; + } + + public String getVal() { + return val; + } + + @Override + public int hashCode() { + return Objects.hash(id, edgeId, keyName, val); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } else if (obj instanceof JanusUniqueEdgeKey && getClass() == obj.getClass()) { + JanusUniqueEdgeKey other = (JanusUniqueEdgeKey) obj; + + return Objects.equals(id, other.id) && + Objects.equals(edgeId, other.edgeId) && + Objects.equals(keyName, other.keyName) && + Objects.equals(val, other.val); + } + + return false; + } + + @Override + public String toString() { + return "JanusUniqueEdgeKey(id=" + id + ", edgeId=" + edgeId + ", keyName=" + keyName + ", val=" + val + ")"; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueEdgeTypeKey.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueEdgeTypeKey.java new file mode 100644 index 00000000000..954eb72b578 --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueEdgeTypeKey.java @@ -0,0 +1,137 @@ +/* + * 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.janusgraph.diskstorage.rdbms.entity; + +import org.eclipse.persistence.annotations.Index; + +import javax.persistence.Cacheable; +import javax.persistence.Column; +import javax.persistence.Entity; +import javax.persistence.GeneratedValue; +import javax.persistence.GenerationType; +import javax.persistence.Id; +import javax.persistence.Lob; +import javax.persistence.SequenceGenerator; +import javax.persistence.Table; +import javax.persistence.UniqueConstraint; + +import java.util.Objects; + +@Entity +@Cacheable(false) +@Table(name = "janus_unique_edge_type_key", + uniqueConstraints = {@UniqueConstraint(name = "janus_unique_edge_type_key_uk", columnNames = {"type_name", "key_name", "val"})}) +public class JanusUniqueEdgeTypeKey implements java.io.Serializable { + private static final long serialVersionUID = 1L; + + @Id + @SequenceGenerator(name = "janus_unique_edge_type_key_seq", sequenceName = "janus_unique_edge_type_key_seq", allocationSize = 1) + @GeneratedValue(strategy = GenerationType.AUTO, generator = "janus_unique_edge_type_key_seq") + @Column(name = "id") + protected Long id; + + @Column(name = "edge_id", nullable = false) + @Index + protected Long edgeId; + + @Column(name = "type_name", nullable = false) + protected String typeName; + + @Column(name = "key_name", nullable = false) + protected String keyName; + + @Lob + @Column(name = "val", nullable = false) + protected String val; + + public JanusUniqueEdgeTypeKey() { } + + public JanusUniqueEdgeTypeKey(String typeName, String keyName, String val) { + this.typeName = typeName; + this.keyName = keyName; + this.val = val; + } + + public void setId(Long id) { + this.id = id; + } + + public Long getId() { + return id; + } + + public void setEdgeId(Long edgeId) { + this.edgeId = edgeId; + } + + public Long getEdgeId() { + return edgeId; + } + + public void setTypeName(String typeName) { + this.typeName = typeName; + } + + public String getTypeName() { + return typeName; + } + + public void setKeyName(String keyName) { + this.keyName = keyName; + } + + public String getKeyName() { + return keyName; + } + + public void setVal(String val) { + this.val = val; + } + + public String getVal() { + return val; + } + + @Override + public int hashCode() { + return Objects.hash(id, edgeId, typeName, keyName, val); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } else if (obj instanceof JanusUniqueEdgeTypeKey && getClass() == obj.getClass()) { + JanusUniqueEdgeTypeKey other = (JanusUniqueEdgeTypeKey) obj; + + return Objects.equals(id, other.id) && + Objects.equals(edgeId, other.edgeId) && + Objects.equals(typeName, other.typeName) && + Objects.equals(keyName, other.keyName) && + Objects.equals(val, other.val); + } + + return false; + } + + @Override + public String toString() { + return "JanusUniqueEdgeTypeKey(id=" + id + ", edgeId=" + edgeId + ", typeName=" + typeName + ", keyName=" + keyName + ", val=" + val + ")"; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueVertexKey.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueVertexKey.java new file mode 100644 index 00000000000..a20c328b631 --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueVertexKey.java @@ -0,0 +1,124 @@ +/* + * 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.janusgraph.diskstorage.rdbms.entity; + +import org.eclipse.persistence.annotations.Index; + +import javax.persistence.Cacheable; +import javax.persistence.Column; +import javax.persistence.Entity; +import javax.persistence.GeneratedValue; +import javax.persistence.GenerationType; +import javax.persistence.Id; +import javax.persistence.Lob; +import javax.persistence.SequenceGenerator; +import javax.persistence.Table; +import javax.persistence.UniqueConstraint; + +import java.util.Objects; + +@Entity +@Cacheable(false) +@Table(name = "janus_unique_vertex_key", + uniqueConstraints = {@UniqueConstraint(name = "janus_unique_vertex_key_uk", columnNames = {"key_name", "val"})}) +public class JanusUniqueVertexKey implements java.io.Serializable { + private static final long serialVersionUID = 1L; + + @Id + @SequenceGenerator(name = "janus_unique_vertex_key_seq", sequenceName = "janus_unique_vertex_key_seq", allocationSize = 1) + @GeneratedValue(strategy = GenerationType.AUTO, generator = "janus_unique_vertex_key_seq") + @Column(name = "id") + protected Long id; + + @Column(name = "vertex_id", nullable = false) + @Index + protected Long vertexId; + + @Column(name = "key_name", nullable = false) + protected String keyName; + + @Lob + @Column(name = "val", nullable = false) + protected String val; + + public JanusUniqueVertexKey() { } + + public JanusUniqueVertexKey(String keyName, String val) { + this.keyName = keyName; + this.val = val; + } + + public void setId(Long id) { + this.id = id; + } + + public Long getId() { + return id; + } + + public void setVertexId(Long vertexId) { + this.vertexId = vertexId; + } + + public Long getVertexId() { + return vertexId; + } + + public void setKeyName(String keyName) { + this.keyName = keyName; + } + + public String getKeyName() { + return keyName; + } + + public void setVal(String val) { + this.val = val; + } + + public String getVal() { + return val; + } + + @Override + public int hashCode() { + return Objects.hash(id, vertexId, keyName, val); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } else if (obj instanceof JanusUniqueVertexKey && getClass() == obj.getClass()) { + JanusUniqueVertexKey other = (JanusUniqueVertexKey) obj; + + return Objects.equals(id, other.id) && + Objects.equals(vertexId, other.vertexId) && + Objects.equals(keyName, other.keyName) && + Objects.equals(val, other.val); + } + + return false; + } + + @Override + public String toString() { + return "JanusUniqueVertexKey(id=" + id + ", vertexId=" + vertexId + ", keyName=" + keyName + ", val=" + val + ")"; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueVertexTypeKey.java b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueVertexTypeKey.java new file mode 100644 index 00000000000..d3e6edb3c21 --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/java/org/janusgraph/diskstorage/rdbms/entity/JanusUniqueVertexTypeKey.java @@ -0,0 +1,137 @@ +/* + * 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.janusgraph.diskstorage.rdbms.entity; + +import org.eclipse.persistence.annotations.Index; + +import javax.persistence.Cacheable; +import javax.persistence.Column; +import javax.persistence.Entity; +import javax.persistence.GeneratedValue; +import javax.persistence.GenerationType; +import javax.persistence.Id; +import javax.persistence.Lob; +import javax.persistence.SequenceGenerator; +import javax.persistence.Table; +import javax.persistence.UniqueConstraint; + +import java.util.Objects; + +@Entity +@Cacheable(false) +@Table(name = "janus_unique_vertex_type_key", + uniqueConstraints = {@UniqueConstraint(name = "janus_unique_vertex_type_key_uk", columnNames = {"type_name", "key_name", "val"})}) +public class JanusUniqueVertexTypeKey implements java.io.Serializable { + private static final long serialVersionUID = 1L; + + @Id + @SequenceGenerator(name = "janus_unique_vertex_type_key_seq", sequenceName = "janus_unique_vertex_type_key_seq", allocationSize = 1) + @GeneratedValue(strategy = GenerationType.AUTO, generator = "janus_unique_vertex_type_key_seq") + @Column(name = "id") + protected Long id; + + @Column(name = "vertex_id", nullable = false) + @Index + protected Long vertexId; + + @Column(name = "type_name", nullable = false) + protected String typeName; + + @Column(name = "key_name", nullable = false) + protected String keyName; + + @Lob + @Column(name = "val", nullable = false) + protected String val; + + public JanusUniqueVertexTypeKey() { } + + public JanusUniqueVertexTypeKey(String typeName, String keyName, String val) { + this.typeName = typeName; + this.keyName = keyName; + this.val = val; + } + + public void setId(Long id) { + this.id = id; + } + + public Long getId() { + return id; + } + + public void setVertexId(Long vertexId) { + this.vertexId = vertexId; + } + + public Long getVertexId() { + return vertexId; + } + + public void setTypeName(String typeName) { + this.typeName = typeName; + } + + public String getTypeName() { + return typeName; + } + + public void setKeyName(String keyName) { + this.keyName = keyName; + } + + public String getKeyName() { + return keyName; + } + + public void setVal(String val) { + this.val = val; + } + + public String getVal() { + return val; + } + + @Override + public int hashCode() { + return Objects.hash(id, vertexId, typeName, keyName, val); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } else if (obj instanceof JanusUniqueVertexTypeKey && getClass() == obj.getClass()) { + JanusUniqueVertexTypeKey other = (JanusUniqueVertexTypeKey) obj; + + return Objects.equals(id, other.id) && + Objects.equals(vertexId, other.vertexId) && + Objects.equals(typeName, other.typeName) && + Objects.equals(keyName, other.keyName) && + Objects.equals(val, other.val); + } + + return false; + } + + @Override + public String toString() { + return "JanusUniqueVertexTypeKey(id=" + id + ", vertexId=" + vertexId + ", typeName=" + typeName + ", keyName=" + keyName + ", val=" + val + ")"; + } +} diff --git a/graphdb/janusgraph-rdbms/src/main/resources/META-INF/janus-jpa_named_queries.xml b/graphdb/janusgraph-rdbms/src/main/resources/META-INF/janus-jpa_named_queries.xml new file mode 100644 index 00000000000..676277ae100 --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/resources/META-INF/janus-jpa_named_queries.xml @@ -0,0 +1,96 @@ + + + + + + SELECT obj FROM DbEntityAudit obj + WHERE obj.entityId = :entityId + ORDER BY obj.eventTime DESC, obj.eventIndex DESC + + + + + SELECT obj FROM DbEntityAudit obj + WHERE obj.entityId = :entityId + AND obj.action = :action + ORDER BY obj.eventTime DESC, obj.eventIndex DESC + + + + + SELECT obj FROM DbEntityAudit obj + WHERE obj.entityId = :entityId + AND obj.action = :action + AND obj.eventTime >= :eventTimeStart + AND obj.eventIndex >= :eventIdxStart + ORDER BY obj.eventTime DESC, obj.eventIndex DESC + + + + + SELECT s.id FROM JanusStore s WHERE s.name = :name + + + + SELECT k.id FROM JanusKey k WHERE k.storeId = :storeId AND k.name = :name + + + + SELECT k.name FROM JanusKey k WHERE k.storeId = :storeId + + + + SELECT c.id + FROM JanusColumn c + WHERE c.keyId = :keyId AND c.name = :name + + + + UPDATE JanusColumn c SET c.val = :val WHERE c.id = :id + + + + UPDATE JanusColumn c SET c.val = :val WHERE c.keyId = :keyId AND c.name = :name + + + + DELETE FROM JanusColumn c WHERE c.keyId = :keyId AND c.name = :name + + + + SELECT c.val FROM JanusColumn c WHERE c.keyId = :keyId AND c.name = :name + + + + SELECT c.name, c.val + FROM JanusColumn c + WHERE c.keyId = :keyId AND c.name >= :startName AND c.name < :endName + + + + SELECT k.id, k.name, c.name, c.val + FROM JanusColumn c, JanusKey k + WHERE k.storeId = :storeId + AND c.keyId = k.id AND c.name >= :startName AND c.name < :endName + ORDER BY k.name, c.name + + + + SELECT k.id, k.name, c.name, c.val + FROM JanusColumn c, JanusKey k + WHERE k.storeId = :storeId AND k.name >= :startKey AND k.name < :endKey + AND c.keyId = k.id AND c.name >= :startName AND c.name < :endName + ORDER BY k.name, c.name + + diff --git a/graphdb/janusgraph-rdbms/src/main/resources/META-INF/janus-persistence.xml b/graphdb/janusgraph-rdbms/src/main/resources/META-INF/janus-persistence.xml new file mode 100644 index 00000000000..e352418e85b --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/resources/META-INF/janus-persistence.xml @@ -0,0 +1,38 @@ + + + + + META-INF/janus-jpa_named_queries.xml + + org.apache.atlas.repository.audit.rdbms.entity.DbEntityAudit + + org.janusgraph.diskstorage.rdbms.entity.JanusStore + org.janusgraph.diskstorage.rdbms.entity.JanusKey + org.janusgraph.diskstorage.rdbms.entity.JanusColumn + org.janusgraph.diskstorage.rdbms.entity.JanusUniqueVertexKey + org.janusgraph.diskstorage.rdbms.entity.JanusUniqueVertexTypeKey + org.janusgraph.diskstorage.rdbms.entity.JanusUniqueEdgeKey + org.janusgraph.diskstorage.rdbms.entity.JanusUniqueEdgeTypeKey + NONE + + + + + + + diff --git a/graphdb/janusgraph-rdbms/src/main/resources/META-INF/postgres/create_schema.sql b/graphdb/janusgraph-rdbms/src/main/resources/META-INF/postgres/create_schema.sql new file mode 100644 index 00000000000..346d92eb64b --- /dev/null +++ b/graphdb/janusgraph-rdbms/src/main/resources/META-INF/postgres/create_schema.sql @@ -0,0 +1,63 @@ +-- 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. + +-- DB objects for Atlas entity audit +CREATE SEQUENCE IF NOT EXISTS atlas_entity_audit_seq; + +CREATE TABLE IF NOT EXISTS atlas_entity_audit(id BIGINT DEFAULT nextval('atlas_entity_audit_seq'::regclass), entity_id VARCHAR(64) NOT NULL, event_time BIGINT NOT NULL, event_idx INT NOT NULL, user_name VARCHAR(64) NOT NULL, operation INT NOT NULL, details TEXT DEFAULT NULL, entity TEXT DEFAULT NULL, audit_type INT NOT NULL, PRIMARY KEY(id)); + +CREATE INDEX IF NOT EXISTS atlas_entity_audit_idx_entity_id ON atlas_entity_audit (entity_id); +CREATE INDEX IF NOT EXISTS atlas_entity_audit_idx_event_time ON atlas_entity_audit (event_time); +CREATE INDEX IF NOT EXISTS atlas_entity_audit_idx_user_name ON atlas_entity_audit (user_name); +CREATE INDEX IF NOT EXISTS atlas_entity_audit_idx_entity_id_event_time ON atlas_entity_audit (entity_id, event_time); + + +-- DB objects for JanusGraph backend store +CREATE SEQUENCE IF NOT EXISTS janus_store_seq; +CREATE SEQUENCE IF NOT EXISTS janus_key_seq; +CREATE SEQUENCE IF NOT EXISTS janus_column_seq ; + +CREATE TABLE IF NOT EXISTS janus_store(id BIGINT DEFAULT NEXTVAL('janus_store_seq'::regclass), name VARCHAR(255) NOT NULL, PRIMARY KEY(id)); +CREATE TABLE IF NOT EXISTS janus_key(id BIGINT DEFAULT NEXTVAL('janus_key_seq'::regclass), store_id BIGINT NOT NULL, name BYTEA NOT NULL, PRIMARY KEY(id)); +CREATE TABLE IF NOT EXISTS janus_column(id BIGINT DEFAULT NEXTVAL('janus_column_seq'::regclass), key_id BIGINT NOT NULL, name BYTEA NOT NULL, val BYTEA NOT NULL, PRIMARY KEY(id)); + +CREATE UNIQUE INDEX IF NOT EXISTS janus_store_uk_name ON janus_store(name); +CREATE UNIQUE INDEX IF NOT EXISTS janus_key_uk_store_name ON janus_key(store_id, name); +CREATE UNIQUE INDEX IF NOT EXISTS janus_column_uk_key_name ON janus_column(key_id, name); + +CREATE INDEX IF NOT EXISTS janus_key_idx_store_id ON janus_key (store_id); +CREATE INDEX IF NOT EXISTS janus_column_idx_key_id ON janus_column (key_id); + + +-- DB objects for JanusGraph unique key constraints +CREATE SEQUENCE IF NOT EXISTS janus_unique_vertex_key_seq; +CREATE SEQUENCE IF NOT EXISTS janus_unique_vertex_type_key_seq; +CREATE SEQUENCE IF NOT EXISTS janus_unique_edge_key_seq; +CREATE SEQUENCE IF NOT EXISTS janus_unique_edge_type_key_seq; + +CREATE TABLE IF NOT EXISTS janus_unique_vertex_key(id BIGINT DEFAULT NEXTVAL('janus_unique_vertex_key_seq'::regclass), vertex_id BIGINT NOT NULL, key_name VARCHAR(255) NOT NULL, val TEXT NOT NULL, PRIMARY KEY(id)); +CREATE TABLE IF NOT EXISTS janus_unique_vertex_type_key(id BIGINT DEFAULT NEXTVAL('janus_unique_vertex_type_key_seq'::regclass), vertex_id BIGINT NOT NULL, type_name VARCHAR(255) NOT NULL, key_name VARCHAR(255) NOT NULL, val TEXT NOT NULL, PRIMARY KEY(id)); +CREATE TABLE IF NOT EXISTS janus_unique_edge_key(id BIGINT DEFAULT NEXTVAL('janus_unique_edge_key_seq'::regclass), edge_id BIGINT NOT NULL, key_name VARCHAR(255) NOT NULL, val TEXT NOT NULL, PRIMARY KEY(id)); +CREATE TABLE IF NOT EXISTS janus_unique_edge_type_key(id BIGINT DEFAULT NEXTVAL('janus_unique_edge_type_key_seq'::regclass), edge_id BIGINT NOT NULL, type_name VARCHAR(255) NOT NULL, key_name VARCHAR(255) NOT NULL, val TEXT NOT NULL, PRIMARY KEY(id)); + +CREATE UNIQUE INDEX IF NOT EXISTS janus_unique_vertex_key_uk ON janus_unique_vertex_key(key_name, val); +CREATE UNIQUE INDEX IF NOT EXISTS janus_unique_vertex_type_key_uk ON janus_unique_vertex_type_key(type_name, key_name, val); +CREATE UNIQUE INDEX IF NOT EXISTS janus_unique_edge_key_uk ON janus_unique_edge_key(key_name, val); +CREATE UNIQUE INDEX IF NOT EXISTS janus_unique_edge_type_key_uk ON janus_unique_edge_type_key(type_name, key_name, val); + +CREATE INDEX IF NOT EXISTS janus_unique_vertex_key_idx_vertex_id ON janus_unique_vertex_key (vertex_id); +CREATE INDEX IF NOT EXISTS janus_unique_vertex_type_key_idx_vertex_id ON janus_unique_vertex_type_key (vertex_id); +CREATE INDEX IF NOT EXISTS janus_unique_edge_key_idx_edge_id ON janus_unique_edge_key (edge_id); +CREATE INDEX IF NOT EXISTS janus_unique_edge_type_key_idx_edge_id ON janus_unique_edge_type_key (edge_id); diff --git a/graphdb/pom.xml b/graphdb/pom.xml index d58286dc1ce..c4c07948ccd 100644 --- a/graphdb/pom.xml +++ b/graphdb/pom.xml @@ -37,6 +37,7 @@ common graphdb-impls janus + janusgraph-rdbms diff --git a/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java b/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java index 6a5ace48f3f..b993c26e9fb 100644 --- a/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java +++ b/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java @@ -81,6 +81,7 @@ public enum AtlasConfiguration { LINEAGE_MAX_NODE_COUNT("atlas.lineage.max.node.count", 9000), HTTP_HEADER_SERVER_VALUE("atlas.http.header.server.value", "Apache Atlas"), + STORAGE_BACKEND_TYPE("atlas.graph.storage.backend", ""), STORAGE_CONSISTENCY_LOCK_ENABLED("atlas.graph.storage.consistency-lock.enabled", true), REBUILD_INDEX("atlas.rebuild.index", false), PROCESS_NAME_UPDATE_PATCH("atlas.process.name.update.patch", false), diff --git a/pom.xml b/pom.xml index ac2d4df6c52..c5ce12febf0 100644 --- a/pom.xml +++ b/pom.xml @@ -109,6 +109,7 @@ 4.3.0 1.8 3.2.2 + 2.7.15 7.17.8 org.apache.atlas.repository.audit.InMemoryEntityAuditRepository 2.13.2 @@ -137,6 +138,7 @@ 1.8 1.8 1 + 2.2.1 3.1.0 2.3.1 1.19.4 @@ -166,6 +168,7 @@ 2.7 5.4.0 5.4.0 + 42.7.7 run build build-minify build-minify diff --git a/repository/pom.xml b/repository/pom.xml index a724cf61789..d365228aa23 100644 --- a/repository/pom.xml +++ b/repository/pom.xml @@ -123,6 +123,11 @@ org.apache.atlas atlas-server-api + + org.apache.atlas + janusgraph-rdbms + ${project.version} + org.apache.cassandra @@ -222,11 +227,29 @@ + + org.eclipse.persistence + eclipselink + ${eclipse.jpa.version} + + + + org.eclipse.persistence + javax.persistence + ${javax.persistence.version} + + org.mockito mockito-all + + org.postgresql + postgresql + ${postgresql.version} + + org.springframework spring-aop diff --git a/repository/src/main/java/org/apache/atlas/repository/audit/AbstractStorageBasedAuditRepository.java b/repository/src/main/java/org/apache/atlas/repository/audit/AbstractStorageBasedAuditRepository.java index cc5312d0cca..a12e56d44e1 100644 --- a/repository/src/main/java/org/apache/atlas/repository/audit/AbstractStorageBasedAuditRepository.java +++ b/repository/src/main/java/org/apache/atlas/repository/audit/AbstractStorageBasedAuditRepository.java @@ -159,10 +159,42 @@ protected void setApplicationProperties(Configuration config) { applicationProperties = config; } + protected String getKeyStr(String id, Long ts, int index) { + return id + FIELD_SEPARATOR + ts + FIELD_SEPARATOR + index + FIELD_SEPARATOR + System.currentTimeMillis(); + } + protected byte[] getKey(String id, Long ts, int index) { - String keyStr = id + FIELD_SEPARATOR + ts + FIELD_SEPARATOR + index + FIELD_SEPARATOR + System.currentTimeMillis(); + return Bytes.toBytes(getKeyStr(id, ts, index)); + } + + protected long getTimestampFromKey(String key) { + String[] parts = key.split(FIELD_SEPARATOR); - return Bytes.toBytes(keyStr); + if (parts.length < 3) { + return 0L; + } + + try { + return Long.parseLong(parts[1]); + } catch (NumberFormatException e) { + LOG.error("Error parsing timestamp from key: {}", key, e); + return 0L; + } + } + + protected int getIndexFromKey(String key) { + String[] parts = key.split(FIELD_SEPARATOR); + + if (parts.length < 3) { + return 0; + } + + try { + return Integer.parseInt(parts[2]); + } catch (NumberFormatException e) { + LOG.error("Error parsing index from key: {}", key, e); + return 0; + } } static { diff --git a/repository/src/main/java/org/apache/atlas/repository/audit/rdbms/RdbmsBasedAuditRepository.java b/repository/src/main/java/org/apache/atlas/repository/audit/rdbms/RdbmsBasedAuditRepository.java new file mode 100644 index 00000000000..a63e73160ed --- /dev/null +++ b/repository/src/main/java/org/apache/atlas/repository/audit/rdbms/RdbmsBasedAuditRepository.java @@ -0,0 +1,161 @@ +/** + * 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.atlas.repository.audit.rdbms; + +import org.apache.atlas.ApplicationProperties; +import org.apache.atlas.AtlasException; +import org.apache.atlas.EntityAuditEvent; +import org.apache.atlas.annotation.ConditionalOnAtlasProperty; +import org.apache.atlas.exception.AtlasBaseException; +import org.apache.atlas.model.audit.EntityAuditEventV2; +import org.apache.atlas.repository.audit.AbstractStorageBasedAuditRepository; +import org.apache.atlas.repository.audit.rdbms.dao.DbEntityAuditDao; +import org.apache.atlas.repository.audit.rdbms.entity.DbEntityAudit; +import org.apache.commons.configuration.Configuration; +import org.springframework.core.annotation.Order; +import org.springframework.stereotype.Component; + +import javax.inject.Singleton; + +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; + +@Singleton +@Component +@ConditionalOnAtlasProperty(property = "atlas.EntityAuditRepository.impl", isDefault = true) +@Order(0) +public class RdbmsBasedAuditRepository extends AbstractStorageBasedAuditRepository { + @Override + public void putEventsV1(List events) throws AtlasException { + throw new UnsupportedOperationException(); + } + + @Override + public List listEventsV1(String entityId, String startKey, short n) throws AtlasException { + throw new UnsupportedOperationException(); + } + + @Override + public void putEventsV2(List events) throws AtlasBaseException { + try (RdbmsTransaction trx = new RdbmsTransaction()) { + DbEntityAuditDao dao = new DbEntityAuditDao(trx.getEntityManager()); + + for (int i = 0; i < events.size(); i++) { + EntityAuditEventV2 event = events.get(i); + DbEntityAudit dbEvent = toDbEntityAudit(event); + + dbEvent.setEventIndex(i); + + dao.create(dbEvent); + } + + trx.commit(); + } catch (Exception excp) { + throw new AtlasBaseException("Error while persisting audit events", excp); + } + } + + @Override + public List listEventsV2(String entityId, EntityAuditEventV2.EntityAuditActionV2 auditAction, String startKey, short maxResultCount) throws AtlasBaseException { + try (RdbmsTransaction trx = new RdbmsTransaction()) { + DbEntityAuditDao dao = new DbEntityAuditDao(trx.getEntityManager()); + + List dbEvents = dao.getByEntityIdActionStartTimeStartIdx(entityId, auditAction.ordinal(), getTimestampFromKey(startKey), getIndexFromKey(startKey), maxResultCount); + + return dbEvents.stream().map(RdbmsBasedAuditRepository::fromDbEntityAudit).collect(Collectors.toList()); + } catch (Exception excp) { + throw new AtlasBaseException("Error while retrieving audit events", excp); + } + } + + @Override + public List listEventsV2(String entityId, EntityAuditEventV2.EntityAuditActionV2 auditAction, String sortByColumn, boolean sortOrderDesc, int offset, short limit) throws AtlasBaseException { + try (RdbmsTransaction trx = new RdbmsTransaction()) { + DbEntityAuditDao dao = new DbEntityAuditDao(trx.getEntityManager()); + + List dbEvents = dao.getByEntityIdAction(entityId, auditAction == null ? null : auditAction.ordinal(), offset, limit); + + return dbEvents.stream().map(RdbmsBasedAuditRepository::fromDbEntityAudit).collect(Collectors.toList()); + } catch (Exception excp) { + throw new AtlasBaseException("Error while retrieving audit events", excp); + } + } + + @Override + public Set getEntitiesWithTagChanges(long fromTimestamp, long toTimestamp) throws AtlasBaseException { + // TODO: + return Collections.emptySet(); + } + + @Override + public void start() throws AtlasException { + } + + @Override + public void stop() throws AtlasException { + } + + public static DbEntityAudit toDbEntityAudit(EntityAuditEventV2 event) { + DbEntityAudit ret = new DbEntityAudit(); + + ret.setEntityId(event.getEntityId()); + ret.setEventTime(event.getTimestamp()); + ret.setUser(event.getUser()); + ret.setAction(event.getAction().ordinal()); + ret.setDetails(event.getDetails()); + + if (event.getType() == null) { + ret.setAuditType(EntityAuditEventV2.EntityAuditType.ENTITY_AUDIT_V2.ordinal()); + } else { + ret.setAuditType(event.getType().ordinal()); + } + + if (PERSIST_ENTITY_DEFINITION) { + ret.setEntity(event.getEntityDefinitionString()); + } + + return ret; + } + + public static EntityAuditEventV2 fromDbEntityAudit(DbEntityAudit dbEntityAudit) { + EntityAuditEventV2 ret = new EntityAuditEventV2(); + + ret.setEntityId(dbEntityAudit.getEntityId()); + ret.setTimestamp(dbEntityAudit.getEventTime()); + ret.setUser(dbEntityAudit.getUser()); + ret.setAction(EntityAuditEventV2.EntityAuditActionV2.values()[dbEntityAudit.getAction()]); + ret.setDetails(dbEntityAudit.getDetails()); + ret.setType(EntityAuditEventV2.EntityAuditType.values()[dbEntityAudit.getAuditType()]); + + if (PERSIST_ENTITY_DEFINITION) { + ret.setEntityDefinition(dbEntityAudit.getEntity()); + } + + return ret; + } + + private Configuration getConfiguration() { + try { + return ApplicationProperties.get(); + } catch (AtlasException e) { + throw new RuntimeException("Failed to get application properties", e); + } + } +} diff --git a/repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedSearchIndexer.java b/repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedSearchIndexer.java index d6aaab0250a..c9df3c3b330 100755 --- a/repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedSearchIndexer.java +++ b/repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedSearchIndexer.java @@ -43,6 +43,7 @@ import org.apache.atlas.repository.graphdb.AtlasGraphIndex; import org.apache.atlas.repository.graphdb.AtlasGraphManagement; import org.apache.atlas.repository.graphdb.AtlasPropertyKey; +import org.apache.atlas.repository.graphdb.AtlasUniqueKeyHandler; import org.apache.atlas.repository.store.graph.v2.AtlasGraphUtilsV2; import org.apache.atlas.type.AtlasArrayType; import org.apache.atlas.type.AtlasBusinessMetadataType; @@ -163,13 +164,16 @@ public class GraphBackedSearchIndexer implements SearchIndexer, ActiveStateChangeHandler, TypeDefChangeListener { private static final Logger LOG = LoggerFactory.getLogger(GraphBackedSearchIndexer.class); - private static final String VERTEX_ID_IN_IMPORT_KEY = "__vIdInImport"; - private static final String EDGE_ID_IN_IMPORT_KEY = "__eIdInImport"; - private static final List> INDEX_EXCLUSION_CLASSES = new ArrayList<>(Arrays.asList(Boolean.class, BigDecimal.class, BigInteger.class)); + private static final String VERTEX_ID_IN_IMPORT_KEY = "__vIdInImport"; + private static final String EDGE_ID_IN_IMPORT_KEY = "__eIdInImport"; + private static final List> INDEX_EXCLUSION_CLASSES = new ArrayList<>(Arrays.asList(Boolean.class, BigDecimal.class, BigInteger.class)); + private static final Set GLOBAL_UNIQUE_INDEX_KEYS = new HashSet<>(); + private static final Set TYPE_UNIQUE_INDEX_KEYS = new HashSet<>(); // Added for type lookup when indexing the new typedefs private final AtlasTypeRegistry typeRegistry; private final List indexChangeListeners = new ArrayList<>(); + private AtlasUniqueKeyHandler uniqueKeyHandler; //allows injection of a dummy graph for testing private final IAtlasGraphProvider provider; @@ -199,6 +203,14 @@ public GraphBackedSearchIndexer(AtlasTypeRegistry typeRegistry) throws AtlasExce notifyInitializationStart(); } + public static boolean isGlobalUniqueIndexKey(String key) { + return GLOBAL_UNIQUE_INDEX_KEYS.contains(key); + } + + public static boolean isTypeUniqueIndexKey(String key) { + return TYPE_UNIQUE_INDEX_KEYS.contains(key); + } + public static boolean isValidSearchWeight(int searchWeight) { if (searchWeight != -1) { return searchWeight >= 1 && searchWeight <= 10; @@ -290,6 +302,7 @@ public void onChange(ChangedTypeDefs changedTypeDefs) throws AtlasBaseException recomputeIndexedKeys = true; } + populateUniqueIndexKeys(); notifyChangeListeners(changedTypeDefs); } @@ -311,6 +324,7 @@ public void onLoadCompletion() throws AtlasBaseException { //Commit indexes management.setIsSuccess(true); + populateUniqueIndexKeys(); notifyInitializationCompletion(changedTypeDefs); } catch (Exception e) { LOG.error("Failed to update indexes for changed typedefs", e); @@ -500,6 +514,8 @@ private void initialize(AtlasGraph graph) throws RepositoryException, IndexExcep try (AtlasGraphManagement management = graph.getManagementSystem()) { LOG.info("Creating indexes for graph."); + uniqueKeyHandler = graph.getUniqueKeyHandler(); + if (management.getGraphIndex(VERTEX_INDEX) == null) { management.createVertexMixedIndex(VERTEX_INDEX, BACKING_INDEX, Collections.emptyList()); @@ -965,6 +981,16 @@ private AtlasPropertyKey createFullTextIndex(AtlasGraphManagement management, St private void createVertexCompositeIndex(AtlasGraphManagement management, Class propertyClass, AtlasPropertyKey propertyKey, boolean enforceUniqueness) { String propertyName = propertyKey.getName(); + if (enforceUniqueness) { + GLOBAL_UNIQUE_INDEX_KEYS.add(propertyName); + + if (uniqueKeyHandler != null) { + LOG.warn("ignoring uniqueness for composite index for property: {}", propertyName); + + enforceUniqueness = false; + } + } + LOG.debug("Creating composite index for property {} of type {}; isUnique={} ", propertyName, propertyClass.getName(), enforceUniqueness); AtlasGraphIndex existingIndex = management.getGraphIndex(propertyName); @@ -1001,6 +1027,14 @@ private void createVertexCompositeIndexWithSuperTypeName(AtlasGraphManagement ma private void createVertexCompositeIndexWithSystemProperty(AtlasGraphManagement management, Class propertyClass, AtlasPropertyKey propertyKey, final String systemPropertyKey, AtlasCardinality cardinality, boolean isUnique) { LOG.debug("Creating composite index for property {} of type {} and {}", propertyKey.getName(), propertyClass.getName(), systemPropertyKey); + if (isUnique) { + if (uniqueKeyHandler != null) { + LOG.warn("ignoring uniqueness for composite index with system property: {} + {}", systemPropertyKey, propertyKey.getName()); + + isUnique = false; + } + } + AtlasPropertyKey typePropertyKey = management.getPropertyKey(systemPropertyKey); if (typePropertyKey == null) { @@ -1123,5 +1157,24 @@ private void createEdgeLabels(AtlasGraphManagement management, AtlasRelationship createEdgeLabelUsingLabelName(management, relationshipLabel); } + private void populateUniqueIndexKeys() { + typeRegistry.getAllEntityTypes().forEach(this::populateUniqueIndexKeys); + typeRegistry.getAllClassificationTypes().forEach(this::populateUniqueIndexKeys); + typeRegistry.getAllStructTypes().forEach(this::populateUniqueIndexKeys); + typeRegistry.getAllRelationshipTypes().forEach(this::populateUniqueIndexKeys); + typeRegistry.getAllBusinessMetadataTypes().forEach(this::populateUniqueIndexKeys); + + LOG.info("{} global unique index keys found", GLOBAL_UNIQUE_INDEX_KEYS.size()); + LOG.info("{} type unique index keys found", TYPE_UNIQUE_INDEX_KEYS.size()); + } + + private void populateUniqueIndexKeys(AtlasStructType structType) { + if (structType.getUniqAttributes() != null) { + for (AtlasAttribute attribute : structType.getUniqAttributes().values()) { + TYPE_UNIQUE_INDEX_KEYS.add(attribute.getVertexUniquePropertyName()); + } + } + } + public enum UniqueKind { NONE, GLOBAL_UNIQUE, PER_TYPE_UNIQUE } } diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v1/DeleteHandlerV1.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v1/DeleteHandlerV1.java index 57e120b7a48..24c80c0a1a8 100644 --- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v1/DeleteHandlerV1.java +++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v1/DeleteHandlerV1.java @@ -36,6 +36,7 @@ import org.apache.atlas.repository.graphdb.AtlasEdge; import org.apache.atlas.repository.graphdb.AtlasEdgeDirection; import org.apache.atlas.repository.graphdb.AtlasGraph; +import org.apache.atlas.repository.graphdb.AtlasUniqueKeyHandler; import org.apache.atlas.repository.graphdb.AtlasVertex; import org.apache.atlas.repository.store.graph.v2.AtlasGraphUtilsV2; import org.apache.atlas.repository.store.graph.v2.EntityGraphRetriever; @@ -134,6 +135,7 @@ public abstract class DeleteHandlerV1 { private final boolean shouldUpdateInverseReferences; private final boolean softDelete; private final TaskManagement taskManagement; + private final AtlasUniqueKeyHandler uniqueKeyHandler; public DeleteHandlerV1(AtlasGraph graph, AtlasTypeRegistry typeRegistry, boolean shouldUpdateInverseReference, boolean softDelete, TaskManagement taskManagement) { this.typeRegistry = typeRegistry; @@ -142,6 +144,7 @@ public DeleteHandlerV1(AtlasGraph graph, AtlasTypeRegistry typeRegistry, boolean this.shouldUpdateInverseReferences = shouldUpdateInverseReference; this.softDelete = softDelete; this.taskManagement = taskManagement; + this.uniqueKeyHandler = graph.getUniqueKeyHandler(); } /** @@ -654,6 +657,10 @@ public void deletePropagatedEdge(AtlasEdge edge) throws AtlasBaseException { removeFromPropagatedClassificationNames(entityVertex, classificationName); + if (uniqueKeyHandler != null) { + uniqueKeyHandler.removeUniqueKeysForEdgeId(edge.getId()); + } + deleteEdge(edge, true); updateModificationMetadata(entityVertex); @@ -842,6 +849,10 @@ protected void deleteEdge(AtlasEdge edge, boolean updateInverseAttribute, boolea RequestContext.get().getDeleteType() == DeleteType.HARD ? PURGED.name() : DELETED.name()); } + if (uniqueKeyHandler != null) { + uniqueKeyHandler.removeUniqueKeysForEdgeId(edge.getId()); + } + deleteEdge(edge, force); } @@ -941,6 +952,10 @@ protected void deleteTypeVertex(AtlasVertex instanceVertex, boolean force) throw } } + if (uniqueKeyHandler != null) { + uniqueKeyHandler.removeUniqueKeysForVertexId(instanceVertex.getId()); + } + deleteVertex(instanceVertex, force); } diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasGraphUtilsV2.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasGraphUtilsV2.java index 5967c4b3381..2ab0972346b 100644 --- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasGraphUtilsV2.java +++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasGraphUtilsV2.java @@ -29,12 +29,14 @@ import org.apache.atlas.model.typedef.AtlasBaseTypeDef; import org.apache.atlas.model.typedef.AtlasEnumDef; import org.apache.atlas.repository.Constants; +import org.apache.atlas.repository.graph.GraphBackedSearchIndexer; import org.apache.atlas.repository.graph.GraphHelper; import org.apache.atlas.repository.graphdb.AtlasEdge; import org.apache.atlas.repository.graphdb.AtlasElement; import org.apache.atlas.repository.graphdb.AtlasGraph; import org.apache.atlas.repository.graphdb.AtlasGraphQuery; import org.apache.atlas.repository.graphdb.AtlasIndexQuery; +import org.apache.atlas.repository.graphdb.AtlasUniqueKeyHandler; import org.apache.atlas.repository.graphdb.AtlasVertex; import org.apache.atlas.type.AtlasEntityType; import org.apache.atlas.type.AtlasEnumType; @@ -229,17 +231,61 @@ public static void setProperty(T element, String proper LOG.debug("==> setProperty({}, {}, {})", toString(element), propertyName, value); } + AtlasUniqueKeyHandler uniqueKeyHandler = getGraphInstance().getUniqueKeyHandler(); + if (!isEncoded) { propertyName = encodePropertyKey(propertyName); } if (value == null) { + if (uniqueKeyHandler != null) { + if (GraphBackedSearchIndexer.isTypeUniqueIndexKey(propertyName)) { + String typeName = getProperty(element, TYPE_NAME_PROPERTY_KEY, String.class); + Object propValue = getProperty(element, propertyName, Object.class); + + uniqueKeyHandler.removeTypeUniqueKey(typeName, propertyName, propValue, element instanceof AtlasVertex); + } else if (GraphBackedSearchIndexer.isGlobalUniqueIndexKey(propertyName)) { + Object propValue = getProperty(element, propertyName, Object.class); + + uniqueKeyHandler.removeUniqueKey(propertyName, propValue, element instanceof AtlasVertex); + } + } + element.removeProperty(propertyName); } else { if (value instanceof Date) { value = ((Date) value).getTime(); } + if (uniqueKeyHandler != null) { + if (GraphBackedSearchIndexer.isTypeUniqueIndexKey(propertyName)) { + Object existingValue = getProperty(element, propertyName, Object.class); + String typeName = getProperty(element, TYPE_NAME_PROPERTY_KEY, String.class); + + if (existingValue != null) { + if (!existingValue.equals(value)) { + // remove the existing value from unique key index + uniqueKeyHandler.removeTypeUniqueKey(typeName, propertyName, existingValue, element instanceof AtlasVertex); + uniqueKeyHandler.addTypeUniqueKey(typeName, propertyName, value, element.getId(), element instanceof AtlasVertex); + } + } else { + uniqueKeyHandler.addTypeUniqueKey(typeName, propertyName, value, element.getId(), element instanceof AtlasVertex); + } + } else if (GraphBackedSearchIndexer.isGlobalUniqueIndexKey(propertyName)) { + String existingValue = getProperty(element, propertyName, String.class); + + if (existingValue != null) { + if (!existingValue.equals(value)) { + // remove the existing value from global unique key index + uniqueKeyHandler.removeUniqueKey(propertyName, existingValue, element instanceof AtlasVertex); + uniqueKeyHandler.addUniqueKey(propertyName, value, element.getId(), element instanceof AtlasVertex); + } + } else { + uniqueKeyHandler.addUniqueKey(propertyName, value, element.getId(), element instanceof AtlasVertex); + } + } + } + element.setProperty(propertyName, value); } }