diff --git a/community/community-it/cypher-it/src/test/java/org/neo4j/cypher/QueryInvalidationIT.java b/community/community-it/cypher-it/src/test/java/org/neo4j/cypher/QueryInvalidationIT.java
index 7d8ab07cbd13d..3158d8fbb8752 100644
--- a/community/community-it/cypher-it/src/test/java/org/neo4j/cypher/QueryInvalidationIT.java
+++ b/community/community-it/cypher-it/src/test/java/org/neo4j/cypher/QueryInvalidationIT.java
@@ -28,11 +28,13 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.neo4j.cypher.internal.ParameterTypeMap;
 import org.neo4j.cypher.internal.compatibility.CypherCacheHitMonitor;
 import org.neo4j.graphdb.Label;
 import org.neo4j.graphdb.Result;
 import org.neo4j.graphdb.Transaction;
 import org.neo4j.graphdb.factory.GraphDatabaseSettings;
+import org.neo4j.helpers.collection.Pair;
 import org.neo4j.kernel.configuration.Config;
 import org.neo4j.kernel.monitoring.Monitors;
 import org.neo4j.test.rule.DatabaseRule;
@@ -186,7 +188,7 @@ private static int randomInt( int max )
         return ThreadLocalRandom.current().nextInt( max );
     }
 
