Repository: vitessio/vitess
Pull request #5406 (2019-11-05 19:44:56)

java: Bump SNAPSHOT version to 4.1-SNAPSHOT after Vitess release 4.0
Signed-off-by: Harshit Gangal <harshit.gangal@gmail.com>

deepthi: @morgo should this be 5.0-SNAPSHOT instead?
morgo: > @morgo should this be 5.0-SNAPSHOT instead?

Yes I agree. It reflects 5.0's development, so we should bump it.

Patch:
diff --git a/.github/bootstrap.sh b/.github/bootstrap.sh
deleted file mode 100755
index 603b4310d6..0000000000
--- a/.github/bootstrap.sh
+++ /dev/null
@@ -1,173 +0,0 @@
-#!/bin/bash
-# shellcheck disable=SC2164
-
-# Copyright 2019 The Vitess Authors.
-#
-# 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.
-
-# This is a next-gen bootstrap which skips Python and Java tests,
-# and does not use the VTROOT/VTTOP layout.
-#
-# My original intention was to use the same bootstrap.sh and gate
-# for new features, but it has turned out to be difficult to do,
-# due to the way that Docker cache works in the CI environment.
-
-function fail() {
-  echo "ERROR: $1"
-  exit 1
-}
-
-[[ "$(dirname "$0")" = "." ]] || fail "bootstrap.sh must be run from its current directory"
-
-# Create main directories.
-
-VTROOT="$PWD"
-
-mkdir -p dist
-mkdir -p bin
-mkdir -p lib
-mkdir -p vthook
-
-source ./dev.env
-
-go version &>/dev/null  || fail "Go is not installed or is not on \$PATH"
-goversion_min 1.12 || fail "Go is not version 1.12+"
-
-# Set up required soft links.
-# TODO(mberlin): Which of these can be deleted?
-ln -snf "$VTROOT/py" "$VTROOT/py-vtdb"
-ln -snf "$VTROOT/go/vt/zkctl/zksrv.sh" "$VTROOT/bin/zksrv.sh"
-ln -snf "$VTROOT/test/vthook-test.sh" "$VTROOT/vthook/test.sh"
-ln -snf "$VTROOT/test/vthook-test_backup_error" "$VTROOT/vthook/test_backup_error"
-ln -snf "$VTROOT/test/vthook-test_backup_transform" "$VTROOT/vthook/test_backup_transform"
-
-# git hooks are only required if someone intends to contribute.
-
-echo "creating git hooks"
-mkdir -p "$VTROOT/.git/hooks"
-ln -sf "$VTROOT/misc/git/pre-commit" "$VTROOT/.git/hooks/pre-commit"
-ln -sf "$VTROOT/misc/git/commit-msg" "$VTROOT/.git/hooks/commit-msg"
-git config core.hooksPath "$VTROOT/.git/hooks"
-
-# install_dep is a helper function to generalize the download and installation of dependencies.
-#
-# If the installation is successful, it puts the installed version string into
-# the $dist/.installed_version file. If the version has not changed, bootstrap
-# will skip future installations.
-function install_dep() {
-  if [[ $# != 4 ]]; then
-    fail "install_dep function requires exactly 4 parameters (and not $#). Parameters: $*"
-  fi
-  local name="$1"
-  local version="$2"
-  local dist="$3"
-  local install_func="$4"
-
-  version_file="$dist/.installed_version"
-  if [[ -f "$version_file" && "$(cat "$version_file")" == "$version" ]]; then
-    echo "skipping $name install. remove $dist to force re-install."
-    return
-  fi
-
-  echo "installing $name $version"
-
-  # shellcheck disable=SC2064
-  trap "fail '$name build failed'; exit 1" ERR
-
-  # Cleanup any existing data and re-create the directory.
-  rm -rf "$dist"
-  mkdir -p "$dist"
-
-  # Change $CWD to $dist before calling "install_func".
-  pushd "$dist" >/dev/null
-  # -E (same as "set -o errtrace") makes sure that "install_func" inherits the
-  # trap. If here's an error, the trap will be called which will exit this
-  # script.
-  set -E
-  $install_func "$version" "$dist"
-  set +E
-  popd >/dev/null
-
-  trap - ERR
-
-  echo "$version" > "$version_file"
-}
-
-
-#
-# 1. Installation of dependencies.
-#
-
-# Wrapper around the `arch` command which plays nice with OS X
-function get_arch() {
-  case $(uname) in
-    Linux) arch;;
-    Darwin) uname -m;;
-  esac
-}
-
-# Install protoc.
-function install_protoc() {
-  local version="$1"
-  local dist="$2"
-
-  case $(uname) in
-    Linux)  local platform=linux;;
-    Darwin) local platform=osx;;
-  esac
-
-  case $(get_arch) in
-      aarch64)  local target=aarch_64;;
-      x86_64)  local target=x86_64;;
-      *)   echo "ERROR: unsupported architecture"; exit 1;;
-  esac
-
-  wget https://github.com/protocolbuffers/protobuf/releases/download/v$version/protoc-$version-$platform-${target}.zip
-  unzip "protoc-$version-$platform-${target}.zip"
-  ln -snf "$dist/bin/protoc" "$VTROOT/bin/protoc"
-}
-protoc_ver=3.6.1
-install_dep "protoc" "$protoc_ver" "$VTROOT/dist/vt-protoc-$protoc_ver" install_protoc
-
-# Download and install etcd, link etcd binary into our root.
-function install_etcd() {
-  local version="$1"
-  local dist="$2"
-
-  case $(uname) in
-    Linux)  local platform=linux; local ext=tar.gz;;
-    Darwin) local platform=darwin; local ext=zip;;
-  esac
-
-  case $(get_arch) in
-      aarch64)  local target=arm64;;
-      x86_64)  local target=amd64;;
-      *)   echo "ERROR: unsupported architecture"; exit 1;;
-  esac
-
-  download_url=https://github.com/coreos/etcd/releases/download
-  file="etcd-${version}-${platform}-${target}.${ext}"
-
-  wget "$download_url/$version/$file"
-  if [ "$ext" = "tar.gz" ]; then
-    tar xzf "$file"
-  else
-    unzip "$file"
-  fi
-  rm "$file"
-  ln -snf "$dist/etcd-${version}-${platform}-${target}/etcd" "$VTROOT/bin/etcd"
-}
-install_dep "etcd" "v3.3.10" "$VTROOT/dist/etcd" install_etcd
-
-echo
-echo "bootstrap finished"
diff --git a/.github/workflows/local-example.yml b/.github/workflows/local-example.yml
deleted file mode 100644
index a4ec46301a..0000000000
--- a/.github/workflows/local-example.yml
+++ /dev/null
@@ -1,42 +0,0 @@
-name: Go
-on: [push]
-jobs:
-
-  build:
-    name: Build
-    runs-on: ubuntu-latest
-    steps:
-
-    - name: Set up Go
-      uses: actions/setup-go@v1
-      with:
-        go-version: 1.13
-      id: local-example
-
-    - name: Check out code
-      uses: actions/checkout@v1
-
-    - name: Get dependencies
-      run: |
-        sudo apt-get install -y mysql-server mysql-client make unzip g++ etcd curl git wget
-        sudo service mysql stop
-        sudo service etcd stop
-        sudo ln -s /etc/apparmor.d/usr.sbin.mysqld /etc/apparmor.d/disable/
-        sudo apparmor_parser -R /etc/apparmor.d/usr.sbin.mysqld
-        go mod download
-
-    - name: Run bootstrap.sh
-      run: |
-        echo "Copying new bootstrap over location of legacy one."
-        cp .github/bootstrap.sh .
-        ./bootstrap.sh
-
-    - name: Build
-      run: |
-        GOBIN=$PWD/bin make build
-
-    - name: Run Local Example
-      run: |
-        export PATH=$PWD/bin:$PATH
-        VTDATAROOT=/tmp/vtdataroot VTTOP=$PWD VTROOT=$PWD test/local_example.sh
-
diff --git a/bootstrap.sh b/bootstrap.sh
index 20fef11545..3296b06dee 100755
--- a/bootstrap.sh
+++ b/bootstrap.sh
@@ -35,8 +35,10 @@ function fail() {
 
 [[ "$(dirname "$0")" = "." ]] || fail "bootstrap.sh must be run from its current directory"
 
+go version &>/dev/null  || fail "Go is not installed or is not on \$PATH"
+[[ "$(go version 2>&1)" =~ go1\.[1-9][1-9] ]] || fail "Go is not version 1.11+"
+
 # Create main directories.
-VTROOT="${VTROOT:-${PWD/\/src\/vitess.io\/vitess/}}"
 mkdir -p "$VTROOT/dist"
 mkdir -p "$VTROOT/bin"
 mkdir -p "$VTROOT/lib"
@@ -51,9 +53,6 @@ else
     source ./build.env
 fi
 
-go version &>/dev/null  || fail "Go is not installed or is not on \$PATH"
-goversion_min 1.12 || fail "Go is not version 1.12+"
-
 if [ "$BUILD_TESTS" == 1 ] ; then
     # Set up required soft links.
     # TODO(mberlin): Which of these can be deleted?
diff --git a/dev.env b/dev.env
index 1c0d1ebdd5..e185c40c01 100644
--- a/dev.env
+++ b/dev.env
@@ -55,6 +55,17 @@ export PIP
 command -v virtualenv2 >/dev/null && VIRTUALENV=virtualenv2 || VIRTUALENV=virtualenv
 export VIRTUALENV
 
+# Add the current GOBIN
+if [ "$GOBIN" ]; then
+  PATH=$(prepend_path "$PATH" "$GOBIN")
+fi
+
+# Many tests rely on "go install" and assume GOBIN is really $VTROOT/bin.
+# Make sure these take precedence.
+GOBIN=$VTROOT/bin
+export GOBIN
+PATH=$(prepend_path "$PATH" "$GOBIN")
+
 # Add chromedriver to path for Selenium tests.
 PATH=$(prepend_path "$PATH" "$VTROOT/dist/chromedriver")
 
@@ -62,6 +73,16 @@ PATH=$(prepend_path "$PATH" "$VTROOT/dist/chromedriver")
 PATH=$(prepend_path "$PATH" "$VTROOT/dist/node/bin")
 export PATH
 
+# GOROOT sanity
+go_bin=$(which go)
+go_env=$(go env | grep GOROOT | cut -f 2 -d\")
+if [[ "$go_bin" != "" && "$go_bin" != "$go_env/bin/go" ]]; then
+  echo "WARNING: \$GOROOT may not be compatible with the used go binary"
+  echo "Please make sure 'go' comes from \$GOROOT/bin"
+  echo "go_env: $go_env"
+  echo "go_bin: $go_bin"
+fi
+
 # mysql install location. Please set based on your environment.
 # Build will not work if this is incorrect.
 
diff --git a/examples/local/etcd-down.sh b/examples/local/etcd-down.sh
index 9d05f649c4..402c121679 100755
--- a/examples/local/etcd-down.sh
+++ b/examples/local/etcd-down.sh
@@ -24,6 +24,6 @@ script_root=$(dirname "${BASH_SOURCE[0]}")
 # shellcheck disable=SC1091
 source "${script_root}/env.sh"
 
-pid=`cat $VTDATAROOT/tmp/etcd.pid`
-echo "Stopping etcd..."
-kill -9 $pid
+# Stop etcd servers.
+echo "Stopping etcd servers..."
+kill -9 "$(pgrep -f "${ETCD_BINDIR}/etcd")"
diff --git a/examples/local/etcd-up.sh b/examples/local/etcd-up.sh
index 05d25ab0e2..5883d96727 100755
--- a/examples/local/etcd-up.sh
+++ b/examples/local/etcd-up.sh
@@ -26,8 +26,6 @@ script_root=$(dirname "${BASH_SOURCE[0]}")
 source "${script_root}/env.sh"
 
 ${ETCD_BINDIR}/etcd --data-dir "${VTDATAROOT}/etcd/"  --listen-client-urls "http://${ETCD_SERVER}" --advertise-client-urls "http://${ETCD_SERVER}" > "${VTDATAROOT}"/tmp/etcd.out 2>&1 &
-PID=$!
-echo $PID > "${VTDATAROOT}/tmp/etcd.pid"
 sleep 5
 
 echo "add /vitess/global"
diff --git a/examples/local/vtctld-down.sh b/examples/local/vtctld-down.sh
index 398311ca04..38ba62c4fc 100755
--- a/examples/local/vtctld-down.sh
+++ b/examples/local/vtctld-down.sh
@@ -23,4 +23,6 @@ source $script_root/env.sh
 
 pid=`cat $VTDATAROOT/tmp/vtctld.pid`
 echo "Stopping vtctld..."
-kill -9 $pid
+kill $pid
+
+kill -9 "$(pgrep -f "/bin/vtctld")"
diff --git a/go.mod b/go.mod
index ae3b4d9aee..e3081b3479 100644
--- a/go.mod
+++ b/go.mod
@@ -24,6 +24,7 @@ require (
 	github.com/golang/mock v1.3.1
 	github.com/golang/protobuf v1.3.2
 	github.com/golang/snappy v0.0.0-20170215233205-553a64147049
+	github.com/google/btree v1.0.0 // indirect
 	github.com/google/shlex v0.0.0-20181106134648-c34317bd91bf // indirect
 	github.com/gorilla/websocket v0.0.0-20160912153041-2d1e4548da23
 	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0
@@ -48,6 +49,8 @@ require (
 	github.com/minio/minio-go v0.0.0-20190131015406-c8a261de75c1
 	github.com/mitchellh/go-testing-interface v1.0.0 // indirect
 	github.com/mitchellh/mapstructure v1.1.2 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.1 // indirect
 	github.com/olekukonko/tablewriter v0.0.0-20160115111002-cca8bbc07984
 	github.com/opentracing-contrib/go-grpc v0.0.0-20180928155321-4b5a12d3ff02
 	github.com/opentracing/opentracing-go v1.1.0
diff --git a/go/vt/vtgate/engine/insert.go b/go/vt/vtgate/engine/insert.go
index bf1de16def..eab98acba8 100644
--- a/go/vt/vtgate/engine/insert.go
+++ b/go/vt/vtgate/engine/insert.go
@@ -390,49 +390,34 @@ func (ins *Insert) getInsertShardedRoute(vcursor VCursor, bindVars map[string]*q
 	// The output from the following 'process' functions is a list of
 	// keyspace ids. For regular inserts, a failure to find a route
 	// results in an error. For 'ignore' type inserts, the keyspace
-	// id is returned as nil, which is used later to drop the corresponding rows.
-	colVindex := ins.Table.ColumnVindexes[0]
-	keyspaceIDs, err := ins.processPrimary(vcursor, vindexRowsValues[0], colVindex)
+	// id is returned as nil, which is used later to drop such rows.
+	keyspaceIDs, err := ins.processPrimary(vcursor, vindexRowsValues[0], ins.Table.ColumnVindexes[0], bindVars)
 	if err != nil {
 		return nil, nil, vterrors.Wrap(err, "getInsertShardedRoute")
 	}
-	// Primary vindex can be owned. If so, go through the processOwned flow.
-	// If not owned, we don't do processUnowned because there's no need to verify
-	// the keyspace ids we just generated.
-	if colVindex.Owned {
-		if err := ins.processOwned(vcursor, vindexRowsValues[0], colVindex, keyspaceIDs); err != nil {
-			return nil, nil, vterrors.Wrap(err, "getInsertShardedRoute")
-		}
-	}
 
-	for vIdx := 1; vIdx < len(ins.Table.ColumnVindexes); vIdx++ {
+	for vIdx := 1; vIdx < len(vindexRowsValues); vIdx++ {
 		colVindex := ins.Table.ColumnVindexes[vIdx]
 		var err error
 		if colVindex.Owned {
-			err = ins.processOwned(vcursor, vindexRowsValues[vIdx], colVindex, keyspaceIDs)
+			switch ins.Opcode {
+			case InsertSharded:
+				err = ins.processOwned(vcursor, vindexRowsValues[vIdx], colVindex, bindVars, keyspaceIDs)
+			case InsertShardedIgnore:
+				// For InsertShardedIgnore, the work is substantially different.
+				// So, we use a separate function.
+				err = ins.processOwnedIgnore(vcursor, vindexRowsValues[vIdx], colVindex, bindVars, keyspaceIDs)
+			default:
+				err = vterrors.Errorf(vtrpcpb.Code_INTERNAL, "BUG: unexpected opcode: %v", ins.Opcode)
+			}
 		} else {
-			err = ins.processUnowned(vcursor, vindexRowsValues[vIdx], colVindex, keyspaceIDs)
+			err = ins.processUnowned(vcursor, vindexRowsValues[vIdx], colVindex, bindVars, keyspaceIDs)
 		}
 		if err != nil {
 			return nil, nil, vterrors.Wrap(err, "getInsertShardedRoute")
 		}
 	}
 
-	// Build 3-d bindvars. Skip rows with nil keyspace ids in case
-	// we're executing an insert ignore.
-	for vIdx, colVindex := range ins.Table.ColumnVindexes {
-		for rowNum, rowColumnKeys := range vindexRowsValues[vIdx] {
-			if keyspaceIDs[rowNum] == nil {
-				// InsertShardedIgnore: skip the row.
-				continue
-			}
-			for colIdx, vindexKey := range rowColumnKeys {
-				col := colVindex.Columns[colIdx]
-				bindVars[insertVarName(col, rowNum)] = sqltypes.ValueBindVariable(vindexKey)
-			}
-		}
-	}
-
 	// We need to know the keyspace ids and the Mids associated with
 	// each RSS.  So we pass the ksid indexes in as ids, and get them back
 	// as values. We also skip nil KeyspaceIds, no need to resolve them.
@@ -480,8 +465,14 @@ func (ins *Insert) getInsertShardedRoute(vcursor VCursor, bindVars map[string]*q
 }
 
 // processPrimary maps the primary vindex values to the keyspace ids.
-func (ins *Insert) processPrimary(vcursor VCursor, vindexColumnsKeys [][]sqltypes.Value, colVindex *vindexes.ColumnVindex) ([][]byte, error) {
-	destinations, err := vindexes.Map(colVindex.Vindex, vcursor, vindexColumnsKeys)
+func (ins *Insert) processPrimary(vcursor VCursor, vindexKeys [][]sqltypes.Value, colVindex *vindexes.ColumnVindex, bv map[string]*querypb.BindVariable) ([][]byte, error) {
+	var flattenedVindexKeys []sqltypes.Value
+	// TODO: @rafael - this will change once vindex Primary keys also support multicolumns
+	for _, val := range vindexKeys {
+		flattenedVindexKeys = append(flattenedVindexKeys, val...)
+	}
+
+	destinations, err := colVindex.Vindex.Map(vcursor, flattenedVindexKeys)
 	if err != nil {
 		return nil, err
 	}
@@ -495,34 +486,56 @@ func (ins *Insert) processPrimary(vcursor VCursor, vindexColumnsKeys [][]sqltype
 		case key.DestinationNone:
 			// No valid keyspace id, we may return an error.
 			if ins.Opcode != InsertShardedIgnore {
-				return nil, fmt.Errorf("could not map %v to a keyspace id", vindexColumnsKeys[i])
+				return nil, fmt.Errorf("could not map %v to a keyspace id", flattenedVindexKeys[i])
 			}
 		default:
-			return nil, fmt.Errorf("could not map %v to a unique keyspace id: %v", vindexColumnsKeys[i], destination)
+			return nil, fmt.Errorf("could not map %v to a unique keyspace id: %v", flattenedVindexKeys[i], destination)
 		}
 	}
 
+	for rowNum, vindexKey := range flattenedVindexKeys {
+		if keyspaceIDs[rowNum] == nil {
+			// InsertShardedIgnore: skip the row.
+			continue
+		}
+		for _, col := range colVindex.Columns {
+			bv[insertVarName(col, rowNum)] = sqltypes.ValueBindVariable(vindexKey)
+		}
+	}
 	return keyspaceIDs, nil
 }
 
-// processOwned creates vindex entries for the values of an owned column.
-func (ins *Insert) processOwned(vcursor VCursor, vindexColumnsKeys [][]sqltypes.Value, colVindex *vindexes.ColumnVindex, ksids [][]byte) error {
-	if ins.Opcode == InsertSharded {
-		return colVindex.Vindex.(vindexes.Lookup).Create(vcursor, vindexColumnsKeys, ksids, false /* ignoreMode */)
+// processOwned creates vindex entries for the values of an owned column for InsertSharded.
+func (ins *Insert) processOwned(vcursor VCursor, vindexColumnsKeys [][]sqltypes.Value, colVindex *vindexes.ColumnVindex, bv map[string]*querypb.BindVariable, ksids [][]byte) error {
+	for rowNum, rowColumnKeys := range vindexColumnsKeys {
+		for colIdx, vindexKey := range rowColumnKeys {
+			col := colVindex.Columns[colIdx]
+			bv[insertVarName(col, rowNum)] = sqltypes.ValueBindVariable(vindexKey)
+		}
 	}
+	return colVindex.Vindex.(vindexes.Lookup).Create(vcursor, vindexColumnsKeys, ksids, false /* ignoreMode */)
+}
 
-	// InsertShardedIgnore
+// processOwnedIgnore creates vindex entries for the values of an owned column for InsertShardedIgnore.
+func (ins *Insert) processOwnedIgnore(vcursor VCursor, vindexColumnsKeys [][]sqltypes.Value, colVindex *vindexes.ColumnVindex, bv map[string]*querypb.BindVariable, ksids [][]byte) error {
 	var createIndexes []int
 	var createKeys [][]sqltypes.Value
 	var createKsids [][]byte
 
 	for rowNum, rowColumnKeys := range vindexColumnsKeys {
+		var rowKeys []sqltypes.Value
 		if ksids[rowNum] == nil {
 			continue
 		}
 		createIndexes = append(createIndexes, rowNum)
-		createKeys = append(createKeys, rowColumnKeys)
 		createKsids = append(createKsids, ksids[rowNum])
+
+		for colIdx, vindexKey := range rowColumnKeys {
+			rowKeys = append(rowKeys, vindexKey)
+			col := colVindex.Columns[colIdx]
+			bv[insertVarName(col, rowNum)] = sqltypes.ValueBindVariable(vindexKey)
+		}
+		createKeys = append(createKeys, rowKeys)
 	}
 	if createKeys == nil {
 		return nil
@@ -534,7 +547,12 @@ func (ins *Insert) processOwned(vcursor VCursor, vindexColumnsKeys [][]sqltypes.
 	}
 	// After creation, verify that the keys map to the keyspace ids. If not, remove
 	// those that don't map.
-	verified, err := vindexes.Verify(colVindex.Vindex, vcursor, createKeys, createKsids)
+	// If values were supplied, we validate against keyspace id.
+	var ids []sqltypes.Value
+	for _, vindexValues := range createKeys {
+		ids = append(ids, vindexValues[0])
+	}
+	verified, err := colVindex.Vindex.Verify(vcursor, ids, createKsids)
 	if err != nil {
 		return err
 	}
@@ -547,26 +565,26 @@ func (ins *Insert) processOwned(vcursor VCursor, vindexColumnsKeys [][]sqltypes.
 }
 
 // processUnowned either reverse maps or validates the values for an unowned column.
-func (ins *Insert) processUnowned(vcursor VCursor, vindexColumnsKeys [][]sqltypes.Value, colVindex *vindexes.ColumnVindex, ksids [][]byte) error {
+func (ins *Insert) processUnowned(vcursor VCursor, vindexColumnsKeys [][]sqltypes.Value, colVindex *vindexes.ColumnVindex, bv map[string]*querypb.BindVariable, ksids [][]byte) error {
 	var reverseIndexes []int
 	var reverseKsids [][]byte
 	var verifyIndexes []int
-	var verifyKeys [][]sqltypes.Value
+	var verifyKeys []sqltypes.Value
 	var verifyKsids [][]byte
 
 	for rowNum, rowColumnKeys := range vindexColumnsKeys {
 		// Right now, we only validate against the first column of a colvindex.
+		// TODO(sougou): address this when we add multicolumn Map support.
+		vindexKey := rowColumnKeys[0]
 		if ksids[rowNum] == nil {
 			continue
 		}
-		// Perform reverse map only for non-multi-column vindexes.
-		_, isMulti := colVindex.Vindex.(vindexes.MultiColumn)
-		if rowColumnKeys[0].IsNull() && !isMulti {
+		if vindexKey.IsNull() {
 			reverseIndexes = append(reverseIndexes, rowNum)
 			reverseKsids = append(reverseKsids, ksids[rowNum])
 		} else {
 			verifyIndexes = append(verifyIndexes, rowNum)
-			verifyKeys = append(verifyKeys, rowColumnKeys)
+			verifyKeys = append(verifyKeys, vindexKey)
 			verifyKsids = append(verifyKsids, ksids[rowNum])
 		}
 	}
@@ -583,26 +601,38 @@ func (ins *Insert) processUnowned(vcursor VCursor, vindexColumnsKeys [][]sqltype
 			return err
 		}
 		for i, reverseKey := range reverseKeys {
-			// Fill the first column with the reverse-mapped value.
-			vindexColumnsKeys[reverseIndexes[i]][0] = reverseKey
+			rowNum := reverseIndexes[i]
+			for colIdx, col := range colVindex.Columns {
+				if colIdx == 0 {
+					// Fill the first column with the reverse-mapped value.
+					bv[insertVarName(col, rowNum)] = sqltypes.ValueBindVariable(reverseKey)
+				} else {
+					// Fill other columns with supplied values.
+					bv[insertVarName(col, rowNum)] = sqltypes.ValueBindVariable(vindexColumnsKeys[rowNum][colIdx])
+				}
+			}
 		}
 	}
 
 	if verifyKsids != nil {
 		// If values were supplied, we validate against keyspace id.
-		verified, err := vindexes.Verify(colVindex.Vindex, vcursor, verifyKeys, verifyKsids)
+		verified, err := colVindex.Vindex.Verify(vcursor, verifyKeys, verifyKsids)
 		if err != nil {
 			return err
 		}
 		for i, v := range verified {
+			rowNum := verifyIndexes[i]
 			if !v {
 				if ins.Opcode != InsertShardedIgnore {
 					return fmt.Errorf("values %v for column %v does not map to keyspace ids", vindexColumnsKeys, colVindex.Columns)
 				}
 				// InsertShardedIgnore: skip the row.
-				ksids[verifyIndexes[i]] = nil
+				ksids[rowNum] = nil
 				continue
 			}
+			for colIdx, col := range colVindex.Columns {
+				bv[insertVarName(col, rowNum)] = sqltypes.ValueBindVariable(vindexColumnsKeys[rowNum][colIdx])
+			}
 		}
 	}
 	return nil
diff --git a/go/vt/vtgate/vindexes/binary.go b/go/vt/vtgate/vindexes/binary.go
index 1d9e2bf132..e26e9fdaa8 100644
--- a/go/vt/vtgate/vindexes/binary.go
+++ b/go/vt/vtgate/vindexes/binary.go
@@ -54,6 +54,11 @@ func (vind *Binary) IsUnique() bool {
 	return true
 }
 
+// IsFunctional returns true since the Vindex is functional.
+func (vind *Binary) IsFunctional() bool {
+	return true
+}
+
 // Verify returns true if ids maps to ksids.
 func (vind *Binary) Verify(_ VCursor, ids []sqltypes.Value, ksids [][]byte) ([]bool, error) {
 	out := make([]bool, len(ids))
diff --git a/go/vt/vtgate/vindexes/binarymd5.go b/go/vt/vtgate/vindexes/binarymd5.go
index a75326e276..288195d912 100644
--- a/go/vt/vtgate/vindexes/binarymd5.go
+++ b/go/vt/vtgate/vindexes/binarymd5.go
@@ -53,6 +53,11 @@ func (vind *BinaryMD5) IsUnique() bool {
 	return true
 }
 
+// IsFunctional returns true since the Vindex is functional.
+func (vind *BinaryMD5) IsFunctional() bool {
+	return true
+}
+
 // Verify returns true if ids maps to ksids.
 func (vind *BinaryMD5) Verify(_ VCursor, ids []sqltypes.Value, ksids [][]byte) ([]bool, error) {
 	out := make([]bool, len(ids))
diff --git a/go/vt/vtgate/vindexes/consistent_lookup.go b/go/vt/vtgate/vindexes/consistent_lookup.go
index 546b70c244..0bb3291981 100644
--- a/go/vt/vtgate/vindexes/consistent_lookup.go
+++ b/go/vt/vtgate/vindexes/consistent_lookup.go
@@ -199,6 +199,11 @@ func (lu *clCommon) String() string {
 	return lu.name
 }
 
+// IsFunctional returns false since the Vindex is not functional.
+func (lu *clCommon) IsFunctional() bool {
+	return false
+}
+
 // Verify returns true if ids maps to ksids.
 func (lu *clCommon) Verify(vcursor VCursor, ids []sqltypes.Value, ksids [][]byte) ([]bool, error) {
 	return lu.lkp.VerifyCustom(vcursor, ids, ksidsToValues(ksids), vtgate.CommitOrder_PRE)
diff --git a/go/vt/vtgate/vindexes/hash.go b/go/vt/vtgate/vindexes/hash.go
index 43859e4f38..b7e625d4c5 100644
--- a/go/vt/vtgate/vindexes/hash.go
+++ b/go/vt/vtgate/vindexes/hash.go
@@ -62,6 +62,11 @@ func (vind *Hash) IsUnique() bool {
 	return true
 }
 
+// IsFunctional returns true since the Vindex is functional.
+func (vind *Hash) IsFunctional() bool {
+	return true
+}
+
 // Map can map ids to key.Destination objects.
 func (vind *Hash) Map(cursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) {
 	out := make([]key.Destination, len(ids))
diff --git a/go/vt/vtgate/vindexes/lookup.go b/go/vt/vtgate/vindexes/lookup.go
index a6869066e4..daadc2eae5 100644
--- a/go/vt/vtgate/vindexes/lookup.go
+++ b/go/vt/vtgate/vindexes/lookup.go
@@ -61,6 +61,11 @@ func (ln *LookupNonUnique) IsUnique() bool {
 	return false
 }
 
+// IsFunctional returns false since the Vindex is not functional.
+func (ln *LookupNonUnique) IsFunctional() bool {
+	return false
+}
+
 // Map can map ids to key.Destination objects.
 func (ln *LookupNonUnique) Map(vcursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) {
 	out := make([]key.Destination, 0, len(ids))
@@ -211,6 +216,11 @@ func (lu *LookupUnique) IsUnique() bool {
 	return true
 }
 
+// IsFunctional returns false since the Vindex is not functional.
+func (lu *LookupUnique) IsFunctional() bool {
+	return false
+}
+
 // Map can map ids to key.Destination objects.
 func (lu *LookupUnique) Map(vcursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) {
 	out := make([]key.Destination, 0, len(ids))
diff --git a/go/vt/vtgate/vindexes/lookup_hash.go b/go/vt/vtgate/vindexes/lookup_hash.go
index 30118bd274..3772b889b0 100644
--- a/go/vt/vtgate/vindexes/lookup_hash.go
+++ b/go/vt/vtgate/vindexes/lookup_hash.go
@@ -93,6 +93,11 @@ func (lh *LookupHash) IsUnique() bool {
 	return false
 }
 
+// IsFunctional returns false since the Vindex is not functional.
+func (lh *LookupHash) IsFunctional() bool {
+	return false
+}
+
 // Map can map ids to key.Destination objects.
 func (lh *LookupHash) Map(vcursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) {
 	out := make([]key.Destination, 0, len(ids))
@@ -244,6 +249,11 @@ func (lhu *LookupHashUnique) IsUnique() bool {
 	return true
 }
 
+// IsFunctional returns false since the Vindex is not functional.
+func (lhu *LookupHashUnique) IsFunctional() bool {
+	return false
+}
+
 // Map can map ids to key.Destination objects.
 func (lhu *LookupHashUnique) Map(vcursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) {
 	out := make([]key.Destination, 0, len(ids))
diff --git a/go/vt/vtgate/vindexes/lookup_internal.go b/go/vt/vtgate/vindexes/lookup_internal.go
index bfcc38a2c1..211ffc933c 100644
--- a/go/vt/vtgate/vindexes/lookup_internal.go
+++ b/go/vt/vtgate/vindexes/lookup_internal.go
@@ -62,9 +62,6 @@ func (lkp *lookupInternal) Init(lookupQueryParams map[string]string, autocommit,
 
 // Lookup performs a lookup for the ids.
 func (lkp *lookupInternal) Lookup(vcursor VCursor, ids []sqltypes.Value) ([]*sqltypes.Result, error) {
-	if vcursor == nil {
-		return nil, fmt.Errorf("cannot perform lookup: no vcursor provided")
-	}
 	results := make([]*sqltypes.Result, 0, len(ids))
 	for _, id := range ids {
 		bindVars := map[string]*querypb.BindVariable{
diff --git a/go/vt/vtgate/vindexes/lookup_unicodeloosemd5_hash.go b/go/vt/vtgate/vindexes/lookup_unicodeloosemd5_hash.go
index 30b12a33f3..b55fcb88fc 100644
--- a/go/vt/vtgate/vindexes/lookup_unicodeloosemd5_hash.go
+++ b/go/vt/vtgate/vindexes/lookup_unicodeloosemd5_hash.go
@@ -94,6 +94,11 @@ func (lh *LookupUnicodeLooseMD5Hash) IsUnique() bool {
 	return false
 }
 
+// IsFunctional returns false since the Vindex is not functional.
+func (lh *LookupUnicodeLooseMD5Hash) IsFunctional() bool {
+	return false
+}
+
 // Map can map ids to key.Destination objects.
 func (lh *LookupUnicodeLooseMD5Hash) Map(vcursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) {
 	out := make([]key.Destination, 0, len(ids))
@@ -256,6 +261,11 @@ func (lhu *LookupUnicodeLooseMD5HashUnique) IsUnique() bool {
 	return true
 }
 
+// IsFunctional returns false since the Vindex is not functional.
+func (lhu *LookupUnicodeLooseMD5HashUnique) IsFunctional() bool {
+	return false
+}
+
 // Map can map ids to key.Destination objects.
 func (lhu *LookupUnicodeLooseMD5HashUnique) Map(vcursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) {
 	out := make([]key.Destination, 0, len(ids))
diff --git a/go/vt/vtgate/vindexes/null.go b/go/vt/vtgate/vindexes/null.go
index d3fc3fc56b..36ed1c42f5 100644
--- a/go/vt/vtgate/vindexes/null.go
+++ b/go/vt/vtgate/vindexes/null.go
@@ -58,6 +58,11 @@ func (vind *Null) IsUnique() bool {
 	return true
 }
 
+// IsFunctional returns true since the Vindex is functional.
+func (vind *Null) IsFunctional() bool {
+	return true
+}
+
 // Map can map ids to key.Destination objects.
 func (vind *Null) Map(cursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) {
 	out := make([]key.Destination, 0, len(ids))
diff --git a/go/vt/vtgate/vindexes/numeric.go b/go/vt/vtgate/vindexes/numeric.go
index 1b0b5c2a07..34f0648a28 100644
--- a/go/vt/vtgate/vindexes/numeric.go
+++ b/go/vt/vtgate/vindexes/numeric.go
@@ -57,6 +57,11 @@ func (*Numeric) IsUnique() bool {
 	return true
 }
 
+// IsFunctional returns true since the Vindex is functional.
+func (*Numeric) IsFunctional() bool {
+	return true
+}
+
 // Verify returns true if ids and ksids match.
 func (*Numeric) Verify(_ VCursor, ids []sqltypes.Value, ksids [][]byte) ([]bool, error) {
 	out := make([]bool, len(ids))
diff --git a/go/vt/vtgate/vindexes/numeric_static_map.go b/go/vt/vtgate/vindexes/numeric_static_map.go
index 9f51a80a88..fa67794792 100644
--- a/go/vt/vtgate/vindexes/numeric_static_map.go
+++ b/go/vt/vtgate/vindexes/numeric_static_map.go
@@ -80,6 +80,11 @@ func (vind *NumericStaticMap) IsUnique() bool {
 	return true
 }
 
+// IsFunctional returns true since the Vindex is functional.
+func (vind *NumericStaticMap) IsFunctional() bool {
+	return true
+}
+
 // Verify returns true if ids and ksids match.
 func (vind *NumericStaticMap) Verify(_ VCursor, ids []sqltypes.Value, ksids [][]byte) ([]bool, error) {
 	out := make([]bool, len(ids))
diff --git a/go/vt/vtgate/vindexes/region_experimental.go b/go/vt/vtgate/vindexes/region_experimental.go
deleted file mode 100644
index 15159517dd..0000000000
--- a/go/vt/vtgate/vindexes/region_experimental.go
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
-Copyright 2019 The Vitess Authors.
-
-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 vindexes
-
-import (
-	"bytes"
-	"encoding/binary"
-	"fmt"
-
-	"vitess.io/vitess/go/sqltypes"
-	"vitess.io/vitess/go/vt/key"
-)
-
-var (
-	_ Vindex        = (*RegionExperimental)(nil)
-	_ Lookup        = (*RegionExperimental)(nil)
-	_ WantOwnerInfo = (*RegionExperimental)(nil)
-	_ MultiColumn   = (*RegionExperimental)(nil)
-)
-
-func init() {
-	Register("region_experimental", NewRegionExperimental)
-}
-
-// RegionExperimental defines a vindex that uses a lookup table.
-// The table is expected to define the id column as unique. It's
-// Unique and a Lookup.
-type RegionExperimental struct {
-	regionBytes int
-	*ConsistentLookupUnique
-}
-
-// NewRegionExperimental creates a RegionExperimental vindex.
-// The supplied map requires all the fields of "consistent_lookup_unique".
-// Additionally, it requires a region_bytes argument whose value can be "1", or "2".
-func NewRegionExperimental(name string, m map[string]string) (Vindex, error) {
-	rbs, ok := m["region_bytes"]
-	if !ok {
-		return nil, fmt.Errorf("region_experimental missing region_bytes param")
-	}
-	var rb int
-	switch rbs {
-	case "1":
-		rb = 1
-	case "2":
-		rb = 2
-	default:
-		return nil, fmt.Errorf("region_bits must be 1 or 2: %v", rbs)
-	}
-	vindex, err := NewConsistentLookupUnique(name, m)
-	if err != nil {
-		// Unreachable.
-		return nil, err
-	}
-	cl := vindex.(*ConsistentLookupUnique)
-	if len(cl.lkp.FromColumns) != 2 {
-		return nil, fmt.Errorf("two columns are required for region_experimental: %v", cl.lkp.FromColumns)
-	}
-	return &RegionExperimental{
-		regionBytes:            rb,
-		ConsistentLookupUnique: cl,
-	}, nil
-}
-
-// MapMulti satisfies MultiColumn.
-func (ge *RegionExperimental) MapMulti(vcursor VCursor, rowsColValues [][]sqltypes.Value) ([]key.Destination, error) {
-	destinations := make([]key.Destination, 0, len(rowsColValues))
-	for _, row := range rowsColValues {
-		if len(row) != 2 {
-			destinations = append(destinations, key.DestinationNone{})
-			continue
-		}
-		// Compute hash.
-		hn, err := sqltypes.ToUint64(row[0])
-		if err != nil {
-			destinations = append(destinations, key.DestinationNone{})
-			continue
-		}
-		h := vhash(hn)
-
-		// Compute region prefix.
-		rn, err := sqltypes.ToUint64(row[1])
-		if err != nil {
-			destinations = append(destinations, key.DestinationNone{})
-			continue
-		}
-		r := make([]byte, 2)
-		binary.BigEndian.PutUint16(r, uint16(rn))
-
-		// Concatenate and add to destinations.
-		if ge.regionBytes == 1 {
-			r = r[1:]
-		}
-		dest := append(r, h...)
-		destinations = append(destinations, key.DestinationKeyspaceID(dest))
-	}
-	return destinations, nil
-}
-
-// VerifyMulti satisfies MultiColumn.
-func (ge *RegionExperimental) VerifyMulti(vcursor VCursor, rowsColValues [][]sqltypes.Value, ksids [][]byte) ([]bool, error) {
-	result := make([]bool, len(rowsColValues))
-	destinations, _ := ge.MapMulti(vcursor, rowsColValues)
-	for i, dest := range destinations {
-		destksid, ok := dest.(key.DestinationKeyspaceID)
-		if !ok {
-			continue
-		}
-		result[i] = bytes.Equal([]byte(destksid), ksids[i])
-	}
-	// We also need to verify from the lookup.
-	// TODO(sougou): we should only verify true values from previous result.
-	lresult, err := Verify(ge.ConsistentLookupUnique, vcursor, rowsColValues, ksids)
-	if err != nil {
-		return nil, err
-	}
-	for i := range result {
-		result[i] = result[i] && lresult[i]
-	}
-	return result, nil
-}
diff --git a/go/vt/vtgate/vindexes/reverse_bits.go b/go/vt/vtgate/vindexes/reverse_bits.go
index 9957988fe5..eab8ee3993 100644
--- a/go/vt/vtgate/vindexes/reverse_bits.go
+++ b/go/vt/vtgate/vindexes/reverse_bits.go
@@ -59,6 +59,11 @@ func (vind *ReverseBits) IsUnique() bool {
 	return true
 }
 
+// IsFunctional returns true since the Vindex is functional.
+func (vind *ReverseBits) IsFunctional() bool {
+	return true
+}
+
 // Map returns the corresponding KeyspaceId values for the given ids.
 func (vind *ReverseBits) Map(cursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) {
 	out := make([]key.Destination, len(ids))
diff --git a/go/vt/vtgate/vindexes/unicodeloosemd5.go b/go/vt/vtgate/vindexes/unicodeloosemd5.go
index 8264ebfe91..ca1668255c 100644
--- a/go/vt/vtgate/vindexes/unicodeloosemd5.go
+++ b/go/vt/vtgate/vindexes/unicodeloosemd5.go
@@ -62,6 +62,11 @@ func (vind *UnicodeLooseMD5) IsUnique() bool {
 	return true
 }
 
+// IsFunctional returns true since the Vindex is functional.
+func (vind *UnicodeLooseMD5) IsFunctional() bool {
+	return true
+}
+
 // Verify returns true if ids maps to ksids.
 func (vind *UnicodeLooseMD5) Verify(_ VCursor, ids []sqltypes.Value, ksids [][]byte) ([]bool, error) {
 	out := make([]bool, len(ids))
diff --git a/go/vt/vtgate/vindexes/vindex.go b/go/vt/vtgate/vindexes/vindex.go
index 35ab8c853e..66493c4c67 100644
--- a/go/vt/vtgate/vindexes/vindex.go
+++ b/go/vt/vtgate/vindexes/vindex.go
@@ -57,6 +57,16 @@ type Vindex interface {
 	// Which means Map() maps to either a KeyRange or a single KeyspaceID.
 	IsUnique() bool
 
+	// IsFunctional returns true if the Vindex can compute
+	// the keyspace id from the id without a lookup.
+	// A Functional vindex is also required to be Unique.
+	// Which means Map() maps to either a KeyRange or a single KeyspaceID.
+	IsFunctional() bool
+
+	// Verify must be implented by all vindexes. It should return
+	// true if the ids can be mapped to the keyspace ids.
+	Verify(cursor VCursor, ids []sqltypes.Value, ksids [][]byte) ([]bool, error)
+
 	// Map can map ids to key.Destination objects.
 	// If the Vindex is unique, each id would map to either
 	// a KeyRange, or a single KeyspaceID.
@@ -64,18 +74,7 @@ type Vindex interface {
 	// a KeyRange, or a list of KeyspaceID.
 	// If the error returned if nil, then the array len of the
 	// key.Destination array must match len(ids).
-	Map(vcursor VCursor, ids []sqltypes.Value) ([]key.Destination, error)
-
-	// Verify must be implented by all vindexes. It should return
-	// true if the ids can be mapped to the keyspace ids.
-	Verify(vcursor VCursor, ids []sqltypes.Value, ksids [][]byte) ([]bool, error)
-}
-
-// MultiColumn defines the interface for vindexes that can
-// support multi-column vindexes.
-type MultiColumn interface {
-	MapMulti(vcursor VCursor, rowsColValues [][]sqltypes.Value) ([]key.Destination, error)
-	VerifyMulti(vcursor VCursor, rowsColValues [][]sqltypes.Value, ksids [][]byte) ([]bool, error)
+	Map(cursor VCursor, ids []sqltypes.Value) ([]key.Destination, error)
 }
 
 // A Reversible vindex is one that can perform a
@@ -83,7 +82,7 @@ type MultiColumn interface {
 // is optional. If present, VTGate can use it to
 // fill column values based on the target keyspace id.
 type Reversible interface {
-	ReverseMap(vcursor VCursor, ks [][]byte) ([]sqltypes.Value, error)
+	ReverseMap(cursor VCursor, ks [][]byte) ([]sqltypes.Value, error)
 }
 
 // A Lookup vindex is one that needs to lookup
@@ -139,27 +138,3 @@ func CreateVindex(vindexType, name string, params map[string]string) (Vindex, er
 	}
 	return f(name, params)
 }
-
-// Map invokes MapMulti or Map depending on which is available.
-func Map(vindex Vindex, vcursor VCursor, rowsColValues [][]sqltypes.Value) ([]key.Destination, error) {
-	if multi, ok := vindex.(MultiColumn); ok {
-		return multi.MapMulti(vcursor, rowsColValues)
-	}
-	return vindex.Map(vcursor, firstColsOnly(rowsColValues))
-}
-
-// Verify invokes VerifyMulti or Verify depending on which is available.
-func Verify(vindex Vindex, vcursor VCursor, rowsColValues [][]sqltypes.Value, ksids [][]byte) ([]bool, error) {
-	if multi, ok := vindex.(MultiColumn); ok {
-		return multi.VerifyMulti(vcursor, rowsColValues, ksids)
-	}
-	return vindex.Verify(vcursor, firstColsOnly(rowsColValues), ksids)
-}
-
-func firstColsOnly(rowsColValues [][]sqltypes.Value) []sqltypes.Value {
-	firstCols := make([]sqltypes.Value, 0, len(rowsColValues))
-	for _, val := range rowsColValues {
-		firstCols = append(firstCols, val[0])
-	}
-	return firstCols
-}
diff --git a/go/vt/vtgate/vindexes/vschema.go b/go/vt/vtgate/vindexes/vschema.go
index 8f519fb709..fb31f4a21a 100644
--- a/go/vt/vtgate/vindexes/vschema.go
+++ b/go/vt/vtgate/vindexes/vschema.go
@@ -313,6 +313,9 @@ func buildTables(ks *vschemapb.Keyspace, vschema *VSchema, ksvschema *KeyspaceSc
 				if !columnVindex.Vindex.IsUnique() {
 					return fmt.Errorf("primary vindex %s is not Unique for table %s", ind.Name, tname)
 				}
+				if owned {
+					return fmt.Errorf("primary vindex %s cannot be owned for table %s", ind.Name, tname)
+				}
 			}
 			t.ColumnVindexes = append(t.ColumnVindexes, columnVindex)
 			if owned {
diff --git a/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go b/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go
index eed3576ff8..434e821f1b 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go
@@ -87,7 +87,7 @@ const (
 // buildExecutionPlan is the function that builds the full plan.
 func buildReplicatorPlan(filter *binlogdatapb.Filter, tableKeys map[string][]string, copyState map[string]*sqltypes.Result) (*ReplicatorPlan, error) {
 	plan := &ReplicatorPlan{
-		VStreamFilter: &binlogdatapb.Filter{FieldEventMode: filter.FieldEventMode},
+		VStreamFilter: &binlogdatapb.Filter{},
 		TargetTables:  make(map[string]*TablePlan),
 		TablePlans:    make(map[string]*TablePlan),
 		tableKeys:     tableKeys,
diff --git a/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go b/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
index 43affd74d3..c620f83d3c 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
@@ -73,7 +73,7 @@ func (plan *Plan) filter(values []sqltypes.Value) (bool, []sqltypes.Value, error
 	result := make([]sqltypes.Value, len(plan.ColExprs))
 	for i, colExpr := range plan.ColExprs {
 		if colExpr.ColNum >= len(values) {
-			return false, nil, fmt.Errorf("index out of range, colExpr.ColNum: %d, len(values): %d", colExpr.ColNum, len(values))
+			return false, nil, fmt.Errorf("index out of range, colExpr.ColNum: %d, len(values):%d", colExpr.ColNum, len(values))
 		}
 		result[i] = values[colExpr.ColNum]
 	}
@@ -360,8 +360,8 @@ func (plan *Plan) analyzeInKeyRange(kschema *vindexes.KeyspaceSchema, exprs sqlp
 		if err != nil {
 			return err
 		}
-		if !plan.Vindex.IsUnique() {
-			return fmt.Errorf("vindex must be Unique to be used for VReplication: %s", vtype)
+		if !plan.Vindex.IsUnique() || !plan.Vindex.IsFunctional() {
+			return fmt.Errorf("vindex must be Unique and Functional to be used for VReplication: %s", vtype)
 		}
 		krExpr = exprs[2]
 	default:
diff --git a/java/client/pom.xml b/java/client/pom.xml
index 48e4f42b74..571b8866cc 100644
--- a/java/client/pom.xml
+++ b/java/client/pom.xml
@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.vitess</groupId>
     <artifactId>vitess-parent</artifactId>
-    <version>3.1.0-SNAPSHOT</version>
+    <version>5.0-SNAPSHOT</version>
   </parent>
   <artifactId>vitess-client</artifactId>
 
diff --git a/java/example/pom.xml b/java/example/pom.xml
index 86a5bf887d..47301f29bb 100644
--- a/java/example/pom.xml
+++ b/java/example/pom.xml
@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.vitess</groupId>
     <artifactId>vitess-parent</artifactId>
-    <version>3.1.0-SNAPSHOT</version>
+    <version>5.0-SNAPSHOT</version>
   </parent>
   <artifactId>vitess-example</artifactId>
 
diff --git a/java/grpc-client/pom.xml b/java/grpc-client/pom.xml
index 0ec7f82410..1bf7ba2ee7 100644
--- a/java/grpc-client/pom.xml
+++ b/java/grpc-client/pom.xml
@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.vitess</groupId>
     <artifactId>vitess-parent</artifactId>
-    <version>3.1.0-SNAPSHOT</version>
+    <version>5.0-SNAPSHOT</version>
   </parent>
   <artifactId>vitess-grpc-client</artifactId>
 
diff --git a/java/hadoop/pom.xml b/java/hadoop/pom.xml
index c26e9cb0e3..37e815dde2 100644
--- a/java/hadoop/pom.xml
+++ b/java/hadoop/pom.xml
@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.vitess</groupId>
     <artifactId>vitess-parent</artifactId>
-    <version>3.1.0-SNAPSHOT</version>
+    <version>5.0-SNAPSHOT</version>
   </parent>
   <artifactId>vitess-hadoop</artifactId>
 
diff --git a/java/jdbc/pom.xml b/java/jdbc/pom.xml
index 02ada256b3..1a47533a5c 100644
--- a/java/jdbc/pom.xml
+++ b/java/jdbc/pom.xml
@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.vitess</groupId>
     <artifactId>vitess-parent</artifactId>
-    <version>3.1.0-SNAPSHOT</version>
+    <version>5.0-SNAPSHOT</version>
   </parent>
   <artifactId>vitess-jdbc</artifactId>
 
diff --git a/java/pom.xml b/java/pom.xml
index d81deb143a..2e35714fa4 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -11,7 +11,7 @@
 
   <groupId>io.vitess</groupId>
   <artifactId>vitess-parent</artifactId>
-  <version>3.1.0-SNAPSHOT</version>
+  <version>5.0-SNAPSHOT</version>
   <packaging>pom</packaging>
 
   <name>Vitess Java Client libraries and Hadoop support [Parent]</name>


Test patch:
diff --git a/go/vt/vtgate/engine/insert_test.go b/go/vt/vtgate/engine/insert_test.go
index d8373863f9..a33900bbf2 100644
--- a/go/vt/vtgate/engine/insert_test.go
+++ b/go/vt/vtgate/engine/insert_test.go
@@ -20,7 +20,6 @@ import (
 	"errors"
 	"testing"
 
-	"github.com/stretchr/testify/assert"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/vtgate/vindexes"
 
@@ -319,7 +318,7 @@ func TestInsertShardedFail(t *testing.T) {
 
 	// The lookup will fail to map to a keyspace id.
 	_, err = ins.Execute(vc, map[string]*querypb.BindVariable{}, false)
-	expectError(t, "Execute", err, "execInsertSharded: getInsertShardedRoute: could not map [INT64(1)] to a keyspace id")
+	expectError(t, "Execute", err, "execInsertSharded: getInsertShardedRoute: could not map INT64(1) to a keyspace id")
 }
 
 func TestInsertShardedGenerate(t *testing.T) {
@@ -645,92 +644,6 @@ func TestInsertShardedOwnedWithNull(t *testing.T) {
 	})
 }
 
-func TestInsertShardedGeo(t *testing.T) {
-	invschema := &vschemapb.SrvVSchema{
-		Keyspaces: map[string]*vschemapb.Keyspace{
-			"sharded": {
-				Sharded: true,
-				Vindexes: map[string]*vschemapb.Vindex{
-					"geo": {
-						Type: "region_experimental",
-						Params: map[string]string{
-							"region_bytes": "1",
-							"table":        "lkp",
-							"from":         "id,region",
-							"to":           "toc",
-						},
-						Owner: "t1",
-					},
-				},
-				Tables: map[string]*vschemapb.Table{
-					"t1": {
-						ColumnVindexes: []*vschemapb.ColumnVindex{{
-							Name:    "geo",
-							Columns: []string{"id", "region"},
-						}},
-					},
-				},
-			},
-		},
-	}
-	vs, err := vindexes.BuildVSchema(invschema)
-	if err != nil {
-		t.Fatal(err)
-	}
-	ks := vs.Keyspaces["sharded"]
-
-	ins := NewInsert(
-		InsertSharded,
-		ks.Keyspace,
-		[]sqltypes.PlanValue{{
-			// colVindex columns: id, region
-			Values: []sqltypes.PlanValue{{
-				// rows for id
-				Values: []sqltypes.PlanValue{{
-					Value: sqltypes.NewInt64(1),
-				}, {
-					Value: sqltypes.NewInt64(1),
-				}},
-			}, {
-				// rows for region
-				Values: []sqltypes.PlanValue{{
-					Value: sqltypes.NewInt64(1),
-				}, {
-					Value: sqltypes.NewInt64(255),
-				}},
-			}},
-		}},
-		ks.Tables["t1"],
-		"prefix",
-		[]string{" mid1", " mid2"},
-		" suffix",
-	)
-
-	vc := &loggingVCursor{
-		shards:       []string{"-20", "20-"},
-		shardForKsid: []string{"20-", "-20"},
-	}
-	_, err = ins.Execute(vc, map[string]*querypb.BindVariable{}, false)
-	if err != nil {
-		t.Fatal(err)
-	}
-	vc.ExpectLog(t, []string{
-		// ExecutePre proves that keyspace ids are generated, and that they are inserted into the lookup.
-		`ExecutePre insert into lkp(id, region, toc) values(:id0, :region0, :toc0), (:id1, :region1, :toc1) ` +
-			`id0: type:INT64 value:"1" id1: type:INT64 value:"1" ` +
-			`region0: type:INT64 value:"1" region1: type:INT64 value:"255" ` +
-			`toc0: type:VARBINARY value:"\001\026k@\264J\272K\326" toc1: type:VARBINARY value:"\377\026k@\264J\272K\326"  true`,
-		`ResolveDestinations sharded [value:"0"  value:"1" ] Destinations:DestinationKeyspaceID(01166b40b44aba4bd6),DestinationKeyspaceID(ff166b40b44aba4bd6)`,
-		`ExecuteMultiShard sharded.20-: prefix mid1 suffix /* vtgate:: keyspace_id:01166b40b44aba4bd6 */ ` +
-			`{_id0: type:INT64 value:"1" _id1: type:INT64 value:"1" ` +
-			`_region0: type:INT64 value:"1" _region1: type:INT64 value:"255" } ` +
-			`sharded.-20: prefix mid2 suffix /* vtgate:: keyspace_id:ff166b40b44aba4bd6 */ ` +
-			`{_id0: type:INT64 value:"1" _id1: type:INT64 value:"1" ` +
-			`_region0: type:INT64 value:"1" _region1: type:INT64 value:"255" } ` +
-			`true false`,
-	})
-}
-
 func TestInsertShardedIgnoreOwned(t *testing.T) {
 	invschema := &vschemapb.SrvVSchema{
 		Keyspaces: map[string]*vschemapb.Keyspace{
@@ -909,117 +822,19 @@ func TestInsertShardedIgnoreOwned(t *testing.T) {
 		// Bind vars for rows 2 & 3 may be missing because they were not sent.
 		`ExecuteMultiShard ` +
 			`sharded.20-: prefix mid1 suffix /* vtgate:: keyspace_id:00 */ ` +
-			`{_c10: type:INT64 value:"5" _c13: type:INT64 value:"8" ` +
-			`_c20: type:INT64 value:"9" _c23: type:INT64 value:"12" ` +
+			`{_c10: type:INT64 value:"5" _c12: type:INT64 value:"7" _c13: type:INT64 value:"8" ` +
+			`_c20: type:INT64 value:"9" _c22: type:INT64 value:"11" _c23: type:INT64 value:"12" ` +
 			`_c30: type:INT64 value:"13" _c33: type:INT64 value:"16" ` +
-			`_id0: type:INT64 value:"1" _id3: type:INT64 value:"4" } ` +
+			`_id0: type:INT64 value:"1" _id2: type:INT64 value:"3" _id3: type:INT64 value:"4" } ` +
 			`sharded.-20: prefix mid4 suffix /* vtgate:: keyspace_id:00 */ ` +
-			`{_c10: type:INT64 value:"5" _c13: type:INT64 value:"8" ` +
-			`_c20: type:INT64 value:"9" _c23: type:INT64 value:"12" ` +
+			`{_c10: type:INT64 value:"5" _c12: type:INT64 value:"7" _c13: type:INT64 value:"8" ` +
+			`_c20: type:INT64 value:"9" _c22: type:INT64 value:"11" _c23: type:INT64 value:"12" ` +
 			`_c30: type:INT64 value:"13" _c33: type:INT64 value:"16" ` +
-			`_id0: type:INT64 value:"1" _id3: type:INT64 value:"4" } ` +
+			`_id0: type:INT64 value:"1" _id2: type:INT64 value:"3" _id3: type:INT64 value:"4" } ` +
 			`true false`,
 	})
 }
 
-func TestInsertIgnoreGeo(t *testing.T) {
-	invschema := &vschemapb.SrvVSchema{
-		Keyspaces: map[string]*vschemapb.Keyspace{
-			"sharded": {
-				Sharded: true,
-				Vindexes: map[string]*vschemapb.Vindex{
-					"geo": {
-						Type: "region_experimental",
-						Params: map[string]string{
-							"region_bytes": "1",
-							"table":        "lkp",
-							"from":         "id,region",
-							"to":           "toc",
-						},
-						Owner: "t1",
-					},
-				},
-				Tables: map[string]*vschemapb.Table{
-					"t1": {
-						ColumnVindexes: []*vschemapb.ColumnVindex{{
-							Name:    "geo",
-							Columns: []string{"id", "region"},
-						}},
-					},
-				},
-			},
-		},
-	}
-	vs, err := vindexes.BuildVSchema(invschema)
-	if err != nil {
-		t.Fatal(err)
-	}
-	ks := vs.Keyspaces["sharded"]
-
-	ins := NewInsert(
-		InsertShardedIgnore,
-		ks.Keyspace,
-		[]sqltypes.PlanValue{{
-			// colVindex columns: id, region
-			Values: []sqltypes.PlanValue{{
-				// rows for id
-				Values: []sqltypes.PlanValue{{
-					Value: sqltypes.NewInt64(1),
-				}, {
-					Value: sqltypes.NewInt64(2),
-				}},
-			}, {
-				// rows for region
-				Values: []sqltypes.PlanValue{{
-					Value: sqltypes.NewInt64(1),
-				}, {
-					Value: sqltypes.NewInt64(2),
-				}},
-			}},
-		}},
-		ks.Tables["t1"],
-		"prefix",
-		[]string{" mid1", " mid2"},
-		" suffix",
-	)
-
-	ksid0 := sqltypes.MakeTestResult(
-		sqltypes.MakeTestFields(
-			"to",
-			"varbinary",
-		),
-		"\x00",
-	)
-	noresult := &sqltypes.Result{}
-	vc := &loggingVCursor{
-		shards:       []string{"-20", "20-"},
-		shardForKsid: []string{"20-", "-20"},
-		results: []*sqltypes.Result{
-			// insert lkp
-			noresult,
-			// fail one verification (row 2)
-			ksid0,
-			noresult,
-		},
-	}
-	_, err = ins.Execute(vc, map[string]*querypb.BindVariable{}, false)
-	if err != nil {
-		t.Fatal(err)
-	}
-	vc.ExpectLog(t, []string{
-		`ExecutePre insert ignore into lkp(id, region, toc) values(:id0, :region0, :toc0), (:id1, :region1, :toc1) ` +
-			`id0: type:INT64 value:"1" id1: type:INT64 value:"2" ` +
-			`region0: type:INT64 value:"1" region1: type:INT64 value:"2" ` +
-			`toc0: type:VARBINARY value:"\001\026k@\264J\272K\326" toc1: type:VARBINARY value:"\002\006\347\352\"\316\222p\217"  true`,
-		// Row 2 will fail verification. This is what we're testing. The second row should not get inserted.
-		`ExecutePre select id from lkp where id = :id and toc = :toc id: type:INT64 value:"1" toc: type:VARBINARY value:"\001\026k@\264J\272K\326"  false`,
-		`ExecutePre select id from lkp where id = :id and toc = :toc id: type:INT64 value:"2" toc: type:VARBINARY value:"\002\006\347\352\"\316\222p\217"  false`,
-		`ResolveDestinations sharded [value:"0" ] Destinations:DestinationKeyspaceID(01166b40b44aba4bd6)`,
-		`ExecuteMultiShard sharded.20-: prefix mid1 suffix /* vtgate:: keyspace_id:01166b40b44aba4bd6 */ ` +
-			`{_id0: type:INT64 value:"1" _region0: type:INT64 value:"1" } true true`,
-	})
-}
-
 func TestInsertShardedIgnoreOwnedWithNull(t *testing.T) {
 	invschema := &vschemapb.SrvVSchema{
 		Keyspaces: map[string]*vschemapb.Keyspace{
@@ -1092,6 +907,7 @@ func TestInsertShardedIgnoreOwnedWithNull(t *testing.T) {
 		),
 		"\x00",
 	)
+	//noresult := &sqltypes.Result{}
 	vc := &loggingVCursor{
 		shards:       []string{"-20", "20-"},
 		shardForKsid: []string{"-20", "20-"},
@@ -1270,91 +1086,6 @@ func TestInsertShardedUnownedVerify(t *testing.T) {
 	})
 }
 
-func TestInsertUnownedGeo(t *testing.T) {
-	invschema := &vschemapb.SrvVSchema{
-		Keyspaces: map[string]*vschemapb.Keyspace{
-			"sharded": {
-				Sharded: true,
-				Vindexes: map[string]*vschemapb.Vindex{
-					"primary": {
-						Type: "hash",
-					},
-					"geo": {
-						Type: "region_experimental",
-						Params: map[string]string{
-							"region_bytes": "1",
-							"table":        "lkp",
-							"from":         "other_id,region",
-							"to":           "toc",
-						},
-					},
-				},
-				Tables: map[string]*vschemapb.Table{
-					"t1": {
-						ColumnVindexes: []*vschemapb.ColumnVindex{{
-							Name:    "primary",
-							Columns: []string{"id"},
-						}, {
-							Name:    "geo",
-							Columns: []string{"other_id", "region"},
-						}},
-					},
-				},
-			},
-		},
-	}
-	vs, err := vindexes.BuildVSchema(invschema)
-	if err != nil {
-		t.Fatal(err)
-	}
-	ks := vs.Keyspaces["sharded"]
-
-	ins := NewInsert(
-		InsertSharded,
-		ks.Keyspace,
-		[]sqltypes.PlanValue{{
-			// colVindex columns: id
-			Values: []sqltypes.PlanValue{{
-				// rows for id
-				Values: []sqltypes.PlanValue{{
-					Value: sqltypes.NewInt64(1),
-				}},
-			}},
-		}, {
-			// colVindex columns: other_id, region
-			Values: []sqltypes.PlanValue{{
-				// rows for other_id
-				Values: []sqltypes.PlanValue{{
-					Value: sqltypes.NewInt64(2),
-				}},
-			}, {
-				// rows for region
-				Values: []sqltypes.PlanValue{{
-					Value: sqltypes.NewInt64(3),
-				}},
-			}},
-		}},
-		ks.Tables["t1"],
-		"prefix",
-		[]string{" mid1"},
-		" suffix",
-	)
-
-	noresult := &sqltypes.Result{}
-	vc := &loggingVCursor{
-		shards: []string{"-20", "20-"},
-		results: []*sqltypes.Result{
-			// fail verification
-			noresult,
-		},
-	}
-	_, err = ins.Execute(vc, map[string]*querypb.BindVariable{}, false)
-	assert.EqualError(t, err, "execInsertSharded: getInsertShardedRoute: values [[INT64(2) INT64(3)]] for column [other_id region] does not map to keyspace ids")
-	vc.ExpectLog(t, []string{
-		`ExecutePre select other_id from lkp where other_id = :other_id and toc = :toc other_id: type:INT64 value:"2" toc: type:VARBINARY value:"\026k@\264J\272K\326"  false`,
-	})
-}
-
 func TestInsertShardedIgnoreUnownedVerify(t *testing.T) {
 	invschema := &vschemapb.SrvVSchema{
 		Keyspaces: map[string]*vschemapb.Keyspace{
@@ -1461,10 +1192,10 @@ func TestInsertShardedIgnoreUnownedVerify(t *testing.T) {
 		`ExecuteMultiShard ` +
 			`sharded.20-: prefix mid1 suffix /* vtgate:: keyspace_id:166b40b44aba4bd6 */ ` +
 			`{_c30: type:INT64 value:"10" _c32: type:INT64 value:"12" ` +
-			`_id0: type:INT64 value:"1" _id2: type:INT64 value:"3" } ` +
+			`_id0: type:INT64 value:"1" _id1: type:INT64 value:"2" _id2: type:INT64 value:"3" } ` +
 			`sharded.-20: prefix mid3 suffix /* vtgate:: keyspace_id:4eb190c9a2fa169c */ ` +
 			`{_c30: type:INT64 value:"10" _c32: type:INT64 value:"12" ` +
-			`_id0: type:INT64 value:"1" _id2: type:INT64 value:"3" } ` +
+			`_id0: type:INT64 value:"1" _id1: type:INT64 value:"2" _id2: type:INT64 value:"3" } ` +
 			`true false`,
 	})
 }
diff --git a/go/vt/vtgate/engine/vindex_func_test.go b/go/vt/vtgate/engine/vindex_func_test.go
index 917fc73f1b..2b9e973327 100644
--- a/go/vt/vtgate/engine/vindex_func_test.go
+++ b/go/vt/vtgate/engine/vindex_func_test.go
@@ -30,9 +30,10 @@ import (
 // uvindex is Unique.
 type uvindex struct{ matchid, matchkr bool }
 
-func (*uvindex) String() string { return "uvindex" }
-func (*uvindex) Cost() int      { return 1 }
-func (*uvindex) IsUnique() bool { return true }
+func (*uvindex) String() string     { return "uvindex" }
+func (*uvindex) Cost() int          { return 1 }
+func (*uvindex) IsUnique() bool     { return true }
+func (*uvindex) IsFunctional() bool { return false }
 func (*uvindex) Verify(vindexes.VCursor, []sqltypes.Value, [][]byte) ([]bool, error) {
 	panic("unimplemented")
 }
@@ -59,9 +60,10 @@ func (v *uvindex) Map(cursor vindexes.VCursor, ids []sqltypes.Value) ([]key.Dest
 // nvindex is NonUnique.
 type nvindex struct{ matchid, matchkr bool }
 
-func (*nvindex) String() string { return "nvindex" }
-func (*nvindex) Cost() int      { return 1 }
-func (*nvindex) IsUnique() bool { return false }
+func (*nvindex) String() string     { return "nvindex" }
+func (*nvindex) Cost() int          { return 1 }
+func (*nvindex) IsUnique() bool     { return false }
+func (*nvindex) IsFunctional() bool { return false }
 func (*nvindex) Verify(vindexes.VCursor, []sqltypes.Value, [][]byte) ([]bool, error) {
 	panic("unimplemented")
 }
diff --git a/go/vt/vtgate/executor_dml_test.go b/go/vt/vtgate/executor_dml_test.go
index f492e8e3c0..551d462566 100644
--- a/go/vt/vtgate/executor_dml_test.go
+++ b/go/vt/vtgate/executor_dml_test.go
@@ -654,7 +654,7 @@ func TestInsertShardedKeyrange(t *testing.T) {
 
 	// If a unique vindex returns a keyrange, we fail the insert
 	_, err := executorExec(executor, "insert into keyrange_table(krcol_unique, krcol) values(1, 1)", nil)
-	want := "execInsertSharded: getInsertShardedRoute: could not map [INT64(1)] to a unique keyspace id: DestinationKeyRange(-10)"
+	want := "execInsertSharded: getInsertShardedRoute: could not map INT64(1) to a unique keyspace id: DestinationKeyRange(-10)"
 	if err == nil || err.Error() != want {
 		t.Errorf("executorExec error: %v, want %s", err, want)
 	}
@@ -786,9 +786,13 @@ func TestInsertShardedIgnore(t *testing.T) {
 		Sql: "insert ignore into insert_ignore_test(pv, owned, verify) values (:_pv0, :_owned0, :_verify0),(:_pv4, :_owned4, :_verify4) /* vtgate:: keyspace_id:166b40b44aba4bd6,166b40b44aba4bd6 */",
 		BindVariables: map[string]*querypb.BindVariable{
 			"_pv0":     sqltypes.Int64BindVariable(1),
+			"_pv2":     sqltypes.Int64BindVariable(3),
+			"_pv3":     sqltypes.Int64BindVariable(4),
 			"_pv4":     sqltypes.Int64BindVariable(5),
 			"_pv5":     sqltypes.Int64BindVariable(6),
 			"_owned0":  sqltypes.Int64BindVariable(1),
+			"_owned2":  sqltypes.Int64BindVariable(3),
+			"_owned3":  sqltypes.Int64BindVariable(4),
 			"_owned4":  sqltypes.Int64BindVariable(5),
 			"_owned5":  sqltypes.Int64BindVariable(6),
 			"_verify0": sqltypes.Int64BindVariable(1),
@@ -803,9 +807,13 @@ func TestInsertShardedIgnore(t *testing.T) {
 		Sql: "insert ignore into insert_ignore_test(pv, owned, verify) values (:_pv5, :_owned5, :_verify5) /* vtgate:: keyspace_id:4eb190c9a2fa169c */",
 		BindVariables: map[string]*querypb.BindVariable{
 			"_pv0":     sqltypes.Int64BindVariable(1),
+			"_pv2":     sqltypes.Int64BindVariable(3),
+			"_pv3":     sqltypes.Int64BindVariable(4),
 			"_pv4":     sqltypes.Int64BindVariable(5),
 			"_pv5":     sqltypes.Int64BindVariable(6),
 			"_owned0":  sqltypes.Int64BindVariable(1),
+			"_owned2":  sqltypes.Int64BindVariable(3),
+			"_owned3":  sqltypes.Int64BindVariable(4),
 			"_owned4":  sqltypes.Int64BindVariable(5),
 			"_owned5":  sqltypes.Int64BindVariable(6),
 			"_verify0": sqltypes.Int64BindVariable(1),
diff --git a/go/vt/vtgate/executor_framework_test.go b/go/vt/vtgate/executor_framework_test.go
index 1478144e5b..1956738248 100644
--- a/go/vt/vtgate/executor_framework_test.go
+++ b/go/vt/vtgate/executor_framework_test.go
@@ -282,9 +282,10 @@ func (dp DestinationAnyShardPickerFirstShard) PickShard(shardCount int) int {
 type keyRangeLookuper struct {
 }
 
-func (v *keyRangeLookuper) String() string { return "keyrange_lookuper" }
-func (*keyRangeLookuper) Cost() int        { return 0 }
-func (*keyRangeLookuper) IsUnique() bool   { return false }
+func (v *keyRangeLookuper) String() string   { return "keyrange_lookuper" }
+func (*keyRangeLookuper) Cost() int          { return 0 }
+func (*keyRangeLookuper) IsUnique() bool     { return false }
+func (*keyRangeLookuper) IsFunctional() bool { return false }
 func (*keyRangeLookuper) Verify(vindexes.VCursor, []sqltypes.Value, [][]byte) ([]bool, error) {
 	return []bool{}, nil
 }
@@ -306,9 +307,10 @@ func newKeyRangeLookuper(name string, params map[string]string) (vindexes.Vindex
 type keyRangeLookuperUnique struct {
 }
 
-func (v *keyRangeLookuperUnique) String() string { return "keyrange_lookuper" }
-func (*keyRangeLookuperUnique) Cost() int        { return 0 }
-func (*keyRangeLookuperUnique) IsUnique() bool   { return true }
+func (v *keyRangeLookuperUnique) String() string   { return "keyrange_lookuper" }
+func (*keyRangeLookuperUnique) Cost() int          { return 0 }
+func (*keyRangeLookuperUnique) IsUnique() bool     { return true }
+func (*keyRangeLookuperUnique) IsFunctional() bool { return false }
 func (*keyRangeLookuperUnique) Verify(vindexes.VCursor, []sqltypes.Value, [][]byte) ([]bool, error) {
 	return []bool{}, nil
 }
diff --git a/go/vt/vtgate/planbuilder/plan_test.go b/go/vt/vtgate/planbuilder/plan_test.go
index 9e0e72ed30..4a445b8d31 100644
--- a/go/vt/vtgate/planbuilder/plan_test.go
+++ b/go/vt/vtgate/planbuilder/plan_test.go
@@ -39,9 +39,10 @@ import (
 // hashIndex is a functional, unique Vindex.
 type hashIndex struct{ name string }
 
-func (v *hashIndex) String() string { return v.name }
-func (*hashIndex) Cost() int        { return 1 }
-func (*hashIndex) IsUnique() bool   { return true }
+func (v *hashIndex) String() string   { return v.name }
+func (*hashIndex) Cost() int          { return 1 }
+func (*hashIndex) IsUnique() bool     { return true }
+func (*hashIndex) IsFunctional() bool { return true }
 func (*hashIndex) Verify(vindexes.VCursor, []sqltypes.Value, [][]byte) ([]bool, error) {
 	return []bool{}, nil
 }
@@ -56,9 +57,10 @@ func newHashIndex(name string, _ map[string]string) (vindexes.Vindex, error) {
 // lookupIndex is a unique Vindex, and satisfies Lookup.
 type lookupIndex struct{ name string }
 
-func (v *lookupIndex) String() string { return v.name }
-func (*lookupIndex) Cost() int        { return 2 }
-func (*lookupIndex) IsUnique() bool   { return true }
+func (v *lookupIndex) String() string   { return v.name }
+func (*lookupIndex) Cost() int          { return 2 }
+func (*lookupIndex) IsUnique() bool     { return true }
+func (*lookupIndex) IsFunctional() bool { return false }
 func (*lookupIndex) Verify(vindexes.VCursor, []sqltypes.Value, [][]byte) ([]bool, error) {
 	return []bool{}, nil
 }
@@ -80,9 +82,10 @@ var _ vindexes.Lookup = (*lookupIndex)(nil)
 // multiIndex satisfies Lookup, NonUnique.
 type multiIndex struct{ name string }
 
-func (v *multiIndex) String() string { return v.name }
-func (*multiIndex) Cost() int        { return 3 }
-func (*multiIndex) IsUnique() bool   { return false }
+func (v *multiIndex) String() string   { return v.name }
+func (*multiIndex) Cost() int          { return 3 }
+func (*multiIndex) IsUnique() bool     { return false }
+func (*multiIndex) IsFunctional() bool { return false }
 func (*multiIndex) Verify(vindexes.VCursor, []sqltypes.Value, [][]byte) ([]bool, error) {
 	return []bool{}, nil
 }
@@ -105,9 +108,10 @@ var _ vindexes.Lookup = (*multiIndex)(nil)
 // costlyIndex satisfies Lookup, NonUnique.
 type costlyIndex struct{ name string }
 
-func (v *costlyIndex) String() string { return v.name }
-func (*costlyIndex) Cost() int        { return 10 }
-func (*costlyIndex) IsUnique() bool   { return false }
+func (v *costlyIndex) String() string   { return v.name }
+func (*costlyIndex) Cost() int          { return 10 }
+func (*costlyIndex) IsUnique() bool     { return false }
+func (*costlyIndex) IsFunctional() bool { return false }
 func (*costlyIndex) Verify(vindexes.VCursor, []sqltypes.Value, [][]byte) ([]bool, error) {
 	return []bool{}, nil
 }
diff --git a/go/vt/vtgate/vindexes/consistent_lookup_test.go b/go/vt/vtgate/vindexes/consistent_lookup_test.go
index e1e571d07e..ee17dd88f8 100644
--- a/go/vt/vtgate/vindexes/consistent_lookup_test.go
+++ b/go/vt/vtgate/vindexes/consistent_lookup_test.go
@@ -55,6 +55,9 @@ func TestConsistentLookupInfo(t *testing.T) {
 	if lookup.IsUnique() {
 		t.Errorf("IsUnique(): %v, want false", lookup.IsUnique())
 	}
+	if lookup.IsFunctional() {
+		t.Errorf("IsFunctional(): %v, want false", lookup.IsFunctional())
+	}
 }
 
 func TestConsistentLookupUniqueInfo(t *testing.T) {
@@ -68,6 +71,9 @@ func TestConsistentLookupUniqueInfo(t *testing.T) {
 	if !lookup.IsUnique() {
 		t.Errorf("IsUnique(): %v, want true", lookup.IsUnique())
 	}
+	if lookup.IsFunctional() {
+		t.Errorf("IsFunctional(): %v, want false", lookup.IsFunctional())
+	}
 }
 
 func TestConsistentLookupMap(t *testing.T) {
@@ -493,7 +499,7 @@ func (vc *loggingVCursor) verifyLog(t *testing.T, want []string) {
 			t.Fatalf("index exceeded: %v", vc.log[i:])
 		}
 		if got != want[i] {
-			t.Errorf("log(%d):\n%q, want\n%q", i, got, want[i])
+			t.Errorf("log(%d):\n%s, want\n%s", i, got, want[i])
 		}
 	}
 	if len(want) > len(vc.log) {
diff --git a/go/vt/vtgate/vindexes/lookup_test.go b/go/vt/vtgate/vindexes/lookup_test.go
index ca26ce61e9..3d89d1d433 100644
--- a/go/vt/vtgate/vindexes/lookup_test.go
+++ b/go/vt/vtgate/vindexes/lookup_test.go
@@ -127,16 +127,6 @@ func TestLookupNonUniqueString(t *testing.T) {
 	}
 }
 
-func TestLookupNilVCursor(t *testing.T) {
-	lookupNonUnique := createLookup(t, "lookup", false)
-
-	_, err := lookupNonUnique.Map(nil, []sqltypes.Value{sqltypes.NewInt64(1), sqltypes.NewInt64(2)})
-	want := "cannot perform lookup: no vcursor provided"
-	if err == nil || err.Error() != want {
-		t.Errorf("Map(nil vcursor) err: %v, want %v", err, want)
-	}
-}
-
 func TestLookupNonUniqueMap(t *testing.T) {
 	lookupNonUnique := createLookup(t, "lookup", false)
 	vc := &vcursor{numRows: 2}
diff --git a/go/vt/vtgate/vindexes/region_experimental_test.go b/go/vt/vtgate/vindexes/region_experimental_test.go
deleted file mode 100644
index ef85c65b90..0000000000
--- a/go/vt/vtgate/vindexes/region_experimental_test.go
+++ /dev/null
@@ -1,141 +0,0 @@
-/*
-Copyright 2019 The Vitess Authors.
-
-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 vindexes
-
-import (
-	"strconv"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"vitess.io/vitess/go/sqltypes"
-	"vitess.io/vitess/go/vt/key"
-)
-
-func TestRegionExperimentalMapMulti1(t *testing.T) {
-	ge, err := createRegionVindex(t, "region_experimental", "f1,f2", 1)
-	assert.NoError(t, err)
-	got, err := ge.(MultiColumn).MapMulti(nil, [][]sqltypes.Value{{
-		sqltypes.NewInt64(1), sqltypes.NewInt64(1),
-	}, {
-		sqltypes.NewInt64(1), sqltypes.NewInt64(255),
-	}, {
-		sqltypes.NewInt64(1), sqltypes.NewInt64(256),
-	}, {
-		// Invalid length.
-		sqltypes.NewInt64(1),
-	}, {
-		// Invalid id.
-		sqltypes.NewVarBinary("abcd"), sqltypes.NewInt64(256),
-	}, {
-		// Invalid region.
-		sqltypes.NewInt64(1), sqltypes.NewVarBinary("abcd"),
-	}})
-	assert.NoError(t, err)
-
-	want := []key.Destination{
-		key.DestinationKeyspaceID([]byte("\x01\x16k@\xb4J\xbaK\xd6")),
-		key.DestinationKeyspaceID([]byte("\xff\x16k@\xb4J\xbaK\xd6")),
-		key.DestinationKeyspaceID([]byte("\x00\x16k@\xb4J\xbaK\xd6")),
-		key.DestinationNone{},
-		key.DestinationNone{},
-		key.DestinationNone{},
-	}
-	assert.Equal(t, want, got)
-}
-
-func TestRegionExperimentalMapMulti2(t *testing.T) {
-	ge, err := createRegionVindex(t, "region_experimental", "f1,f2", 2)
-	assert.NoError(t, err)
-	got, err := ge.(MultiColumn).MapMulti(nil, [][]sqltypes.Value{{
-		sqltypes.NewInt64(1), sqltypes.NewInt64(1),
-	}, {
-		sqltypes.NewInt64(1), sqltypes.NewInt64(255),
-	}, {
-		sqltypes.NewInt64(1), sqltypes.NewInt64(256),
-	}, {
-		sqltypes.NewInt64(1), sqltypes.NewInt64(0x10000),
-	}})
-	assert.NoError(t, err)
-
-	want := []key.Destination{
-		key.DestinationKeyspaceID([]byte("\x00\x01\x16k@\xb4J\xbaK\xd6")),
-		key.DestinationKeyspaceID([]byte("\x00\xff\x16k@\xb4J\xbaK\xd6")),
-		key.DestinationKeyspaceID([]byte("\x01\x00\x16k@\xb4J\xbaK\xd6")),
-		key.DestinationKeyspaceID([]byte("\x00\x00\x16k@\xb4J\xbaK\xd6")),
-	}
-	assert.Equal(t, want, got)
-}
-
-func TestRegionExperimentalVerifyMulti(t *testing.T) {
-
-	ge, err := createRegionVindex(t, "region_experimental", "f1,f2", 1)
-	assert.NoError(t, err)
-	vals := [][]sqltypes.Value{{
-		// One for match
-		sqltypes.NewInt64(1), sqltypes.NewInt64(1),
-	}, {
-		// One for mismatch by lookup
-		sqltypes.NewInt64(1), sqltypes.NewInt64(1),
-	}, {
-		// One for mismatch
-		sqltypes.NewInt64(1), sqltypes.NewInt64(1),
-	}, {
-		// One invalid value
-		sqltypes.NewInt64(1),
-	}}
-	ksids := [][]byte{
-		[]byte("\x01\x16k@\xb4J\xbaK\xd6"),
-		[]byte("\x01\x16k@\xb4J\xbaK\xd6"),
-		[]byte("no match"),
-		[]byte(""),
-	}
-	vc := &loggingVCursor{}
-	vc.AddResult(makeTestResult(1), nil)
-	// The second value should return a mismatch.
-	vc.AddResult(&sqltypes.Result{}, nil)
-	vc.AddResult(makeTestResult(1), nil)
-	vc.AddResult(makeTestResult(1), nil)
-
-	want := []bool{true, false, false, false}
-	got, err := ge.(MultiColumn).VerifyMulti(vc, vals, ksids)
-	assert.NoError(t, err)
-	vc.verifyLog(t, []string{
-		"ExecutePre select f1 from t where f1 = :f1 and toc = :toc [{f1 1} {toc \x01\x16k@\xb4J\xbaK\xd6}] false",
-		"ExecutePre select f1 from t where f1 = :f1 and toc = :toc [{f1 1} {toc \x01\x16k@\xb4J\xbaK\xd6}] false",
-		"ExecutePre select f1 from t where f1 = :f1 and toc = :toc [{f1 1} {toc no match}] false",
-		"ExecutePre select f1 from t where f1 = :f1 and toc = :toc [{f1 1} {toc }] false",
-	})
-	assert.Equal(t, want, got)
-}
-
-func TestRegionExperimentalCreateErrors(t *testing.T) {
-	_, err := createRegionVindex(t, "region_experimental", "f1,f2", 3)
-	assert.EqualError(t, err, "region_bits must be 1 or 2: 3")
-	_, err = CreateVindex("region_experimental", "region_experimental", nil)
-	assert.EqualError(t, err, "region_experimental missing region_bytes param")
-	_, err = createRegionVindex(t, "region_experimental", "f1", 2)
-	assert.EqualError(t, err, "two columns are required for region_experimental: [f1]")
-}
-
-func createRegionVindex(t *testing.T, name, from string, rb int) (Vindex, error) {
-	return CreateVindex(name, name, map[string]string{
-		"region_bytes": strconv.Itoa(rb),
-		"table":        "t",
-		"from":         from,
-		"to":           "toc",
-	})
-}
diff --git a/go/vt/vtgate/vindexes/vindex_test.go b/go/vt/vtgate/vindexes/vindex_test.go
deleted file mode 100644
index 1c85f83a1f..0000000000
--- a/go/vt/vtgate/vindexes/vindex_test.go
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
-Copyright 2019 The Vitess Authors.
-
-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 vindexes
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"vitess.io/vitess/go/sqltypes"
-	"vitess.io/vitess/go/vt/key"
-)
-
-func TestVindexMap(t *testing.T) {
-	ge, err := createRegionVindex(t, "region_experimental", "f1,f2", 1)
-	assert.NoError(t, err)
-
-	got, err := Map(ge, nil, [][]sqltypes.Value{{
-		sqltypes.NewInt64(1), sqltypes.NewInt64(1),
-	}})
-	assert.NoError(t, err)
-
-	want := []key.Destination{
-		key.DestinationKeyspaceID([]byte("\x01\x16k@\xb4J\xbaK\xd6")),
-	}
-	assert.Equal(t, want, got)
-
-	hash, err := CreateVindex("hash", "hash", nil)
-	assert.NoError(t, err)
-	got, err = Map(hash, nil, [][]sqltypes.Value{{
-		sqltypes.NewInt64(1),
-	}})
-	assert.NoError(t, err)
-	want = []key.Destination{
-		key.DestinationKeyspaceID([]byte("\x16k@\xb4J\xbaK\xd6")),
-	}
-	assert.Equal(t, want, got)
-}
-
-func TestVindexVerify(t *testing.T) {
-	vc := &loggingVCursor{}
-	vc.AddResult(makeTestResult(1), nil)
-	ge, err := createRegionVindex(t, "region_experimental", "f1,f2", 1)
-	assert.NoError(t, err)
-
-	got, err := Verify(ge, vc, [][]sqltypes.Value{{
-		sqltypes.NewInt64(1), sqltypes.NewInt64(1),
-	}},
-		[][]byte{
-			[]byte("\x01\x16k@\xb4J\xbaK\xd6"),
-		},
-	)
-	assert.NoError(t, err)
-	vc.verifyLog(t, []string{
-		"ExecutePre select f1 from t where f1 = :f1 and toc = :toc [{f1 1} {toc \x01\x16k@\xb4J\xbaK\xd6}] false",
-	})
-
-	want := []bool{true}
-	assert.Equal(t, want, got)
-
-	hash, err := CreateVindex("hash", "hash", nil)
-	assert.NoError(t, err)
-	got, err = Verify(hash, nil, [][]sqltypes.Value{{
-		sqltypes.NewInt64(1),
-	}},
-		[][]byte{
-			[]byte("\x16k@\xb4J\xbaK\xd6"),
-		},
-	)
-	assert.NoError(t, err)
-	assert.Equal(t, want, got)
-}
diff --git a/go/vt/vtgate/vindexes/vschema_test.go b/go/vt/vtgate/vindexes/vschema_test.go
index 200636c238..e209919f2a 100644
--- a/go/vt/vtgate/vindexes/vschema_test.go
+++ b/go/vt/vtgate/vindexes/vschema_test.go
@@ -45,6 +45,7 @@ type stFU struct {
 func (v *stFU) String() string                                                    { return v.name }
 func (*stFU) Cost() int                                                           { return 1 }
 func (*stFU) IsUnique() bool                                                      { return true }
+func (*stFU) IsFunctional() bool                                                  { return true }
 func (*stFU) Verify(VCursor, []sqltypes.Value, [][]byte) ([]bool, error)          { return []bool{}, nil }
 func (*stFU) Map(cursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) { return nil, nil }
 
@@ -63,6 +64,7 @@ type stLN struct {
 func (v *stLN) String() string                                                    { return v.name }
 func (*stLN) Cost() int                                                           { return 0 }
 func (*stLN) IsUnique() bool                                                      { return false }
+func (*stLN) IsFunctional() bool                                                  { return false }
 func (*stLN) Verify(VCursor, []sqltypes.Value, [][]byte) ([]bool, error)          { return []bool{}, nil }
 func (*stLN) Map(cursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) { return nil, nil }
 func (*stLN) Create(VCursor, [][]sqltypes.Value, [][]byte, bool) error            { return nil }
@@ -85,6 +87,7 @@ type stLU struct {
 func (v *stLU) String() string                                                    { return v.name }
 func (*stLU) Cost() int                                                           { return 2 }
 func (*stLU) IsUnique() bool                                                      { return true }
+func (*stLU) IsFunctional() bool                                                  { return false }
 func (*stLU) Verify(VCursor, []sqltypes.Value, [][]byte) ([]bool, error)          { return []bool{}, nil }
 func (*stLU) Map(cursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) { return nil, nil }
 func (*stLU) Create(VCursor, [][]sqltypes.Value, [][]byte, bool) error            { return nil }
@@ -110,6 +113,7 @@ type stLO struct {
 func (v *stLO) String() string                                                    { return v.name }
 func (*stLO) Cost() int                                                           { return 2 }
 func (*stLO) IsUnique() bool                                                      { return true }
+func (*stLO) IsFunctional() bool                                                  { return false }
 func (*stLO) Verify(VCursor, []sqltypes.Value, [][]byte) ([]bool, error)          { return []bool{}, nil }
 func (*stLO) Map(cursor VCursor, ids []sqltypes.Value) ([]key.Destination, error) { return nil, nil }
 func (*stLO) Create(VCursor, [][]sqltypes.Value, [][]byte, bool) error            { return nil }
@@ -1554,6 +1558,38 @@ func TestBuildVSchemaNotUniqueFail(t *testing.T) {
 	}
 }
 
+func TestBuildVSchemaPrimaryNonFunctionalFail(t *testing.T) {
+	bad := vschemapb.SrvVSchema{
+		Keyspaces: map[string]*vschemapb.Keyspace{
+			"sharded": {
+				Sharded: true,
+				Vindexes: map[string]*vschemapb.Vindex{
+					"stlu": {
+						Type:  "stlu",
+						Owner: "t1",
+					},
+				},
+				Tables: map[string]*vschemapb.Table{
+					"t1": {
+						ColumnVindexes: []*vschemapb.ColumnVindex{
+							{
+								Column: "c1",
+								Name:   "stlu",
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	got, _ := BuildVSchema(&bad)
+	err := got.Keyspaces["sharded"].Error
+	want := "primary vindex stlu cannot be owned for table t1"
+	if err == nil || err.Error() != want {
+		t.Errorf("BuildVSchema: %v, want %v", err, want)
+	}
+}
+
 func TestSequence(t *testing.T) {
 	good := vschemapb.SrvVSchema{
 		Keyspaces: map[string]*vschemapb.Keyspace{
diff --git a/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go b/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
index 731de846b9..5d46fb2eed 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
@@ -362,7 +362,7 @@ func TestPlanbuilder(t *testing.T) {
 	}, {
 		inTable: t1,
 		inRule:  &binlogdatapb.Rule{Match: "t1", Filter: "select id, val from t1 where in_keyrange(id, 'lookup', '-80')"},
-		outErr:  `vindex must be Unique to be used for VReplication: lookup`,
+		outErr:  `vindex must be Unique and Functional to be used for VReplication: lookup`,
 	}, {
 		inTable: t1,
 		inRule:  &binlogdatapb.Rule{Match: "t1", Filter: "select id, val from t1 where in_keyrange(id, 'hash', '80')"},
diff --git a/test/vtgatev3_test.py b/test/vtgatev3_test.py
index 83039e40ef..6d6e02743d 100755
--- a/test/vtgatev3_test.py
+++ b/test/vtgatev3_test.py
@@ -1822,7 +1822,7 @@ class TestVTGateFunctions(unittest.TestCase):
     try:
       vtgate_conn.begin()
       with self.assertRaisesRegexp(dbexceptions.DatabaseError,
-                                   '.*could not map.*NULL.*to a keyspace id.*'):
+                                   '.*could not map NULL to a keyspace id.*'):
         self.execute_on_master(
             vtgate_conn,
             'insert into vt_user_extra (email) values (:email)',