/*
 * Copyright (c) "Neo4j"
 * Neo4j Sweden AB [http://neo4j.com]
 *
 * This file is part of Neo4j.
 *
 * Licensed 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 apoc.meta;

import static apoc.util.MapUtil.map;
import static apoc.util.TestUtil.testCall;
import static apoc.util.TestUtil.testResult;
import static java.util.Arrays.asList;
import static java.util.Collections.emptyMap;
import static java.util.Collections.singletonMap;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.neo4j.configuration.SettingImpl.newBuilder;
import static org.neo4j.configuration.SettingValueParsers.BOOL;
import static org.neo4j.driver.Values.isoDuration;
import static org.neo4j.graphdb.traversal.Evaluators.toDepth;

import apoc.HelperProcedures;
import apoc.graph.Graphs;
import apoc.nodes.Nodes;
import apoc.util.MapUtil;
import apoc.util.TestUtil;
import apoc.util.Util;
import apoc.util.collection.Iterables;
import com.neo4j.test.extension.EnterpriseDbmsExtension;
import java.io.InputStreamReader;
import java.time.Clock;
import java.time.LocalDate;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import org.apache.commons.io.IOUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.neo4j.configuration.GraphDatabaseInternalSettings;
import org.neo4j.configuration.GraphDatabaseSettings;
import org.neo4j.graphdb.*;
import org.neo4j.test.TestDatabaseManagementServiceBuilder;
import org.neo4j.test.extension.ExtensionCallback;
import org.neo4j.test.extension.Inject;
import org.neo4j.values.storable.CoordinateReferenceSystem;
import org.neo4j.values.storable.DateTimeValue;
import org.neo4j.values.storable.DateValue;
import org.neo4j.values.storable.LocalDateTimeValue;
import org.neo4j.values.storable.LocalTimeValue;
import org.neo4j.values.storable.TimeValue;
import org.neo4j.values.storable.Values;

@EnterpriseDbmsExtension(configurationCallback = "configure")
public class MetaTest {

    @Inject
    GraphDatabaseService db;

    @ExtensionCallback
    void configure(TestDatabaseManagementServiceBuilder builder) {
        builder.setConfig(
                        GraphDatabaseSettings.procedure_unrestricted,
                        List.of(
                                "apoc.meta.nodes.count",
                                "apoc.meta.stats",
                                "apoc.meta.data",
                                "apoc.meta.schema",
                                "apoc.meta.nodeTypeProperties",
                                "apoc.meta.relTypeProperties",
                                "apoc.meta.graph",
                                "apoc.meta.graph.of",
                                "apoc.meta.graphSample",
                                "apoc.meta.subGraph"))
                .setConfig(GraphDatabaseInternalSettings.latest_kernel_version, Byte.MAX_VALUE)
                .setConfig(GraphDatabaseInternalSettings.latest_runtime_version, Integer.MAX_VALUE)
                .setConfig(
                        newBuilder("internal.dbms.debug.track_cursor_close", BOOL, false)
                                .build(),
                        false)
                .setConfig(
                        newBuilder("internal.dbms.debug.trace_cursors", BOOL, false)
                                .build(),
                        false);
    }

    @BeforeAll
    void setUp() {
        TestUtil.registerProcedure(
                db, Meta.class, MetaRestricted.class, Graphs.class, Nodes.class, HelperProcedures.class);
    }

    public static boolean hasRecordMatching(List<Map<String, Object>> records, Map<String, Object> record) {
        return hasRecordMatching(records, row -> {
            boolean okSoFar = true;

            for (String k : record.keySet()) {
                okSoFar = okSoFar
                        && row.containsKey(k)
                        && (row.get(k) == null
                                ? (record.get(k) == null)
                                : row.get(k).equals(record.get(k)));
            }

            return okSoFar;
        });
    }

    public static boolean hasRecordMatching(
            List<Map<String, Object>> records, Predicate<Map<String, Object>> predicate) {
        return records.stream().anyMatch(predicate);
    }

    public static List<Map<String, Object>> gatherRecords(Result r) {
        List<Map<String, Object>> rows = new ArrayList<>();
        while (r.hasNext()) {
            Map<String, Object> row = r.next();
            rows.add(row);
        }
        return rows;
    }
    // Can be valuable for debugging purposes
    @SuppressWarnings("unused")
    private static String toCSV(List<Map<String, Object>> list) {
        List<String> headers =
                list.stream().flatMap(map -> map.keySet().stream()).distinct().toList();
        final StringBuilder sb = new StringBuilder();
        for (int i = 0; i < headers.size(); i++) {
            sb.append(headers.get(i));
            sb.append(i == headers.size() - 1 ? "\n" : ",");
        }
        for (Map<String, Object> map : list) {
            for (int i = 0; i < headers.size(); i++) {
                sb.append(map.get(headers.get(i)));
                sb.append(i == headers.size() - 1 ? "\n" : ",");
            }
        }
        return sb.toString();
    }

    public static boolean testDBCallEquivalence(GraphDatabaseService db, String testCall, String equivalentToCall) {
        AtomicReference<List<Map<String, Object>>> compareTo = new AtomicReference<>();
        AtomicReference<List<Map<String, Object>>> testSet = new AtomicReference<>();

        TestUtil.testResult(db, equivalentToCall, r -> compareTo.set(gatherRecords(r)));

        TestUtil.testResult(db, testCall, r -> testSet.set(gatherRecords(r)));

        // Uncomment this for debugging purposes
        /*
        System.out.println("COMPARE TO:");
        System.out.println(toCSV(compareTo.get()));
        System.out.println("TEST SET:");
        System.out.println(toCSV(testSet.get()));
        */

        return resultSetsEquivalent(compareTo.get(), testSet.get());
    }

    public static boolean resultSetsEquivalent(List<Map<String, Object>> baseSet, List<Map<String, Object>> testSet) {
        if (baseSet.size() != testSet.size()) {
            System.err.println("Result sets have different cardinality");
            return false;
        }

        boolean allMatch = true;

        for (Map<String, Object> baseRecord : baseSet) {
            allMatch = allMatch && hasRecordMatching(testSet, baseRecord);
        }

        return allMatch;
    }

    @Test
    void testMetaGraphExtraRels() {
        db.executeTransactionally(
                """
                CREATE (a:S1 {SomeName1:'aaa'})
                CREATE (b:S2 {SomeName2:'bbb'})
                CREATE (c:S3 {SomeName3:'ccc'})
                CREATE (a)-[:HAS]->(b)
                CREATE (b)-[:HAS]->(c)""");

        testCall(db, "call apoc.meta.graph()", (row) -> {
            List<Node> nodes = (List<Node>) row.get("nodes");
            List<Relationship> relationships = (List<Relationship>) row.get("relationships");
            assertEquals(3, nodes.size());
            assertEquals(2, relationships.size());
        });
    }

    @Test
    void testMetaGraphMaxRels() {
        db.executeTransactionally("CREATE (:S2 {id:'another'}), (:S2 {id:'another2'}), (:S2 {id:'another3'}), \n" +
                // create nodes to be linked
                "(a:S1 {id:'aaa'}), (b:S2 {id:'bbb'}), (c:S3 {id:'ccc'}), (d:S4 {id:'ddd'}), "
                + "(e:S5 {id:'eee'}), (f:S6 {id:'fff'}), (g:S7 {id:'ggg'}),"
                +
                // create rels
                "(a)-[:HAS]->(b), (a)-[:HAS]->(c), (a)-[:HAS]->(d), (a)-[:HAS]->(e), (a)-[:HAS]->(f), (a)-[:HAS]->(g),"
                + "(b)-[:HAS]->(c), (b)-[:HAS]->(d), (b)-[:HAS]->(e), (b)-[:HAS]->(f), (b)-[:HAS]->(g)");

        testCall(db, "call apoc.meta.graph()", (row) -> {
            List<Relationship> relationships = (List<Relationship>) row.get("relationships");
            assertEquals(11, relationships.size());
        });

        testCall(db, "call apoc.meta.graph({maxRels: 1})", (row) -> {
            List<Relationship> relationships = (List<Relationship>) row.get("relationships");
            assertEquals(8, relationships.size());
        });

        db.executeTransactionally("MATCH (n) DETACH DELETE n");
    }

    @Test
    void testRelationshipExistsSampling() {
        List<Long> skipSizes = List.of(2L, 100L, 1000L, -1L, 0L, 1L);
        List<Long> fromNodeCounts = List.of(4L, 1000L, 100L, 500L, 10000L, 4L);
        List<Integer> expectedChecks = List.of(2, 10, 1, 500, 10000, 4);

        for (int i = 0; i < skipSizes.size(); i++) {
            testSampling(skipSizes.get(i), fromNodeCounts.get(i), expectedChecks.get(i));
        }
    }

    public void testSampling(Long skipSize, Long fromNodeCount, int expectedChecks) {
        SampleMetaConfig config = Mockito.mock(SampleMetaConfig.class);
        Mockito.when(config.getSample()).thenReturn(skipSize);

        Label labelFromLabel = Label.label("A");
        Label labelToLabel = Label.label("B");
        RelationshipType relationshipType = RelationshipType.withName("R");
        Direction direction = Direction.OUTGOING;

        Transaction tx = Mockito.mock(Transaction.class);
        ResourceIterator<Node> nodes = Mockito.mock(ResourceIterator.class);
        ResourceIterable<Relationship> relationships = Mockito.mock(ResourceIterable.class);
        ResourceIterator<Relationship> relationshipIterator = Mockito.mock(ResourceIterator.class);
        Node node = Mockito.mock(Node.class);

        Mockito.when(tx.findNodes(labelFromLabel)).thenReturn(nodes);
        List<Boolean> nodesNext = new ArrayList<>(Arrays.asList(new Boolean[fromNodeCount.intValue()]));
        Collections.fill(nodesNext, Boolean.TRUE);
        nodesNext.set(fromNodeCount.intValue() - 1, false);
        Mockito.when(nodes.hasNext()).thenReturn(true, nodesNext.toArray(Boolean[]::new));
        Mockito.when(nodes.next()).thenReturn(node);

        Mockito.when(node.getRelationships(direction, relationshipType)).thenReturn(relationships);
        Mockito.when(relationships.iterator()).thenReturn(relationshipIterator);

        assertFalse(Meta.relationshipExists(tx, labelFromLabel, labelToLabel, relationshipType, direction, config));
        Mockito.verify(nodes, Mockito.times(fromNodeCount.intValue())).next();

        Mockito.verify(node, Mockito.times(expectedChecks)).getRelationships(Mockito.any(), Mockito.any());
    }

    @Test
    void testMetaType() {
        try (Transaction tx = db.beginTx()) {
            Node node = tx.createNode();
            Relationship rel = node.createRelationshipTo(node, RelationshipType.withName("FOO"));
            testTypeName(node, "NODE");
            testTypeName(rel, "RELATIONSHIP");
            Path path = tx.traversalDescription()
                    .evaluator(toDepth(1))
                    .traverse(node)
                    .iterator()
                    .next();
            // TODO PATH FAILS              testTypeName(path, "PATH");
            tx.rollback();
        }
        testTypeName(singletonMap("a", 10), "MAP");
        testTypeName(asList(1, 2), "LIST OF INTEGER");
        testTypeName(1L, "INTEGER");
        testTypeName(1, "INTEGER");
        testTypeName(1.0D, "FLOAT");
        testTypeName(1.0, "FLOAT");
        testTypeName("a", "STRING");
        testTypeName(false, "BOOLEAN");
        testTypeName(true, "BOOLEAN");
        testTypeName(null, "NULL");
    }

    @Test
    void testMetaTypeArray() {
        testTypeName(asList(1, 2), "LIST OF INTEGER");
        testTypeName(asList(LocalDate.of(2018, 1, 1), 2), "LIST OF ANY");
        testTypeName(new Integer[] {1, 2}, "LIST OF INTEGER");
        testTypeName(new Float[] {1f, 2f}, "LIST OF FLOAT");
        testTypeName(new Double[] {1d, 2d}, "LIST OF FLOAT");
        testTypeName(new String[] {"a", "b"}, "LIST OF STRING");
        testTypeName(new Long[] {1L, 2L}, "LIST OF INTEGER");
        testTypeName(new LocalDate[] {LocalDate.of(2018, 1, 1), LocalDate.of(2018, 1, 1)}, "LIST OF DATE");
        testTypeName(new Object[] {1d, ""}, "LIST OF ANY");
    }

    @Test
    void testMetaIsType() {
        try (Transaction tx = db.beginTx()) {
            Node node = tx.createNode();
            Relationship rel = node.createRelationshipTo(node, RelationshipType.withName("FOO"));
            testIsTypeName(node, "NODE");
            testIsTypeName(rel, "RELATIONSHIP");
            Path path = tx.traversalDescription()
                    .evaluator(toDepth(1))
                    .traverse(node)
                    .iterator()
                    .next();
            // TODO PATH FAILS            testIsTypeName(path, "PATH");
            tx.rollback();
        }
        testIsTypeName(singletonMap("a", 10), "MAP");
        testIsTypeName(asList(1, 2), "LIST OF INTEGER");
        testIsTypeName(1L, "INTEGER");
        testIsTypeName(1, "INTEGER");
        testIsTypeName(1.0D, "FLOAT");
        testIsTypeName(1.0, "FLOAT");
        testIsTypeName("a", "STRING");
        testIsTypeName(false, "BOOLEAN");
        testIsTypeName(true, "BOOLEAN");
        testIsTypeName(null, "NULL");
    }

    @Test
    void testMetaTypes() {

        Map<String, Object> param = map(
                "MAP",
                singletonMap("a", 10),
                "LIST OF INTEGER",
                asList(1, 2),
                "INTEGER",
                1L,
                "FLOAT",
                1.0D,
                "STRING",
                "a",
                "BOOLEAN",
                true,
                "NULL",
                null);
        TestUtil.testCall(db, "RETURN apoc.meta.cypher.types($param) AS value", singletonMap("param", param), row -> {
            Map<String, String> res = (Map) row.get("value");
            res.forEach(Assertions::assertEquals);
        });
    }

    private void testTypeName(Object value, String type) {
        TestUtil.testCall(
                db,
                "RETURN apoc.meta.cypher.type($value) AS value",
                singletonMap("value", value),
                row -> assertEquals(type, row.get("value")));
    }

    @Test
    void testVectorTypes() {
        var vectorTypes = List.of("INT64", "INT32", "INT16", "INT8", "FLOAT64", "FLOAT32");
        for (String type : vectorTypes) {
            TestUtil.testCall(
                    db,
                    """
                            CYPHER 25
                            WITH VECTOR([1, 2, 3], 3, %s) AS v
                            RETURN apoc.meta.cypher.type(v) AS value"""
                            .formatted(type),
                    row -> assertEquals("VECTOR", row.get("value")));

            TestUtil.testCall(
                    db,
                    """
                            CYPHER 25
                            WITH VECTOR([1, 2, 3], 3, %s) AS v
                            RETURN apoc.meta.cypher.isType(v, "VECTOR") AS value"""
                            .formatted(type),
                    row -> assertEquals(true, row.get("value")));
        }
    }

    private void testIsTypeName(Object value, String type) {
        TestUtil.testCall(
                db,
                "RETURN apoc.meta.cypher.isType($value,$type) AS value",
                map("value", value, "type", type),
                result -> assertEquals(true, result.get("value")));
        TestUtil.testCall(
                db,
                "RETURN apoc.meta.cypher.isType($value,$type) AS value",
                map("value", value, "type", type + "foo"),
                result -> assertEquals(false, result.get("value")));
    }

    private void assertStats(String setupQuery, Map<String, Object> expected) {
        // Stats works on committed data
        db.executeTransactionally(setupQuery);
        try (final var tx = db.beginTx()) {
            assertThat(tx.execute("CALL apoc.meta.stats()").stream().toList())
                    .satisfiesExactly(row -> assertThat(row).containsExactlyInAnyOrderEntriesOf(expected));
            tx.commit();
        }

        // Stats works on uncommited data
        db.executeTransactionally("CREATE (:UnrelatedLabel)-[:UNRELATED_REL]->()");
        try (final var tx = db.beginTx()) {
            assertThat(tx.execute("MATCH (n) DETACH DELETE n").stream().toList())
                    .size()
                    .isLessThanOrEqualTo(0);
            assertThat(tx.execute(setupQuery).stream().toList()).size().isGreaterThanOrEqualTo(0);
            assertThat(tx.execute("CALL apoc.meta.stats()").stream().toList())
                    .satisfiesExactly(row -> assertThat(row).containsExactlyInAnyOrderEntriesOf(expected));
            tx.commit();
        }
    }

    private Map<String, Object> statsMap(Map<String, Object> values) {
        final var stats = values.entrySet().stream()
                .filter(e -> !"relTypesCount".equals(e.getKey()))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
        final var result = new HashMap<String, Object>(values);
        result.put("stats", stats);
        return Collections.unmodifiableMap(result);
    }

    @Test
    void testMetaStats() {
        final var setup =
                "CREATE (:Actor)-[:ACTED_IN]->(:Movie), ()-[:ACTED_IN]->(:Movie), (:Actor)-[:ACTED_IN]->(), ()-[:ACTED_IN]->()";
        final var expected = statsMap(Map.of(
                "relTypeCount", 1L,
                "propertyKeyCount", 0L,
                "labelCount", 2L,
                "nodeCount", 8L,
                "relCount", 4L,
                "labels", Map.of("Movie", 2L, "Actor", 2L),
                "relTypes",
                        Map.of(
                                "()-[:ACTED_IN]->(:Movie)",
                                2L,
                                "()-[:ACTED_IN]->()",
                                4L,
                                "(:Actor)-[:ACTED_IN]->()",
                                2L),
                "relTypesCount", Map.of("ACTED_IN", 4L)));
        assertStats(setup, expected);
    }

    @Test
    void testMetaStats2() {
        final var nodeLabels = List.of("", ":A", ":B", ":A:B");
        final var setup = new StringBuilder();
        for (final var labelsA : nodeLabels) {
            setup.append("CREATE (%s)%n".formatted(labelsA));
            for (final var labelsB : nodeLabels) {
                setup.append("CREATE (%s)-[:R1]->(%s)%n".formatted(labelsA, labelsB));
                setup.append("CREATE (%s)<-[:R2]-(%s)%n".formatted(labelsA, labelsB));
            }
        }
        final var expected = statsMap(Map.of(
                "relTypeCount", 2L,
                "propertyKeyCount", 0L,
                "labelCount", 2L,
                "nodeCount", 68L,
                "relCount", 32L,
                "labels", Map.of("A", 34L, "B", 34L),
                "relTypes",
                        Map.of(
                                "()-[:R1]->()", 16L,
                                "()-[:R2]->()", 16L,
                                "()-[:R1]->(:A)", 8L,
                                "()-[:R1]->(:B)", 8L,
                                "()-[:R2]->(:A)", 8L,
                                "()-[:R2]->(:B)", 8L,
                                "(:A)-[:R1]->()", 8L,
                                "(:A)-[:R2]->()", 8L,
                                "(:B)-[:R1]->()", 8L,
                                "(:B)-[:R2]->()", 8L),
                "relTypesCount", Map.of("R1", 16L, "R2", 16L)));
        assertStats(setup.toString(), expected);
    }

    @Test
    void testMetaGraph() {
        db.executeTransactionally("CREATE (a:Actor)-[:ACTED_IN]->(m1:Movie),(a)-[:ACTED_IN]->(m2:Movie)");
        TestUtil.testCall(db, "CALL apoc.meta.graph()", (row) -> {
            List<Node> nodes = (List<Node>) row.get("nodes");
            Node n1 = nodes.getFirst();
            assertTrue(n1.hasLabel(Label.label("Actor")));
            assertEquals(1L, n1.getProperty("count"));
            assertEquals("Actor", n1.getProperty("name"));
            Node n2 = nodes.get(1);
            assertTrue(n2.hasLabel(Label.label("Movie")));
            assertEquals("Movie", n2.getProperty("name"));
            assertEquals(2L, n2.getProperty("count"));
            List<Relationship> rels = (List<Relationship>) row.get("relationships");
            Relationship rel = rels.getFirst();
            assertEquals("ACTED_IN", rel.getType().name());
            assertEquals(2L, rel.getProperty("count"));
        });
    }

    @Test
    void testMetaGraph2() {
        db.executeTransactionally("CREATE (:Actor)-[:ACTED_IN]->(:Movie) ");
        TestUtil.testCall(db, "CALL apoc.meta.graphSample()", (row) -> {
            List<Node> nodes = (List<Node>) row.get("nodes");
            Node n1 = nodes.get(0);
            assertTrue(n1.hasLabel(Label.label("Actor")));
            assertEquals(1L, n1.getProperty("count"));
            assertEquals("Actor", n1.getProperty("name"));
            Node n2 = nodes.get(1);
            assertTrue(n2.hasLabel(Label.label("Movie")));
            assertEquals("Movie", n2.getProperty("name"));
            assertEquals(1L, n1.getProperty("count"));
            List<Relationship> rels = (List<Relationship>) row.get("relationships");
            Relationship rel = rels.iterator().next();
            assertEquals("ACTED_IN", rel.getType().name());
            assertEquals(1L, rel.getProperty("count"));
        });
    }

    @Test
    void testVectorIndexes() {
        db.executeTransactionally(
                """
                CREATE VECTOR INDEX moviePlots IF NOT EXISTS
                FOR (m:Movie)
                ON m.embedding
                OPTIONS { indexConfig: {
                 `vector.dimensions`: 1536,
                 `vector.similarity_function`: 'cosine'
                }}
                """);

        db.executeTransactionally("CREATE (m:Movie {name:'The Movie', embedding: [1, 2, 3]})");

        TestUtil.testResult(
                db,
                """
                CALL apoc.meta.data()
                YIELD label, property, count, unique, index, existence, type, array, left, right, other, otherLabels, elementType
                RETURN * ORDER BY elementType, property""",
                (r) -> {
                    Map<String, Object> row = r.next();
                    assertEquals("node", row.get("elementType"));
                    assertEquals("embedding", row.get("property"));
                    assertEquals("Movie", row.get("label"));
                    assertEquals(true, row.get("index"));
                    row = r.next();
                    assertEquals("node", row.get("elementType"));
                    assertEquals("name", row.get("property"));
                    assertEquals("Movie", row.get("label"));
                    assertEquals(false, row.get("index"));
                    assertFalse(r.hasNext());
                });
    }

    @Test
    void testMetaData() {
        db.executeTransactionally("create index for (n:Movie) on (n.title)");
        db.executeTransactionally("create constraint for (a:Actor) require a.name is unique");
        db.executeTransactionally(
                """
                CREATE (actor1:Actor {name:'Tom Hanks'})-[:ACTED_IN {roles:'Forrest'}]->(movie1:Movie {title:'Forrest Gump'}),
                (actor2:Actor {name: 'Bruce Lee'})-[:ACTED_IN {roles:'FooBaz'}]->(movie1),
                (actor1)-[:ACTED_IN {roles:'Movie2Role'}]->(movie2:Movie {title:'Movie2'}), (actor1)-[:ACTED_IN {roles:'Movie3Role'}]->(movie3:Movie {title:'Movie3'}),
                (actor1)-[:DIRECTED {foo: 'first'}]->(movie2), (actor1)-[:DIRECTED {foo: 'second'}]->(:Movie {title:'Movie4'}),
                (:Studio {name: 'Pixar'})-[:ANIMATED {bar: 'alpha'}]->(movie2)""");
        TestUtil.testResult(
                db,
                """
                CALL apoc.meta.data()
                YIELD label, property, count, unique, index, existence, type, array, left, right, other, otherLabels, elementType
                RETURN * ORDER BY elementType, property""",
                (r) -> {
                    Map<String, Object> row = r.next();
                    assertEquals("node", row.get("elementType"));
                    assertEquals("ACTED_IN", row.get("property"));
                    assertEquals("Actor", row.get("label"));
                    assertRelationshipActedInMetaData(row);
                    row = r.next();
                    assertEquals("node", row.get("elementType"));
                    assertEquals("ANIMATED", row.get("property"));
                    assertEquals("Studio", row.get("label"));
                    assertRelationshipsAnimatedMetaData(row);
                    row = r.next();
                    assertEquals("node", row.get("elementType"));
                    assertEquals("DIRECTED", row.get("property"));
                    assertEquals("Actor", row.get("label"));
                    assertRelationshipsDirectedMetaData(row);
                    row = r.next();
                    assertEquals("node", row.get("elementType"));
                    assertPropertiesMetaData(row);
                    row = r.next();
                    assertEquals("node", row.get("elementType"));
                    assertPropertiesMetaData(row);
                    row = r.next();
                    assertEquals("node", row.get("elementType"));
                    assertPropertiesMetaData(row);
                    row = r.next();
                    assertEquals("relationship", row.get("elementType"));
                    assertEquals("ACTED_IN", row.get("label"));
                    assertEquals("Actor", row.get("property"));
                    assertRelationshipActedInMetaData(row);
                    row = r.next();
                    assertEquals("relationship", row.get("elementType"));
                    assertEquals("DIRECTED", row.get("label"));
                    assertEquals("Actor", row.get("property"));
                    assertRelationshipsDirectedMetaData(row);
                    row = r.next();
                    assertEquals("relationship", row.get("elementType"));
                    assertEquals("ANIMATED", row.get("label"));
                    assertEquals("Studio", row.get("property"));
                    assertRelationshipsAnimatedMetaData(row);
                    row = r.next();
                    assertEquals("relationship", row.get("elementType"));
                    assertPropertiesMetaData(row);
                    row = r.next();
                    assertEquals("relationship", row.get("elementType"));
                    assertPropertiesMetaData(row);
                    row = r.next();
                    assertEquals("relationship", row.get("elementType"));
                    assertPropertiesMetaData(row);
                    assertFalse(r.hasNext());
                });
    }

    private void assertRelationshipsDirectedMetaData(Map<String, Object> row) {
        assertRowMetaData(row, 1L, 2L, 0L, Types.RELATIONSHIP);
    }

    private void assertRelationshipsAnimatedMetaData(Map<String, Object> row) {
        assertRowMetaData(row, 1L, 1L, 0L, Types.RELATIONSHIP);
    }

    private void assertRelationshipActedInMetaData(Map<String, Object> row) {
        assertRowMetaData(row, 2L, 2L, 0L, Types.RELATIONSHIP);
    }

    private void assertPropertiesMetaData(Map<String, Object> row) {
        assertRowMetaData(row, 0L, 0L, 0L, Types.STRING);
    }

    private void assertRowMetaData(Map<String, Object> row, long count, long left, long right, Types type) {
        assertEquals(count, row.get("count"));
        assertEquals(left, row.get("left"));
        assertEquals(right, row.get("right"));
        assertEquals(type.name(), row.get("type"));
    }

    @Test
    void testOutgoingRelsAreCorrectMetaSchema() {
        db.executeTransactionally("CREATE (:A)-[:HAS]->(:B), (:A)<-[:HAS]-(:B), (:A)<-[:HAS]-(:C), (:D)<-[:HAS]-(:E)");

        testCall(db, "CALL apoc.meta.schema", (row) -> {
            Map<String, Object> o = (Map<String, Object>) row.get("value");
            // HAS: count 3, no props, type relationship
            // A: count 3. no props, relationship out as B
            // B: count 2. no props, relationship out as A
            // C: count 1. no props, relationship out as A
            // D: count 1. no props, relationship in as E
            // E: count 1. no props, relationship out as D
            Map<String, Object> relHas = (Map<String, Object>) o.get("HAS");
            Map<String, Object> nodeA = (Map<String, Object>) o.get("A");
            Map<String, Object> nodeB = (Map<String, Object>) o.get("B");
            Map<String, Object> nodeC = (Map<String, Object>) o.get("C");
            Map<String, Object> nodeD = (Map<String, Object>) o.get("D");
            Map<String, Object> nodeE = (Map<String, Object>) o.get("E");
            // Check counts
            assertEquals(4L, relHas.get("count"));
            assertEquals(3L, nodeA.get("count"));
            assertEquals(2L, nodeB.get("count"));
            assertEquals(1L, nodeC.get("count"));
            assertEquals(1L, nodeD.get("count"));
            assertEquals(1L, nodeE.get("count"));

            Map<String, Object> nodeARels =
                    (Map<String, Object>) ((Map<String, Object>) nodeA.get("relationships")).get("HAS");
            Map<String, Object> nodeBRels =
                    (Map<String, Object>) ((Map<String, Object>) nodeB.get("relationships")).get("HAS");
            Map<String, Object> nodeCRels =
                    (Map<String, Object>) ((Map<String, Object>) nodeC.get("relationships")).get("HAS");
            Map<String, Object> nodeDRels =
                    (Map<String, Object>) ((Map<String, Object>) nodeD.get("relationships")).get("HAS");
            Map<String, Object> nodeERels =
                    (Map<String, Object>) ((Map<String, Object>) nodeE.get("relationships")).get("HAS");
            // Check directions
            assertEquals("out", nodeARels.get("direction"));
            assertEquals("out", nodeBRels.get("direction"));
            assertEquals("out", nodeCRels.get("direction"));
            assertEquals("in", nodeDRels.get("direction"));
            assertEquals("out", nodeERels.get("direction"));

            // Check end nodes
            assertEquals(List.of("B"), nodeARels.get("labels"));
            assertEquals(List.of("A"), nodeBRels.get("labels"));
            assertEquals(List.of("A"), nodeCRels.get("labels"));
            assertEquals(List.of("E"), nodeDRels.get("labels"));
            assertEquals(List.of("D"), nodeERels.get("labels"));
        });
    }

    @Test
    void testMetaSchema() {
        db.executeTransactionally("create index for (n:Movie) on (n.title)");
        db.executeTransactionally("create constraint for (p:Person) require p.name is unique");
        db.executeTransactionally(
                "CREATE (:Person:Actor:Director {name:'Tom', born:'05-06-1956', dead:false})-[:ACTED_IN {roles:'Forrest'}]->(:Movie {title:'Forrest Gump'})");
        testCall(db, "CALL apoc.meta.schema()", (row) -> {
            List<String> emprtyList = new ArrayList<>();
            List<String> fullList = Arrays.asList("Actor", "Director");

            Map<String, Object> o = (Map<String, Object>) row.get("value");
            assertEquals(5, o.size());

            Map<String, Object> movie = (Map<String, Object>) o.get("Movie");