-    private static class TestMonitor implements CypherCacheHitMonitor<String>
+    private static class TestMonitor implements CypherCacheHitMonitor<Pair<String,ParameterTypeMap>>
     {
         private final AtomicInteger hits = new AtomicInteger();
         private final AtomicInteger misses = new AtomicInteger();
@@ -194,19 +196,19 @@ private static class TestMonitor implements CypherCacheHitMonitor<String>
         private final AtomicLong waitTime = new AtomicLong();
 
         @Override
-        public void cacheHit( String key )
+        public void cacheHit( Pair<String,ParameterTypeMap> key )
         {
             hits.incrementAndGet();
         }
 
         @Override
-        public void cacheMiss( String key )
+        public void cacheMiss( Pair<String,ParameterTypeMap> key )
         {
             misses.incrementAndGet();
         }
 
         @Override
-        public void cacheDiscard( String key, String ignored, int secondsSinceReplan )
+        public void cacheDiscard( Pair<String,ParameterTypeMap> key, String ignored, int secondsSinceReplan )
         {
             discards.incrementAndGet();
             waitTime.addAndGet( secondsSinceReplan );
diff --git a/community/community-it/cypher-it/src/test/scala/org/neo4j/cypher/QueryCachingTest.scala b/community/community-it/cypher-it/src/test/scala/org/neo4j/cypher/QueryCachingTest.scala
index b75403c7ec547..e68ae698c6073 100644
--- a/community/community-it/cypher-it/src/test/scala/org/neo4j/cypher/QueryCachingTest.scala
+++ b/community/community-it/cypher-it/src/test/scala/org/neo4j/cypher/QueryCachingTest.scala
@@ -19,14 +19,14 @@
  */
 package org.neo4j.cypher
 
-import org.neo4j.cypher.internal.StringCacheMonitor
+import org.neo4j.cypher.internal.{ParameterTypeMap, StringCacheMonitor}
 import org.neo4j.graphdb.Label
 import org.neo4j.helpers.collection.Pair
 import org.neo4j.kernel.impl.core.ThreadToStatementContextBridge
-import org.neo4j.values.virtual.MapValue
 import org.opencypher.v9_0.util.test_helpers.CypherFunSuite
 import org.scalatest.prop.TableDrivenPropertyChecks
 
+import scala.collection.JavaConversions._
 import scala.collection.mutable
 
 class QueryCachingTest extends CypherFunSuite with GraphDatabaseTestSupport with TableDrivenPropertyChecks {
@@ -41,6 +41,7 @@ class QueryCachingTest extends CypherFunSuite with GraphDatabaseTestSupport with
     val query = "MATCH (n:Person) RETURN n"
     val profileQuery = s"PROFILE $query"
     val explainQuery = s"EXPLAIN $query"
+    val empty_parameters = "Map()"
 
     val modeCombinations = Table(
       ("firstQuery", "secondQuery"),
@@ -73,14 +74,80 @@ class QueryCachingTest extends CypherFunSuite with GraphDatabaseTestSupport with
         val actual = cacheListener.trace.map(str => str.replaceAll("\\s+", " "))
         val expected = List(
           s"cacheFlushDetected",
-          s"cacheMiss: CYPHER 3.5 $query",
-          s"cacheHit: CYPHER 3.5 $query",
-          s"cacheHit: CYPHER 3.5 $query")
+          s"cacheMiss: (CYPHER 3.5 $query, $empty_parameters)",
+          s"cacheHit: (CYPHER 3.5 $query, $empty_parameters)",
+          s"cacheHit: (CYPHER 3.5 $query, $empty_parameters)")
 
         actual should equal(expected)
     }
   }
 
+  test("repeating query with same parameters should hit the cache") {
+
+    val cacheListener = new LoggingStringCacheListener
+    kernelMonitors.addMonitorListener(cacheListener)
+
+    val query = "RETURN $n"
+    val params1: Map[String, AnyRef] = Map("n" -> Long.box(42))
+
+    graph.execute(query, params1).resultAsString()
+    graph.execute(query, params1).resultAsString()
+
+    val actual = cacheListener.trace.map(str => str.replaceAll("\\s+", " "))
+    val expected = List(
+      s"cacheFlushDetected",
+      s"cacheMiss: (CYPHER 3.5 $query, Map(n -> class org.neo4j.values.storable.LongValue))",
+      s"cacheHit: (CYPHER 3.5 $query, Map(n -> class org.neo4j.values.storable.LongValue))",
+      s"cacheHit: (CYPHER 3.5 $query, Map(n -> class org.neo4j.values.storable.LongValue))")
+
+    actual should equal(expected)
+  }
+
+  test("repeating query with same parameter types but different values should hit the cache") {
+
+    val cacheListener = new LoggingStringCacheListener
+    kernelMonitors.addMonitorListener(cacheListener)
+
+    val query = "RETURN $n"
+    val params1: Map[String, AnyRef] = Map("n" -> Long.box(12))
+    val params2: Map[String, AnyRef] = Map("n" -> Long.box(42))
+    graph.execute(query, params1).resultAsString()
+    graph.execute(query, params2).resultAsString()
+
+    val actual = cacheListener.trace.map(str => str.replaceAll("\\s+", " "))
+    val expected = List(
+      s"cacheFlushDetected",
+      s"cacheMiss: (CYPHER 3.5 $query, Map(n -> class org.neo4j.values.storable.LongValue))",
+      s"cacheHit: (CYPHER 3.5 $query, Map(n -> class org.neo4j.values.storable.LongValue))",
+      s"cacheHit: (CYPHER 3.5 $query, Map(n -> class org.neo4j.values.storable.LongValue))")
+
+    actual should equal(expected)
+  }
+
+  test("repeating query with different parameters types should not hit the cache") {
+
+    val cacheListener = new LoggingStringCacheListener
+    kernelMonitors.addMonitorListener(cacheListener)
+
+    val query = "RETURN $n"
+    val params1: Map[String, AnyRef] = Map("n" -> Long.box(42))
+    val params2: Map[String, AnyRef] = Map("n" -> "nope")
+
+    graph.execute(query, params1).resultAsString()
+    graph.execute(query, params2).resultAsString()
+
+    val actual = cacheListener.trace.map(str => str.replaceAll("\\s+", " "))
+    val expected = List(
+      s"cacheFlushDetected",
+      s"cacheMiss: (CYPHER 3.5 $query, Map(n -> class org.neo4j.values.storable.LongValue))",
+      s"cacheHit: (CYPHER 3.5 $query, Map(n -> class org.neo4j.values.storable.LongValue))",
+      s"cacheMiss: (CYPHER 3.5 $query, Map(n -> class org.neo4j.values.storable.StringWrappingStringValue))",
+      s"cacheHit: (CYPHER 3.5 $query, Map(n -> class org.neo4j.values.storable.StringWrappingStringValue))")
+
+    actual should equal(expected)
+
+  }
+
   private class LoggingStringCacheListener extends StringCacheMonitor {
     private var log: mutable.Builder[String, List[String]] = List.newBuilder
 
@@ -94,15 +161,15 @@ class QueryCachingTest extends CypherFunSuite with GraphDatabaseTestSupport with
       log += s"cacheFlushDetected"
     }
 
-    override def cacheHit(key: Pair[String, MapValue]): Unit = {
+    override def cacheHit(key: Pair[String, ParameterTypeMap]): Unit = {
       log += s"cacheHit: $key"
     }
 
-    override def cacheMiss(key: Pair[String, MapValue]): Unit = {
+    override def cacheMiss(key: Pair[String, ParameterTypeMap]): Unit = {
       log += s"cacheMiss: $key"
     }
 
-    override def cacheDiscard(key: Pair[String, MapValue], ignored: String, secondsSinceReplan: Int): Unit = {
+    override def cacheDiscard(key: Pair[String, ParameterTypeMap], ignored: String, secondsSinceReplan: Int): Unit = {
       log += s"cacheDiscard: $key"
     }
   }
diff --git a/community/community-it/cypher-it/src/test/scala/org/neo4j/cypher/internal/compatibility/v3_5/CypherCompilerAstCacheAcceptanceTest.scala b/community/community-it/cypher-it/src/test/scala/org/neo4j/cypher/internal/compatibility/v3_5/CypherCompilerAstCacheAcceptanceTest.scala
index 1d1839dbd69b1..e1d7cab25a18e 100644
--- a/community/community-it/cypher-it/src/test/scala/org/neo4j/cypher/internal/compatibility/v3_5/CypherCompilerAstCacheAcceptanceTest.scala
+++ b/community/community-it/cypher-it/src/test/scala/org/neo4j/cypher/internal/compatibility/v3_5/CypherCompilerAstCacheAcceptanceTest.scala
@@ -23,15 +23,15 @@ import java.time.{Clock, Instant, ZoneOffset}
 
 import org.neo4j.cypher
 import org.neo4j.cypher._
-import org.neo4j.cypher.internal.compiler.v3_5.{CypherPlannerConfiguration, StatsDivergenceCalculator}
 import org.neo4j.cypher.internal.compatibility.{CommunityRuntimeContextCreator, CypherCurrentCompiler, RuntimeContext}
+import org.neo4j.cypher.internal.compiler.v3_5.{CypherPlannerConfiguration, StatsDivergenceCalculator}
 import org.neo4j.cypher.internal.runtime.interpreted.CSVResources
-import org.neo4j.cypher.internal.{CacheTracer, CommunityRuntimeFactory, PreParsedQuery}
+import org.neo4j.cypher.internal.{CacheTracer, CommunityRuntimeFactory, ParameterTypeMap, PreParsedQuery}
 import org.neo4j.graphdb.config.Setting
 import org.neo4j.graphdb.factory.GraphDatabaseSettings
+import org.neo4j.helpers.collection.Pair
 import org.neo4j.logging.AssertableLogProvider.inLog
 import org.neo4j.logging.{AssertableLogProvider, Log, NullLog}
-import org.opencypher.v9_0.ast.Statement
 import org.opencypher.v9_0.frontend.phases.CompilationPhaseTracer
 import org.opencypher.v9_0.util.DummyPosition
 import org.opencypher.v9_0.util.test_helpers.CypherFunSuite
@@ -75,12 +75,12 @@ class CypherCompilerAstCacheAcceptanceTest extends CypherFunSuite with GraphData
     override def toString = s"hits = $hits, misses = $misses, flushes = $flushes, evicted = $evicted"
   }
 
-  class CacheCounter(var counts: CacheCounts = CacheCounts()) extends CacheTracer[Statement] {
-    override def queryCacheHit(key: Statement, metaData: String) {
+  class CacheCounter(var counts: CacheCounts = CacheCounts()) extends CacheTracer[Pair[AnyRef,ParameterTypeMap]] {
+    override def queryCacheHit(key: Pair[AnyRef,ParameterTypeMap], metaData: String) {
       counts = counts.copy(hits = counts.hits + 1)
     }
 
-    override def queryCacheMiss(key: Statement, metaData: String) {
+    override def queryCacheMiss(key: Pair[AnyRef,ParameterTypeMap], metaData: String) {
       counts = counts.copy(misses = counts.misses + 1)
     }
 
@@ -88,7 +88,7 @@ class CypherCompilerAstCacheAcceptanceTest extends CypherFunSuite with GraphData
       counts = counts.copy(flushes = counts.flushes + 1)
     }
 
-    override def queryCacheStale(key: Statement, secondsSincePlan: Int, metaData: String): Unit = {
+    override def queryCacheStale(key: Pair[AnyRef,ParameterTypeMap], secondsSincePlan: Int, metaData: String): Unit = {
       counts = counts.copy(evicted = counts.evicted + 1)
     }
   }
@@ -105,10 +105,10 @@ class CypherCompilerAstCacheAcceptanceTest extends CypherFunSuite with GraphData
     compiler.kernelMonitors.addMonitorListener(counter)
   }
 
-  private def runQuery(query: String, debugOptions: Set[String] = Set.empty): Unit = {
+  private def runQuery(query: String, debugOptions: Set[String] = Set.empty, params: scala.Predef.Map[String, Any] = Map.empty): Unit = {
     graph.withTx { tx =>
       val noTracing = CompilationPhaseTracer.NO_TRACING
-      val context = graph.transactionalContext(query = query -> Map.empty)
+      val context = graph.transactionalContext(query = query -> params)
       compiler.compile(PreParsedQuery(query, DummyPosition(0), query,
                                       isPeriodicCommit = false,
                                       CypherVersion.default,
@@ -260,4 +260,13 @@ class CypherCompilerAstCacheAcceptanceTest extends CypherFunSuite with GraphData
 
     counter.counts.hits should equal(0)
   }
+
+  test("should not find query in cache with different parameter types") {
+    val map1: scala.Predef.Map[String, Any] = scala.Predef.Map("number" -> 42)
+    val map2: scala.Predef.Map[String, Any] = scala.Predef.Map("number" -> "nope")
+    runQuery("return $number", params = map1)
+    runQuery("return $number", params = map2)
+
+    counter.counts should equal(CacheCounts(hits = 0, misses = 2, flushes = 1))
+  }
 }
diff --git a/community/cypher/cypher/src/main/java/org/neo4j/cypher/internal/javacompat/MonitoringCacheTracer.java b/community/cypher/cypher/src/main/java/org/neo4j/cypher/internal/javacompat/MonitoringCacheTracer.java
index 998efbb956699..a77cfe74e612e 100644
--- a/community/cypher/cypher/src/main/java/org/neo4j/cypher/internal/javacompat/MonitoringCacheTracer.java
+++ b/community/cypher/cypher/src/main/java/org/neo4j/cypher/internal/javacompat/MonitoringCacheTracer.java
@@ -20,14 +20,14 @@
 package org.neo4j.cypher.internal.javacompat;
 
 import org.neo4j.cypher.internal.CacheTracer;
+import org.neo4j.cypher.internal.ParameterTypeMap;
 import org.neo4j.cypher.internal.StringCacheMonitor;
 import org.neo4j.helpers.collection.Pair;
-import org.neo4j.values.virtual.MapValue;
 
 /**
  * Adapter for passing CacheTraces into the Monitoring infrastructure.
  */
-public class MonitoringCacheTracer implements CacheTracer<Pair<String,MapValue>>
+public class MonitoringCacheTracer implements CacheTracer<Pair<String,ParameterTypeMap>>
 {
     private final StringCacheMonitor monitor;
 
@@ -37,19 +37,19 @@ public MonitoringCacheTracer( StringCacheMonitor monitor )
     }
 
     @Override
-    public void queryCacheHit( Pair<String,MapValue> queryKey, String metaData )
+    public void queryCacheHit( Pair<String,ParameterTypeMap> queryKey, String metaData )
     {
         monitor.cacheHit( queryKey );
     }
 
     @Override
-    public void queryCacheMiss( Pair<String,MapValue> queryKey, String metaData )
+    public void queryCacheMiss( Pair<String,ParameterTypeMap> queryKey, String metaData )
     {
         monitor.cacheMiss( queryKey );
     }
 
     @Override
-    public void queryCacheStale( Pair<String,MapValue> queryKey, int secondsSincePlan, String metaData )
+    public void queryCacheStale( Pair<String,ParameterTypeMap> queryKey, int secondsSincePlan, String metaData )
     {
         monitor.cacheDiscard( queryKey, metaData, secondsSincePlan );
     }
diff --git a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/PlanCacheMetricsMonitor.scala b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/PlanCacheMetricsMonitor.scala
index 9263d874e4750..92c948a88ab6c 100644
--- a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/PlanCacheMetricsMonitor.scala
+++ b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/PlanCacheMetricsMonitor.scala
@@ -21,15 +21,14 @@ package org.neo4j.cypher
 
 import java.util.concurrent.atomic.AtomicLong
 
-import org.neo4j.cypher.internal.StringCacheMonitor
+import org.neo4j.cypher.internal.{ParameterTypeMap, StringCacheMonitor}
 import org.neo4j.helpers.collection.Pair
-import org.neo4j.values.virtual.MapValue
 
 class PlanCacheMetricsMonitor extends StringCacheMonitor {
   private val counter = new AtomicLong()
   private val waitTime = new AtomicLong()
 
-  override def cacheDiscard(ignored1: Pair[String, MapValue], ignored2: String, secondsSinceReplan: Int): Unit = {
+  override def cacheDiscard(ignored1: Pair[String, ParameterTypeMap], ignored2: String, secondsSinceReplan: Int): Unit = {
     counter.incrementAndGet()
     waitTime.addAndGet(secondsSinceReplan)
   }
diff --git a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/ExecutionEngine.scala b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/ExecutionEngine.scala
index 1400fac5f5ded..1bcb56bf7c577 100644
--- a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/ExecutionEngine.scala
+++ b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/ExecutionEngine.scala
@@ -35,7 +35,7 @@ import org.neo4j.kernel.monitoring.Monitors
 import org.neo4j.logging.LogProvider
 import org.neo4j.values.virtual.MapValue
 
-trait StringCacheMonitor extends CypherCacheMonitor[Pair[String, MapValue]]
+trait StringCacheMonitor extends CypherCacheMonitor[Pair[String, ParameterTypeMap]]
 
 /**
   * This class constructs and initializes both the cypher compilers and runtimes, which are very expensive
@@ -44,7 +44,7 @@ trait StringCacheMonitor extends CypherCacheMonitor[Pair[String, MapValue]]
 class ExecutionEngine(val queryService: GraphDatabaseQueryService,
                       val kernelMonitors: Monitors,
                       val tracer: CompilationTracer,
-                      val cacheTracer: CacheTracer[Pair[String, MapValue]],
+                      val cacheTracer: CacheTracer[Pair[String, ParameterTypeMap]],
                       val config: CypherConfiguration,
                       val compatibilityFactory: CompilerFactory,
                       val logProvider: LogProvider,
@@ -63,7 +63,7 @@ class ExecutionEngine(val queryService: GraphDatabaseQueryService,
   // Log on stale query discard from query cache
   private val log = logProvider.getLog( getClass )
   kernelMonitors.addMonitorListener( new StringCacheMonitor {
-    override def cacheDiscard(ignored: Pair[String, MapValue], query: String, secondsSinceReplan: Int) {
+    override def cacheDiscard(ignored: Pair[String, ParameterTypeMap], query: String, secondsSinceReplan: Int) {
       log.info(s"Discarded stale query from the query cache after ${secondsSinceReplan} seconds: $query")
     }
   })
diff --git a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/QueryCache.scala b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/QueryCache.scala
index 26a9bfa88deb5..288da0977878b 100644
--- a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/QueryCache.scala
+++ b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/QueryCache.scala
@@ -22,9 +22,11 @@ package org.neo4j.cypher.internal
 import com.github.benmanes.caffeine.cache.{Cache, Caffeine}
 import org.neo4j.helpers.collection.Pair
 import org.neo4j.kernel.impl.query.TransactionalContext
-import org.neo4j.values.storable.Values
 import org.neo4j.values.virtual.MapValue
 
+import scala.collection.JavaConversions._
+import scala.collection.mutable
+
 /**
   * The result of one cache lookup.
   */
@@ -60,34 +62,21 @@ trait CacheTracer[QUERY_KEY] {
   * @param tracer Traces cache activity
   */
 class QueryCache[QUERY_REP <: AnyRef, QUERY_KEY <: Pair[QUERY_REP, MapValue], EXECUTABLE_QUERY <: AnyRef](val maximumSize: Int,
-                                                                  val stalenessCaller: PlanStalenessCaller[EXECUTABLE_QUERY],
-                                                                  val tracer: CacheTracer[QUERY_KEY]) {
+                                                                                                          val stalenessCaller: PlanStalenessCaller[EXECUTABLE_QUERY],
+                                                                                                          val tracer: CacheTracer[Pair[QUERY_REP, ParameterTypeMap]]) {
+  type Cache_Key = Pair[QUERY_REP, ParameterTypeMap]
 
-  val inner: Cache[QUERY_KEY, EXECUTABLE_QUERY] = Caffeine.newBuilder().maximumSize(maximumSize).build[QUERY_KEY, EXECUTABLE_QUERY]()
-  val parameterMapping: Cache[QUERY_REP, List[MapValue]] = Caffeine.newBuilder().maximumSize(maximumSize).build[QUERY_REP,List[MapValue]]()
+  val inner: Cache[Cache_Key, EXECUTABLE_QUERY] = Caffeine.newBuilder().maximumSize(maximumSize).build[Cache_Key, EXECUTABLE_QUERY]()
 
   import QueryCache.NOT_PRESENT
 
-  /**
-    * TODO: Explain why this is used
-    * @param potentialQueryKey
-    * @return
-    */
 
-  private def getActualQueryKey(potentialQueryKey: QUERY_KEY): QUERY_KEY = {
-    val queryRep = potentialQueryKey.first()
-    val queryParams = potentialQueryKey.other()
-    val maybeParams = parameterMapping.getIfPresent(queryRep)
-
-    if (maybeParams != null) {
-      for (params <- maybeParams) {
-        if (Values.mapsEqualsOnValueTypes(params, queryParams)) {
-          // found! use old params so that key matches
-          return Pair.of(queryRep, params).asInstanceOf[QUERY_KEY]
-        }
-      }
+  private def extractParameterTypeMap(value: MapValue) = {
+    val resultMap = new ParameterTypeMap()
+    for( key <- value.keySet().iterator()) {
+      resultMap.put(key, value.get(key).getClass)
     }
-    potentialQueryKey
+    resultMap
   }
 
   /**
@@ -108,7 +97,7 @@ class QueryCache[QUERY_REP <: AnyRef, QUERY_KEY <: Pair[QUERY_REP, MapValue], EX
     if (maximumSize == 0)
       CacheDisabled(compile())
     else {
-      val actualKey = getActualQueryKey(queryKey)
+      val actualKey = Pair.of(queryKey.first(), extractParameterTypeMap(queryKey.other()))
       inner.getIfPresent(actualKey) match {
         case NOT_PRESENT =>
           compileAndCache(actualKey, tc, compile, metaData)
@@ -119,22 +108,12 @@ class QueryCache[QUERY_REP <: AnyRef, QUERY_KEY <: Pair[QUERY_REP, MapValue], EX
               hit(actualKey, executableQuery, metaData)
             case Stale(secondsSincePlan) =>
               tracer.queryCacheStale(actualKey, secondsSincePlan, metaData)
-              compileAndCacheForStale(actualKey, tc, compile, metaData)
+              compileAndCache(actualKey, tc, compile, metaData)
           }
       }
     }
   }
 
-  private def compileAndCacheForStale(queryKey: QUERY_KEY,
-                              tc: TransactionalContext,
-                              compile: () => EXECUTABLE_QUERY,
-                              metaData: String
-                             ): CacheLookup[EXECUTABLE_QUERY] = {
-    val oldList = parameterMapping.getIfPresent(queryKey.first()) // at this point it can't be null
-    val result = compileAndCache(queryKey,tc,compile,metaData)
-    parameterMapping.put(queryKey.first(), oldList ::: List(queryKey.other()))
-    result
-  }
 
   /**
     * Ensure this query is recompiled and put it in the cache.
@@ -143,28 +122,27 @@ class QueryCache[QUERY_REP <: AnyRef, QUERY_KEY <: Pair[QUERY_REP, MapValue], EX
     * first. Regardless of who does it, this is treated as a cache miss, because it will
     * take a long time.
     */
-  private def compileAndCache(queryKey: QUERY_KEY,
+  private def compileAndCache(key: Cache_Key,
                         tc: TransactionalContext,
                         compile: () => EXECUTABLE_QUERY,
                         metaData: String
                        ): CacheLookup[EXECUTABLE_QUERY] = {
     val newExecutableQuery = compile()
-    inner.put(queryKey, newExecutableQuery)
-    parameterMapping.put(queryKey.first(), List(queryKey.other()))
-    miss(queryKey, newExecutableQuery, metaData)
+    inner.put(key, newExecutableQuery)
+    miss(key, newExecutableQuery, metaData)
   }
 
-  private def hit(queryKey: QUERY_KEY,
+  private def hit(key: Cache_Key,
                   executableQuery: EXECUTABLE_QUERY,
                   metaData: String) = {
-    tracer.queryCacheHit(queryKey, metaData)
+    tracer.queryCacheHit(key, metaData)
     CacheHit(executableQuery)
   }
 
-  private def miss(queryKey: QUERY_KEY,
+  private def miss(key: Cache_Key,
                    newExecutableQuery: EXECUTABLE_QUERY,
                    metaData: String) = {
-    tracer.queryCacheMiss(queryKey, metaData)
+    tracer.queryCacheMiss(key, metaData)
     CacheMiss(newExecutableQuery)
   }
 
@@ -177,8 +155,6 @@ class QueryCache[QUERY_REP <: AnyRef, QUERY_KEY <: Pair[QUERY_REP, MapValue], EX
     val priorSize = inner.estimatedSize()
     inner.invalidateAll()
     inner.cleanUp()
-    parameterMapping.invalidateAll()
-    parameterMapping.cleanUp()
     tracer.queryCacheFlush(priorSize)
     priorSize
   }
@@ -187,3 +163,28 @@ class QueryCache[QUERY_REP <: AnyRef, QUERY_KEY <: Pair[QUERY_REP, MapValue], EX
 object QueryCache {
   val NOT_PRESENT: ExecutableQuery = null
 }
+
+/*
+   The whole point of this class is to have a map that in addition to key equality, also checks its values for type equality
+ */
+class ParameterTypeMap extends mutable.HashMap[String, Class[_]] {
+  override def equals(that: Any): Boolean = {
+    if (!that.isInstanceOf[ParameterTypeMap])
+      return false
+
+    val other = that.asInstanceOf[ParameterTypeMap]
+
+    if ( this.size != other.size )
+      return false
+
+    for ( key <- this.keys )
+    {
+      val oneType = this (key)
+      val otherType = other(key)
+      if (!(oneType == otherType)) {
+        return false
+      }
+    }
+    true
+  }
+}
diff --git a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/AstLogicalPlanCache.scala b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/AstLogicalPlanCache.scala
index f22c2ef039a1f..43d91abed0776 100644
--- a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/AstLogicalPlanCache.scala
+++ b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/AstLogicalPlanCache.scala
@@ -38,7 +38,7 @@ import org.neo4j.values.virtual.MapValue
   * @tparam STATEMENT Type of AST statement used as key
   */
 class AstLogicalPlanCache[STATEMENT <: AnyRef](override val maximumSize: Int,
-                                               override val tracer: CacheTracer[Pair[STATEMENT, MapValue]],
+                                               override val tracer: CacheTracer[Pair[STATEMENT, ParameterTypeMap]],
                                                clock: Clock,
                                                divergence: StatsDivergenceCalculator,
                                                lastCommittedTxIdProvider: () => Long
diff --git a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/BasePlanner.scala b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/BasePlanner.scala
index d5f64df45603a..3f1023099bf2f 100644
--- a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/BasePlanner.scala
+++ b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/BasePlanner.scala
@@ -29,7 +29,6 @@ import org.neo4j.cypher.internal.planner.v3_5.spi.PlanContext
 import org.neo4j.helpers.collection.Pair
 import org.neo4j.kernel.monitoring.{Monitors => KernelMonitors}
 import org.neo4j.logging.Log
-import org.neo4j.values.virtual.MapValue
 import org.opencypher.v9_0.frontend.phases._
 
 /**
@@ -49,7 +48,7 @@ abstract class BasePlanner[STATEMENT <: AnyRef, PARSED_STATE <: AnyRef](
   protected val logger: InfoLogger = new StringInfoLogger(log)
   protected val monitors: Monitors = WrappedMonitorsv3_5(kernelMonitors)
 
-  protected val cacheTracer: CacheTracer[Pair[STATEMENT, MapValue]] = monitors.newMonitor[CacheTracer[Pair[STATEMENT, MapValue]]]("cypher3.5")
+  protected val cacheTracer: CacheTracer[Pair[STATEMENT, ParameterTypeMap]] = monitors.newMonitor[CacheTracer[Pair[STATEMENT, ParameterTypeMap]]]("cypher3.5")
 
   override def parserCacheSize: Int = config.queryCacheSize
 
diff --git a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/v3_5/Cypher35Planner.scala b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/v3_5/Cypher35Planner.scala
index e497ab219796b..8ec372ee369a9 100644
--- a/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/v3_5/Cypher35Planner.scala
+++ b/community/cypher/cypher/src/main/scala/org/neo4j/cypher/internal/compatibility/v3_5/Cypher35Planner.scala
@@ -153,7 +153,7 @@ case class Cypher35Planner(config: CypherPlannerConfiguration,
         CacheableLogicalPlan(logicalPlanState, reusabilityState)
       }
 
-      val params= ValueConversion.asValues(preparedQuery.extractedParams())
+      val params = ValueConversion.asValues(preparedQuery.extractedParams())
       val cacheableLogicalPlan =
         if (preParsedQuery.debugOptions.isEmpty)
           planCache.computeIfAbsentOrStale(Pair.of(syntacticQuery.statement(), params),
diff --git a/community/cypher/cypher/src/test/scala/org/neo4j/cypher/internal/ParameterTypeMapTest.scala b/community/cypher/cypher/src/test/scala/org/neo4j/cypher/internal/ParameterTypeMapTest.scala
new file mode 100644
index 0000000000000..d8a3b9821892e
--- /dev/null
+++ b/community/cypher/cypher/src/test/scala/org/neo4j/cypher/internal/ParameterTypeMapTest.scala
@@ -0,0 +1,88 @@
+/*
+ * Copyright (c) 2002-2018 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.cypher.internal
+
+import org.neo4j.values.storable.Values._
+import org.opencypher.v9_0.util.test_helpers.CypherFunSuite
+
+class ParameterTypeMapTest extends CypherFunSuite {
+  test("maps should be equal on value types") {
+    val v1 = stringValue("value").getClass
+    val v2 = intValue(1).getClass
+    val v3 = longValue(1L).getClass
+    val v4 = booleanArray(Array[Boolean](true)).getClass
+    val v5 = stringArray("hi").getClass
+
+    val map = new ParameterTypeMap
+    map.put("prop1", v1)
+    map.put("prop2", v2)
+    map.put("prop3", v3)
+    map.put("prop4", v4)
+    map.put("prop5", v5)
+
+    map should equal(map)
+
+    val other = new ParameterTypeMap()
+    other.put("prop1", v1)
+    other.put("prop2", v2)
+    other.put("prop3", v3)
+    other.put("prop4", v4)
+    other.put("prop5", v5)
+
+    map should equal(other)
+    other should equal(map)
+  }
+
+  test ("maps should not be equal on different value types") {
+    val map1 = new ParameterTypeMap()
+    map1.put("prop1", stringValue("value").getClass)
+    map1.put("prop2", intValue(1).getClass)
+    map1.put("prop3", longValue(1L).getClass)
+    map1.put("prop4", booleanArray(Array[Boolean](true)).getClass)
+    map1.put("prop5", stringArray("hi").getClass)
+
+    val map2 = new ParameterTypeMap()
+
+    map1 should not equal map2
+    map2 should not equal map1
+
+    // Compare with almost the same maps
+    // -> not all keys
+    val map3 = new ParameterTypeMap()
+    map3.put("prop1", stringValue("value").getClass)
+    map3.put("prop2", intValue(1).getClass)
+    map3.put("prop4", booleanArray(Array[Boolean](true)).getClass)
+    map3.put("prop3", longValue(1L).getClass)
+
+    map1 should not equal map3
+    map3 should not equal map1
+
+    // -> some different value types
+    val map4 = new ParameterTypeMap()
+    map4.put("prop1", stringValue("value").getClass)
+    map4.put("prop2", longValue(1L).getClass)
+    map4.put("prop4", booleanArray(Array[Boolean](true)).getClass)
+    map4.put("prop3", intValue(1).getClass)
+    map4.put("prop5", stringArray("hi").getClass)
+
+    map1 should not equal map4
+    map4 should not equal map1
+  }
+}
diff --git a/community/values/src/main/java/org/neo4j/values/storable/Values.java b/community/values/src/main/java/org/neo4j/values/storable/Values.java
index aaa7b011a012b..27d4498f2150a 100644
--- a/community/values/src/main/java/org/neo4j/values/storable/Values.java
+++ b/community/values/src/main/java/org/neo4j/values/storable/Values.java
@@ -37,8 +37,6 @@
 
 import org.neo4j.graphdb.spatial.CRS;
 import org.neo4j.graphdb.spatial.Point;
-import org.neo4j.values.AnyValue;
-import org.neo4j.values.virtual.MapValue;
 
 import static java.lang.String.format;
 import static org.neo4j.values.storable.DateTimeValue.datetime;
@@ -750,23 +748,4 @@ public static Value maxValue( ValueGroup valueGroup, Value value )
                 format( "The maxValue for valueGroup %s is not defined yet", valueGroup ) );
         }
     }
-
-    public static boolean mapsEqualsOnValueTypes( MapValue one, MapValue other )
-    {
-        if ( other.size() != one.size() )
-        {
-            return false;
-        }
-
-        for ( String key : one.keySet() )
-        {
-            AnyValue oneValue = one.get( key );
-            AnyValue otherValue = other.get( key );
-            if ( !(oneValue.getClass() == otherValue.getClass()) )
-            {
-                return false;
-            }
-        }
-        return true;
-    }
 }
diff --git a/community/values/src/test/java/org/neo4j/values/storable/ValuesTest.java b/community/values/src/test/java/org/neo4j/values/storable/ValuesTest.java
index 0d1570678646f..9e3240021a4fd 100644
--- a/community/values/src/test/java/org/neo4j/values/storable/ValuesTest.java
+++ b/community/values/src/test/java/org/neo4j/values/storable/ValuesTest.java
@@ -21,12 +21,7 @@
 
 import org.junit.jupiter.api.Test;
 
-import org.neo4j.values.virtual.MapValue;
-import org.neo4j.values.virtual.MapValueBuilder;
-
-import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.neo4j.values.storable.Values.booleanArray;
 import static org.neo4j.values.storable.Values.booleanValue;
 import static org.neo4j.values.storable.Values.byteArray;
@@ -53,8 +48,8 @@ class ValuesTest
     void shouldBeEqualToItself()
     {
         assertEqual( booleanValue( false ), booleanValue( false ) );
-        assertEqual( byteValue( (byte)0 ), byteValue( (byte)0 ) );
-        assertEqual( shortValue( (short)0 ), shortValue( (short)0 ) );
+        assertEqual( byteValue( (byte) 0 ), byteValue( (byte) 0 ) );
+        assertEqual( shortValue( (short) 0 ), shortValue( (short) 0 ) );
         assertEqual( intValue( 0 ), intValue( 0 ) );
         assertEqual( longValue( 0 ), longValue( 0 ) );
         assertEqual( floatValue( 0.0f ), floatValue( 0.0f ) );
@@ -62,8 +57,8 @@ void shouldBeEqualToItself()
         assertEqual( stringValue( "" ), stringValue( "" ) );
 
         assertEqual( booleanValue( true ), booleanValue( true ) );
-        assertEqual( byteValue( (byte)1 ), byteValue( (byte)1 ) );
-        assertEqual( shortValue( (short)1 ), shortValue( (short)1 ) );
+        assertEqual( byteValue( (byte) 1 ), byteValue( (byte) 1 ) );
+        assertEqual( shortValue( (short) 1 ), shortValue( (short) 1 ) );
         assertEqual( intValue( 1 ), intValue( 1 ) );
         assertEqual( longValue( 1 ), longValue( 1 ) );
         assertEqual( floatValue( 1.0f ), floatValue( 1.0f ) );
@@ -95,67 +90,9 @@ void shouldBeEqualToItself()
     @Test
     void pointValueShouldRequireConsistentInput()
     {
-        assertThrows(IllegalArgumentException.class, () -> Values.pointValue( CoordinateReferenceSystem.Cartesian, 1, 2, 3 ) );
-        assertThrows(IllegalArgumentException.class, () -> Values.pointValue( CoordinateReferenceSystem.Cartesian_3D, 1, 2 ) );
-        assertThrows(IllegalArgumentException.class, () -> Values.pointValue( CoordinateReferenceSystem.WGS84, 1, 2, 3 ) );
-        assertThrows(IllegalArgumentException.class, () -> Values.pointValue( CoordinateReferenceSystem.WGS84_3D, 1, 2 ) );
-    }
-
-    @Test
-    void mapsShouldBeEqualOnValueTypes()
-    {
-        MapValueBuilder builder = new MapValueBuilder();
-        builder.add( "prop1", stringValue( "value" ) );
-        builder.add( "prop2", intValue( 1 ) );
-        builder.add( "prop3", longValue( 1L ) );
-        builder.add( "prop4", booleanArray( new boolean[]{true} ) );
-        builder.add( "prop5", stringArray( "hi" ) );
-
-        MapValue map = builder.build();
-
-        assertTrue( Values.mapsEqualsOnValueTypes( map, map ) );
-    }
-
-    @Test
-    void mapsShouldNotBeEqualOnValueTypes()
-    {
-        MapValueBuilder builder = new MapValueBuilder();
-        builder.add( "prop1", stringValue( "value" ) );
-        builder.add( "prop2", intValue( 1 ) );
-        builder.add( "prop3", longValue( 1L ) );
-        builder.add( "prop4", booleanArray( new boolean[]{true} ) );
-        builder.add( "prop5", stringArray( "hi" ) );
-        MapValue map1 = builder.build();
-
-        builder = new MapValueBuilder();
-        MapValue map2 = builder.build();
-
-        assertFalse( Values.mapsEqualsOnValueTypes( map1, map2 ) );
-        assertFalse( Values.mapsEqualsOnValueTypes( map2, map1 ) );
-
-        // Compare with almost the same maps
-        // -> not all keys
-        builder = new MapValueBuilder();
-        builder.add( "prop1", stringValue( "value" ) );
-        builder.add( "prop2", intValue( 1 ) );
-        builder.add( "prop4", booleanArray( new boolean[]{true} ) );
-        builder.add( "prop3", longValue( 1L ) );
-        MapValue map3 = builder.build();
-
-        assertFalse( Values.mapsEqualsOnValueTypes( map1, map3 ) );
-        assertFalse( Values.mapsEqualsOnValueTypes( map3, map1 ) );
-
-        // -> some different value types
-        builder = new MapValueBuilder();
-        builder.add( "prop1", stringValue( "value" ) );
-        builder.add( "prop2", longValue( 1L ) );
-        builder.add( "prop4", booleanArray( new boolean[]{true} ) );
-        builder.add( "prop3", intValue( 1 ) );
-        builder.add( "prop5", stringArray( "hi" ) );
-        MapValue map4 = builder.build();
-
-        assertFalse( Values.mapsEqualsOnValueTypes( map1, map4 ) );
-        assertFalse( Values.mapsEqualsOnValueTypes( map4, map1 ) );
-
+        assertThrows( IllegalArgumentException.class, () -> Values.pointValue( CoordinateReferenceSystem.Cartesian, 1, 2, 3 ) );
+        assertThrows( IllegalArgumentException.class, () -> Values.pointValue( CoordinateReferenceSystem.Cartesian_3D, 1, 2 ) );
+        assertThrows( IllegalArgumentException.class, () -> Values.pointValue( CoordinateReferenceSystem.WGS84, 1, 2, 3 ) );
+        assertThrows( IllegalArgumentException.class, () -> Values.pointValue( CoordinateReferenceSystem.WGS84_3D, 1, 2 ) );
     }
 }
diff --git a/enterprise/cypher/acceptance-spec-suite/src/test/scala/org/neo4j/internal/cypher/acceptance/CypherCompilerStringCacheMonitoringAcceptanceTest.scala b/enterprise/cypher/acceptance-spec-suite/src/test/scala/org/neo4j/internal/cypher/acceptance/CypherCompilerStringCacheMonitoringAcceptanceTest.scala
index bd44f99ae56f9..6782a1752d3de 100644
--- a/enterprise/cypher/acceptance-spec-suite/src/test/scala/org/neo4j/internal/cypher/acceptance/CypherCompilerStringCacheMonitoringAcceptanceTest.scala
+++ b/enterprise/cypher/acceptance-spec-suite/src/test/scala/org/neo4j/internal/cypher/acceptance/CypherCompilerStringCacheMonitoringAcceptanceTest.scala
@@ -25,12 +25,12 @@ package org.neo4j.internal.cypher.acceptance
 import org.hamcrest.Matchers
 import org.neo4j.cypher.ExecutionEngineFunSuite
 import org.neo4j.cypher.ExecutionEngineHelper.createEngine
-import org.neo4j.cypher.internal.{ExecutionEngine, StringCacheMonitor}
+import org.neo4j.cypher.internal.{ExecutionEngine, ParameterTypeMap, StringCacheMonitor}
 import org.neo4j.graphdb.config.Setting
 import org.neo4j.graphdb.factory.GraphDatabaseSettings
 import org.neo4j.helpers.collection.Pair
 import org.neo4j.logging.AssertableLogProvider
-import org.neo4j.values.virtual.{MapValue, VirtualValues}
+import org.neo4j.values.virtual.VirtualValues
 
 import scala.collection.Map
 
@@ -41,11 +41,11 @@ class CypherCompilerStringCacheMonitoringAcceptanceTest extends ExecutionEngineF
   }
 
   class CacheCounter(var counts: CacheCounts = CacheCounts()) extends StringCacheMonitor {
-    override def cacheMiss(key: Pair[String, MapValue]) {
+    override def cacheMiss(key: Pair[String, ParameterTypeMap]) {
       counts = counts.copy(misses = counts.misses + 1)
     }
 
-    override def cacheHit(key: Pair[String, MapValue]) {
+    override def cacheHit(key: Pair[String, ParameterTypeMap]) {
       counts = counts.copy(hits = counts.hits + 1)
     }
 
@@ -53,7 +53,7 @@ class CypherCompilerStringCacheMonitoringAcceptanceTest extends ExecutionEngineF
       counts = counts.copy(flushes = counts.flushes + 1)
     }
 
-    override def cacheDiscard(key: Pair[String, MapValue], key2: String, secondsSinceReplan: Int) {
+    override def cacheDiscard(key: Pair[String, ParameterTypeMap], key2: String, secondsSinceReplan: Int) {
       counts = counts.copy(evicted = counts.evicted + 1)
     }
   }