diff --git a/.github/workflows/upgrade_downgrade_test_backups_e2e.yml b/.github/workflows/upgrade_downgrade_test_backups_e2e.yml
index a56aad2f523..0c558b00684 100644
--- a/.github/workflows/upgrade_downgrade_test_backups_e2e.yml
+++ b/.github/workflows/upgrade_downgrade_test_backups_e2e.yml
@@ -72,7 +72,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
       with:
-        go-version: 1.22.7
+        go-version: 1.23.4
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_backups_manual.yml b/.github/workflows/upgrade_downgrade_test_backups_manual.yml
index 00aab1b78ff..680b0da87e0 100644
--- a/.github/workflows/upgrade_downgrade_test_backups_manual.yml
+++ b/.github/workflows/upgrade_downgrade_test_backups_manual.yml
@@ -76,7 +76,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
       with:
-        go-version: 1.22.7
+        go-version: 1.23.4
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_onlineddl_flow.yml b/.github/workflows/upgrade_downgrade_test_onlineddl_flow.yml
index 72426e70a61..dd389663a35 100644
--- a/.github/workflows/upgrade_downgrade_test_onlineddl_flow.yml
+++ b/.github/workflows/upgrade_downgrade_test_onlineddl_flow.yml
@@ -83,7 +83,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
       with:
-        go-version: 1.22.7
+        go-version: 1.23.4
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_query_serving_queries.yml b/.github/workflows/upgrade_downgrade_test_query_serving_queries.yml
index 5ac1a55334c..e4ccc793933 100644
--- a/.github/workflows/upgrade_downgrade_test_query_serving_queries.yml
+++ b/.github/workflows/upgrade_downgrade_test_query_serving_queries.yml
@@ -75,7 +75,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
       with:
-        go-version: 1.22.7
+        go-version: 1.23.4
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_query_serving_queries_2.yml b/.github/workflows/upgrade_downgrade_test_query_serving_queries_2.yml
new file mode 100644
index 00000000000..87a78d0b659
--- /dev/null
+++ b/.github/workflows/upgrade_downgrade_test_query_serving_queries_2.yml
@@ -0,0 +1,212 @@
+name: Query Serving (Queries - 2) - Upgrade Downgrade Testing
+on:
+  push:
+  pull_request:
+
+concurrency:
+  group: format('{0}-{1}', ${{ github.ref }}, 'Upgrade Downgrade Testing Query Serving (Queries - 2)')
+  cancel-in-progress: true
+
+permissions: read-all
+
+# This test ensures that our end-to-end tests work using Vitess components
+# (vtgate, vttablet, etc) built on different versions.
+
+jobs:
+
+  upgrade_downgrade_test:
+    name: Run Upgrade Downgrade Test - Query Serving (Queries - 2)
+    runs-on: gh-hosted-runners-16cores-1-24.04
+
+    steps:
+    - name: Skip CI
+      run: |
+        if [[ "${{contains( github.event.pull_request.labels.*.name, 'Skip CI')}}" == "true" ]]; then
+          echo "skipping CI due to the 'Skip CI' label"
+          exit 1
+        fi
+
+    - name: Check if workflow needs to be skipped
+      id: skip-workflow
+      run: |
+        skip='false'
+        if [[ "${{github.event.pull_request}}" ==  "" ]] && [[ "${{github.ref}}" != "refs/heads/main" ]] && [[ ! "${{github.ref}}" =~ ^refs/heads/release-[0-9]+\.[0-9]$ ]] && [[ ! "${{github.ref}}" =~ "refs/tags/.*" ]]; then
+          skip='true'
+        fi
+        echo Skip ${skip}
+        echo "skip-workflow=${skip}" >> $GITHUB_OUTPUT
+
+    - name: Check out commit's code
+      if: steps.skip-workflow.outputs.skip-workflow == 'false'
+      uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332 # v4.1.7
+      with:
+        fetch-depth: 0
+
+    - name: Set output with latest release branch
+      id: output-previous-release-ref
+      if: steps.skip-workflow.outputs.skip-workflow == 'false'
+      run: |
+        previous_release_ref=$(./tools/get_previous_release.sh ${{github.base_ref}} ${{github.ref}})
+        echo $previous_release_ref
+        echo "previous_release_ref=${previous_release_ref}" >> $GITHUB_OUTPUT
+
+    - name: Check for changes in relevant files
+      if: steps.skip-workflow.outputs.skip-workflow == 'false'
+      uses: dorny/paths-filter@ebc4d7e9ebcb0b1eb21480bb8f43113e996ac77a # v3.0.1
+      id: changes
+      with:
+        token: ''
+        filters: |
+          end_to_end:
+            - 'go/**'
+            - 'go/**/*.go'
+            - 'test.go'
+            - 'Makefile'
+            - 'build.env'
+            - 'go.sum'
+            - 'go.mod'
+            - 'proto/*.proto'
+            - 'tools/**'
+            - 'config/**'
+            - 'bootstrap.sh'
+            - '.github/workflows/upgrade_downgrade_test_query_serving_queries.yml'
+
+    - name: Set up Go
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
+      with:
+        go-version: 1.23.4
+
+    - name: Set up python
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      uses: actions/setup-python@39cd14951b08e74b54015e9e001cdefcf80e669f # v5.1.1
+
+    - name: Tune the OS
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        sudo sysctl -w net.ipv4.ip_local_port_range="22768 65535"
+
+    - name: Get base dependencies
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        sudo DEBIAN_FRONTEND="noninteractive" apt-get update
+        # Uninstall any previously installed MySQL first
+        sudo systemctl stop apparmor
+        sudo DEBIAN_FRONTEND="noninteractive" apt-get remove -y --purge mysql-server mysql-client mysql-common
+        sudo apt-get -y autoremove
+        sudo apt-get -y autoclean
+        sudo deluser mysql
+        sudo rm -rf /var/lib/mysql
+        sudo rm -rf /etc/mysql
+        # Install mysql80
+        sudo apt-key adv --keyserver keyserver.ubuntu.com --recv-keys A8D3785C
+        wget -c https://dev.mysql.com/get/mysql-apt-config_0.8.33-1_all.deb
+        echo mysql-apt-config mysql-apt-config/select-server select mysql-8.0 | sudo debconf-set-selections
+        sudo DEBIAN_FRONTEND="noninteractive" dpkg -i mysql-apt-config*
+        sudo apt-get update
+        sudo DEBIAN_FRONTEND="noninteractive" apt-get install -y mysql-server mysql-client
+        # Install everything else we need, and configure
+        sudo apt-get install -y make unzip g++ etcd-client etcd-server curl git wget eatmydata
+        sudo service mysql stop
+        sudo service etcd stop
+        sudo bash -c "echo '/usr/sbin/mysqld { }' > /etc/apparmor.d/usr.sbin.mysqld" # https://bugs.launchpad.net/ubuntu/+source/mariadb-10.1/+bug/1806263
+        sudo ln -s /etc/apparmor.d/usr.sbin.mysqld /etc/apparmor.d/disable/
+        sudo apparmor_parser -R /etc/apparmor.d/usr.sbin.mysqld || echo "could not remove mysqld profile"
+
+        # install JUnit report formatter
+        go install github.com/vitessio/go-junit-report@HEAD
+
+    # Build current commit's binaries
+    - name: Get dependencies for this commit
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        go mod download
+
+    - name: Building the binaries for this commit
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      timeout-minutes: 10
+      run: |
+        source build.env
+        NOVTADMINBUILD=1 make build
+        mkdir -p /tmp/vitess-build-current/
+        cp -R bin /tmp/vitess-build-current/
+        rm -Rf bin/*
+
+    # Checkout to the last release of Vitess
+    - name: Check out other version's code (${{ steps.output-previous-release-ref.outputs.previous_release_ref }})
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332 # v4.1.7
+      with:
+        ref: ${{ steps.output-previous-release-ref.outputs.previous_release_ref }}
+
+    - name: Get dependencies for the last release
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        go mod download
+
+    - name: Building last release's binaries
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      timeout-minutes: 10
+      run: |
+        source build.env
+        NOVTADMINBUILD=1 make build
+        mkdir -p /tmp/vitess-build-other/
+        cp -R bin /tmp/vitess-build-other/
+        rm -Rf bin/*
+
+    - name: Convert ErrorContains checks to Error checks
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        find ./go/test/endtoend -name '*.go' -exec sed -i 's/ErrorContains/Error/g' {} +
+        find ./go/test/endtoend -name '*.go' -exec sed -i 's/EqualError/Error/g' {} +
+
+    # Swap the binaries in the bin. Use vtgate version n-1 and keep vttablet at version n
+    - name: Use last release's VTGate
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        source build.env
+
+        cp -r /tmp/vitess-build-current/bin/* $PWD/bin/
+        rm -f $PWD/bin/vtgate
+        cp /tmp/vitess-build-other/bin/vtgate $PWD/bin/vtgate
+        vtgate --version
+
+    # Running a test with vtgate at version n-1 and vttablet/vtctld at version n
+    - name: Run query serving tests (vtgate=N-1, vttablet=N, vtctld=N)
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        rm -rf /tmp/vtdataroot
+        mkdir -p /tmp/vtdataroot
+
+        source build.env
+        eatmydata -- go run test.go -skip-build -keep-data=false -docker=false -print-log -follow -tag upgrade_downgrade_query_serving_queries_2
+
+    # Swap the binaries again. This time, vtgate will be at version n, and vttablet/vtctld will be at version n-1
+    - name: Use current version VTGate, and other version VTTablet/VTctld
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        source build.env
+
+        rm -f $PWD/bin/vtgate $PWD/bin/vttablet $PWD/bin/mysqlctl $PWD/bin/mysqlctld
+        cp /tmp/vitess-build-current/bin/vtgate $PWD/bin/vtgate
+        
+        cp /tmp/vitess-build-other/bin/vtctld $PWD/bin
+        cp /tmp/vitess-build-other/bin/vtctldclient $PWD/bin
+        cp /tmp/vitess-build-other/bin/vtctl $PWD/bin
+        cp /tmp/vitess-build-other/bin/vtctlclient $PWD/bin
+        
+        cp /tmp/vitess-build-other/bin/vttablet $PWD/bin/vttablet
+        cp /tmp/vitess-build-other/bin/mysqlctl $PWD/bin/mysqlctl
+        cp /tmp/vitess-build-other/bin/mysqlctld $PWD/bin/mysqlctld
+        vtgate --version
+        vttablet --version
+
+    # Running a test with vtgate at version n and vttablet/vtctld at version n-1
+    - name: Run query serving tests (vtgate=N, vttablet=N-1, vtctld=N-1)
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        rm -rf /tmp/vtdataroot
+        mkdir -p /tmp/vtdataroot
+
+        source build.env
+        eatmydata -- go run test.go -skip-build -keep-data=false -docker=false -print-log -follow -tag upgrade_downgrade_query_serving_queries_2
diff --git a/.github/workflows/upgrade_downgrade_test_query_serving_queries_2_next_release.yml b/.github/workflows/upgrade_downgrade_test_query_serving_queries_2_next_release.yml
new file mode 100644
index 00000000000..d3cfc662b5b
--- /dev/null
+++ b/.github/workflows/upgrade_downgrade_test_query_serving_queries_2_next_release.yml
@@ -0,0 +1,208 @@
+name: Query Serving (Queries - 2) Next Release - Upgrade Downgrade Testing
+on:
+  push:
+  pull_request:
+
+concurrency:
+  group: format('{0}-{1}', ${{ github.ref }}, 'Upgrade Downgrade Testing Query Serving (Queries - 2) Next Release')
+  cancel-in-progress: true
+
+permissions: read-all
+
+# This test ensures that our end-to-end tests work using Vitess components
+# (vtgate, vttablet, etc) built on different versions.
+
+jobs:
+
+  upgrade_downgrade_test:
+    name: Run Upgrade Downgrade Test - Query Serving (Queries - 2) Next Release
+    runs-on: gh-hosted-runners-16cores-1-24.04
+
+    steps:
+    - name: Skip CI
+      run: |
+        if [[ "${{contains( github.event.pull_request.labels.*.name, 'Skip CI')}}" == "true" ]]; then
+          echo "skipping CI due to the 'Skip CI' label"
+          exit 1
+        fi
+
+    - name: Check out commit's code
+      uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332 # v4.1.7
+      with:
+        fetch-depth: 0
+
+    - name: Set output with latest release branch
+      id: output-next-release-ref
+      run: |
+        next_release_ref=$(./tools/get_next_release.sh ${{github.base_ref}} ${{github.ref}})
+        echo $next_release_ref
+        echo "next_release_ref=${next_release_ref}" >> $GITHUB_OUTPUT
+
+    - name: Check if workflow needs to be skipped
+      id: skip-workflow
+      run: |
+        skip='false'
+        if [[ "${{github.event.pull_request}}" ==  "" ]] && [[ "${{github.ref}}" != "refs/heads/main" ]] && [[ ! "${{github.ref}}" =~ ^refs/heads/release-[0-9]+\.[0-9]$ ]] && [[ ! "${{github.ref}}" =~ "refs/tags/.*" ]]; then
+          skip='true'
+        fi
+        if [[ "${{steps.output-next-release-ref.outputs.next_release_ref}}" == "" ]]; then
+          skip='true'
+        fi
+        echo Skip ${skip}
+        echo "skip-workflow=${skip}" >> $GITHUB_OUTPUT
+
+    - name: Check for changes in relevant files
+      if: steps.skip-workflow.outputs.skip-workflow == 'false'
+      uses: dorny/paths-filter@ebc4d7e9ebcb0b1eb21480bb8f43113e996ac77a # v3.0.1
+      id: changes
+      with:
+        token: ''
+        filters: |
+          end_to_end:
+            - 'go/**'
+            - 'go/**/*.go'
+            - 'test.go'
+            - 'Makefile'
+            - 'build.env'
+            - 'go.sum'
+            - 'go.mod'
+            - 'proto/*.proto'
+            - 'tools/**'
+            - 'config/**'
+            - 'bootstrap.sh'
+            - '.github/workflows/upgrade_downgrade_test_query_serving_queries_next_release.yml'
+
+    - name: Set up Go
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
+      with:
+        go-version-file: go.mod
+
+    - name: Set up python
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      uses: actions/setup-python@39cd14951b08e74b54015e9e001cdefcf80e669f # v5.1.1
+
+    - name: Tune the OS
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        sudo sysctl -w net.ipv4.ip_local_port_range="22768 65535"
+
+    - name: Get base dependencies
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        sudo DEBIAN_FRONTEND="noninteractive" apt-get update
+        # Uninstall any nextly installed MySQL first
+        sudo systemctl stop apparmor
+        sudo DEBIAN_FRONTEND="noninteractive" apt-get remove -y --purge mysql-server mysql-client mysql-common
+        sudo apt-get -y autoremove
+        sudo apt-get -y autoclean
+        sudo deluser mysql
+        sudo rm -rf /var/lib/mysql
+        sudo rm -rf /etc/mysql
+        # Install mysql80
+        sudo apt-key adv --keyserver keyserver.ubuntu.com --recv-keys A8D3785C
+        wget -c https://dev.mysql.com/get/mysql-apt-config_0.8.33-1_all.deb
+        echo mysql-apt-config mysql-apt-config/select-server select mysql-8.0 | sudo debconf-set-selections
+        sudo DEBIAN_FRONTEND="noninteractive" dpkg -i mysql-apt-config*
+        sudo apt-get update
+        sudo DEBIAN_FRONTEND="noninteractive" apt-get install -y mysql-server mysql-client
+        # Install everything else we need, and configure
+        sudo apt-get install -y make unzip g++ etcd-client etcd-server curl git wget eatmydata
+        sudo service mysql stop
+        sudo service etcd stop
+        sudo bash -c "echo '/usr/sbin/mysqld { }' > /etc/apparmor.d/usr.sbin.mysqld" # https://bugs.launchpad.net/ubuntu/+source/mariadb-10.1/+bug/1806263
+        sudo ln -s /etc/apparmor.d/usr.sbin.mysqld /etc/apparmor.d/disable/
+        sudo apparmor_parser -R /etc/apparmor.d/usr.sbin.mysqld || echo "could not remove mysqld profile"
+
+        # install JUnit report formatter
+        go install github.com/vitessio/go-junit-report@HEAD
+
+    # Checkout to the next release of Vitess
+    - name: Check out other version's code (${{ steps.output-next-release-ref.outputs.next_release_ref }})
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332 # v4.1.7
+      with:
+        ref: ${{ steps.output-next-release-ref.outputs.next_release_ref }}
+
+    - name: Get dependencies for the next release
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        go mod download
+
+    - name: Building next release's binaries
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      timeout-minutes: 10
+      run: |
+        source build.env
+        NOVTADMINBUILD=1 make build
+        mkdir -p /tmp/vitess-build-other/
+        cp -R bin /tmp/vitess-build-other/
+        rm -Rf bin/*
+
+    # Checkout to this build's commit
+    - name: Check out commit's code
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332 # v4.1.7
+
+    - name: Get dependencies for this commit
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        go mod download
+
+    - name: Building the binaries for this commit
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      timeout-minutes: 10
+      run: |
+        source build.env
+        NOVTADMINBUILD=1 make build
+        mkdir -p /tmp/vitess-build-current/
+        cp -R bin /tmp/vitess-build-current/
+
+    - name: Convert ErrorContains checks to Error checks
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        find ./go/test/endtoend -name '*.go' -exec sed -i 's/ErrorContains/Error/g' {} +
+        find ./go/test/endtoend -name '*.go' -exec sed -i 's/EqualError/Error/g' {} +
+
+    # Swap the binaries in the bin. Use vtgate version n+1 and keep vttablet at version n
+    - name: Use next release's VTGate
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        source build.env
+        rm -f $PWD/bin/vtgate
+        cp /tmp/vitess-build-other/bin/vtgate $PWD/bin/vtgate
+        vtgate --version
+
+    # Running a test with vtgate at version n+1 and vttablet at version n
+    - name: Run query serving tests (vtgate=N+1, vttablet=N)
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        rm -rf /tmp/vtdataroot
+        mkdir -p /tmp/vtdataroot
+
+        source build.env
+        eatmydata -- go run test.go -skip-build -keep-data=false -docker=false -print-log -follow -tag upgrade_downgrade_query_serving_queries_2
+
+    # Swap the binaries again. This time, vtgate will be at version n, and vttablet will be at version n+1
+    - name: Use current version VTGate, and other version VTTablet
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        source build.env
+
+        rm -f $PWD/bin/vtgate $PWD/bin/vttablet $PWD/bin/mysqlctl $PWD/bin/mysqlctld
+        cp /tmp/vitess-build-current/bin/vtgate $PWD/bin/vtgate
+        cp /tmp/vitess-build-other/bin/vttablet $PWD/bin/vttablet
+        cp /tmp/vitess-build-other/bin/mysqlctl $PWD/bin/mysqlctl
+        cp /tmp/vitess-build-other/bin/mysqlctld $PWD/bin/mysqlctld
+        vtgate --version
+        vttablet --version
+
+    # Running a test with vtgate at version n and vttablet at version n+1
+    - name: Run query serving tests (vtgate=N, vttablet=N+1)
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        rm -rf /tmp/vtdataroot
+        mkdir -p /tmp/vtdataroot
+
+        source build.env
+        eatmydata -- go run test.go -skip-build -keep-data=false -docker=false -print-log -follow -tag upgrade_downgrade_query_serving_queries_2
diff --git a/.github/workflows/upgrade_downgrade_test_query_serving_schema.yml b/.github/workflows/upgrade_downgrade_test_query_serving_schema.yml
index 4b5fad0ab29..5616c247888 100644
--- a/.github/workflows/upgrade_downgrade_test_query_serving_schema.yml
+++ b/.github/workflows/upgrade_downgrade_test_query_serving_schema.yml
@@ -75,7 +75,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
       with:
-        go-version: 1.22.7
+        go-version: 1.23.4
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_reparent_old_vtctl.yml b/.github/workflows/upgrade_downgrade_test_reparent_old_vtctl.yml
index 701025d7ecc..532be0b998e 100644
--- a/.github/workflows/upgrade_downgrade_test_reparent_old_vtctl.yml
+++ b/.github/workflows/upgrade_downgrade_test_reparent_old_vtctl.yml
@@ -75,7 +75,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
       with:
-        go-version: 1.22.7
+        go-version: 1.23.4
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_reparent_old_vttablet.yml b/.github/workflows/upgrade_downgrade_test_reparent_old_vttablet.yml
index 8b121d4af10..2804e757652 100644
--- a/.github/workflows/upgrade_downgrade_test_reparent_old_vttablet.yml
+++ b/.github/workflows/upgrade_downgrade_test_reparent_old_vttablet.yml
@@ -75,7 +75,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
       with:
-        go-version: 1.22.7
+        go-version: 1.23.4
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_semi_sync.yml b/.github/workflows/upgrade_downgrade_test_semi_sync.yml
index f12e323654e..76706f77b65 100644
--- a/.github/workflows/upgrade_downgrade_test_semi_sync.yml
+++ b/.github/workflows/upgrade_downgrade_test_semi_sync.yml
@@ -72,7 +72,7 @@ jobs:
         if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
         uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
         with:
-          go-version: 1.22.7
+          go-version: 1.23.4
 
       - name: Set up python
         if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/changelog/22.0/22.0.0/summary.md b/changelog/22.0/22.0.0/summary.md
index a54a7a4a9ae..bfc55064148 100644
--- a/changelog/22.0/22.0.0/summary.md
+++ b/changelog/22.0/22.0.0/summary.md
@@ -10,6 +10,7 @@
   - **[VTOrc Config File Changes](#vtorc-config-file-changes)**
   - **[VTGate Config File Changes](#vtgate-config-file-changes)**
   - **[Support for More Efficient JSON Replication](#efficient-json-replication)**
+  - **[Support for LAST_INSERT_ID(x)](#last-insert-id)**
 - **[Minor Changes](#minor-changes)**
   - **[VTTablet Flags](#flags-vttablet)**
   - **[Topology read concurrency behaviour changes](#topo-read-concurrency-changes)**
@@ -80,6 +81,12 @@ In [#7345](https://github.com/vitessio/vitess/pull/17345) we added support for [
 
 If you are using MySQL 8.0 or later and using JSON columns, you can now enable this MySQL feature across your Vitess cluster(s) to lower the disk space needed for binary logs and improve the CPU and memory usage in both `mysqld` (standard intrashard MySQL replication) and `vttablet` ([VReplication](https://vitess.io/docs/reference/vreplication/vreplication/)) without losing any capabilities or features.
 
+### <a id="last-insert-id"/>Support for `LAST_INSERT_ID(x)`</a>
+
+In [#17408](https://github.com/vitessio/vitess/pull/17408) and [#17409](https://github.com/vitessio/vitess/pull/17409), we added the ability to use `LAST_INSERT_ID(x)` in Vitess directly at vtgate. This improvement allows certain queries—like `SELECT last_insert_id(123);` or `SELECT last_insert_id(count(*)) ...`—to be handled without relying on MySQL for the final value.
+
+**Limitations**:
+- When using `LAST_INSERT_ID(x)` in ordered queries (e.g., `SELECT last_insert_id(col) FROM table ORDER BY foo`), MySQL sets the session’s last-insert-id value according to the *last row returned*. Vitess does not guarantee the same behavior.
 
 ## <a id="minor-changes"/>Minor Changes</a>
 
diff --git a/examples/local/401_backup.sh b/examples/local/401_backup.sh
new file mode 100755
index 00000000000..1529440ea26
--- /dev/null
+++ b/examples/local/401_backup.sh
@@ -0,0 +1,40 @@
+# Copyright 2025 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 script takes backups of the 'customer' keyspace and all its shards.
+
+# Load common environment variables and functions
+source ../common/env.sh
+
+# Set keyspace and shard details for the 'customer' keyspace
+KEYSPACE="customer"
+SHARDS=("-80" "80-")
+
+# Ensure the keyspace and shards are healthy
+echo "Ensuring keyspace $KEYSPACE exists and shards are healthy..."
+for shard in "${SHARDS[@]}"; do
+    if ! wait_for_healthy_shard "$KEYSPACE" "$shard"; then
+        echo "Shard $shard is not healthy. Exiting..."
+        exit 1
+    fi
+done
+
+# Backup all shards of the customer keyspace
+for shard in "${SHARDS[@]}"; do
+    echo "Backing up shard $shard in keyspace $KEYSPACE..."
+    vtctldclient BackupShard "$KEYSPACE/$shard" || fail "Backup failed for shard $shard."
+    echo "Backup succeeded for shard $shard."
+done
+
+echo "Backup process completed successfully for all shards in $KEYSPACE."
diff --git a/examples/local/402_list_backup.sh b/examples/local/402_list_backup.sh
new file mode 100755
index 00000000000..c19dc5f6df1
--- /dev/null
+++ b/examples/local/402_list_backup.sh
@@ -0,0 +1,28 @@
+# Copyright 2025 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.
+
+# Load common environment variables and functions
+source ../common/env.sh  # Import necessary environment variables and functions from a common script
+
+# Set keyspace and shard details for the 'customer' keyspace
+KEYSPACE="customer"  # Define the keyspace to work with
+SHARDS=("-80" "80-")  # Define the shards within the keyspace to list backups for
+
+# List backups for each shard
+for shard in "${SHARDS[@]}"; do  # Loop through each shard defined earlier
+    echo "Listing available backups for keyspace $KEYSPACE and shard $shard..."  # Log the start of the backup listing
+    vtctldclient GetBackups "$KEYSPACE/$shard" || echo "Failed to list backups for keyspace $KEYSPACE and shard $shard"  # Attempt to list backups; log failure if it occurs
+done
+
+echo "Backup listing process completed."  # Log completion of the backup listing process
diff --git a/examples/local/403_restore_from_backup.sh b/examples/local/403_restore_from_backup.sh
new file mode 100755
index 00000000000..73fafd34aa0
--- /dev/null
+++ b/examples/local/403_restore_from_backup.sh
@@ -0,0 +1,46 @@
+# Copyright 2025 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 script restores the first replica tablet from backups for the 'customer' keyspace.
+
+# Load common environment variables and functions
+source ../common/env.sh  # Import necessary environment variables and functions from a common script
+
+# Set keyspace and shard details for the 'customer' keyspace
+KEYSPACE="customer"  # Define the keyspace to work with
+SHARDS=("-80" "80-")  # Define the shards within the keyspace to restore
+
+# Restore all shards of the customer keyspace from backups
+for shard in "${SHARDS[@]}"; do  # Loop through each shard defined earlier
+    echo "Finding replica tablets for shard $shard..."  # Log the start of the tablet search
+
+    # Fetch the list of replica tablets for the current shard
+    REPLICA_TABLETS=$(vtctldclient GetTablets --keyspace="$KEYSPACE" --shard="$shard" --tablet-type=replica | awk '{print $1}')  # Extract the first column containing tablet names
+    REPLICA_COUNT=$(echo "$REPLICA_TABLETS" | wc -l)  # Count the number of replica tablets found
+
+    # Check if any replica tablets were found
+    if [ "$REPLICA_COUNT" -lt 1 ]; then  # If the count is less than 1, no replicas were found
+        echo "No replica tablets found for shard $shard. Exiting..."  # Log a message and exit if none are found
+        exit 1  # Exit the script with an error code
+    fi
+
+    # Choose the first replica for restoration
+    RESTORE_TABLET=$(echo "$REPLICA_TABLETS" | head -n 1)  # Select the first replica tablet from the list
+    echo "Restoring tablet $RESTORE_TABLET from backup for shard $shard..."  # Log the restoration action
+
+    # Restore from backup and handle any failures
+    vtctldclient RestoreFromBackup "$RESTORE_TABLET" || fail "Restore failed for tablet $RESTORE_TABLET"  # Attempt to restore from backup and log an error message if it fails
+done
+
+echo "Restore process completed successfully for $KEYSPACE."  # Log completion of the restore process
diff --git a/examples/local/401_teardown.sh b/examples/local/501_teardown.sh
similarity index 100%
rename from examples/local/401_teardown.sh
rename to examples/local/501_teardown.sh
diff --git a/go/test/endtoend/onlineddl/flow/onlineddl_flow_test.go b/go/test/endtoend/onlineddl/flow/onlineddl_flow_test.go
index ee8141860f4..d34b63b833b 100644
--- a/go/test/endtoend/onlineddl/flow/onlineddl_flow_test.go
+++ b/go/test/endtoend/onlineddl/flow/onlineddl_flow_test.go
@@ -244,7 +244,7 @@ func TestOnlineDDLFlow(t *testing.T) {
 						select {
 						case <-ticker.C:
 						case <-workloadCtx.Done():
-							t.Logf("Terminating routine throttler check")
+							fmt.Println("Terminating routine throttler check")
 							return
 						}
 					}
@@ -258,8 +258,8 @@ func TestOnlineDDLFlow(t *testing.T) {
 				wg.Add(1)
 				go func() {
 					defer cancel()
-					defer t.Logf("Terminating workload")
 					defer wg.Done()
+					defer fmt.Println("Terminating workload")
 					runMultipleConnections(workloadCtx, t)
 				}()
 			})
diff --git a/go/test/endtoend/utils/cmp.go b/go/test/endtoend/utils/cmp.go
index dd9614e79fa..b2e1eca03e9 100644
--- a/go/test/endtoend/utils/cmp.go
+++ b/go/test/endtoend/utils/cmp.go
@@ -215,6 +215,18 @@ func (mcmp *MySQLCompare) Exec(query string) *sqltypes.Result {
 	return vtQr
 }
 
+// ExecVitessAndMySQLDifferentQueries executes Vitess and MySQL with the queries provided.
+func (mcmp *MySQLCompare) ExecVitessAndMySQLDifferentQueries(vtQ, mQ string) *sqltypes.Result {
+	mcmp.t.Helper()
+	vtQr, err := mcmp.VtConn.ExecuteFetch(vtQ, 1000, true)
+	require.NoError(mcmp.t, err, "[Vitess Error] for query: "+vtQ)
+
+	mysqlQr, err := mcmp.MySQLConn.ExecuteFetch(mQ, 1000, true)
+	require.NoError(mcmp.t, err, "[MySQL Error] for query: "+mQ)
+	compareVitessAndMySQLResults(mcmp.t, vtQ, mcmp.VtConn, vtQr, mysqlQr, CompareOptions{})
+	return vtQr
+}
+
 // ExecAssert is the same as Exec, but it only does assertions, it won't FailNow
 func (mcmp *MySQLCompare) ExecAssert(query string) *sqltypes.Result {
 	mcmp.t.Helper()
diff --git a/go/test/endtoend/vtgate/plan_tests/main_test.go b/go/test/endtoend/vtgate/plan_tests/main_test.go
index 504ec3ffb26..2dc2e70120b 100644
--- a/go/test/endtoend/vtgate/plan_tests/main_test.go
+++ b/go/test/endtoend/vtgate/plan_tests/main_test.go
@@ -22,6 +22,7 @@ import (
 	"os"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"vitess.io/vitess/go/mysql"
@@ -86,7 +87,7 @@ func TestMain(m *testing.M) {
 		// TODO: (@GuptaManan100/@systay): Also run the tests with normalizer on.
 		clusterInstance.VtGateExtraArgs = append(clusterInstance.VtGateExtraArgs,
 			"--normalize_queries=false",
-			"--schema_change_signal=false",
+			"--schema_change_signal=true",
 		)
 
 		// Start vtgate
@@ -178,7 +179,7 @@ func verifyTestExpectations(t *testing.T, pd engine.PrimitiveDescription, test p
 	// 1. Verify that the Join primitive sees atleast 1 row on the left side.
 	engine.WalkPrimitiveDescription(pd, func(description engine.PrimitiveDescription) {
 		if description.OperatorType == "Join" {
-			require.NotZero(t, description.Inputs[0].RowsReceived[0])
+			assert.NotZero(t, description.Inputs[0].RowsReceived[0])
 		}
 	})
 
diff --git a/go/test/endtoend/vtgate/plan_tests/plan_e2e_test.go b/go/test/endtoend/vtgate/plan_tests/plan_e2e_test.go
index 5c5447fe6b6..0068616c3b8 100644
--- a/go/test/endtoend/vtgate/plan_tests/plan_e2e_test.go
+++ b/go/test/endtoend/vtgate/plan_tests/plan_e2e_test.go
@@ -19,11 +19,22 @@ package plan_tests
 import (
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
 	"vitess.io/vitess/go/test/endtoend/utils"
+	"vitess.io/vitess/go/vt/sqlparser"
 )
 
 func TestE2ECases(t *testing.T) {
-	e2eTestCaseFiles := []string{"select_cases.json", "filter_cases.json", "dml_cases.json"}
+	err := utils.WaitForAuthoritative(t, "main", "source_of_ref", clusterInstance.VtgateProcess.ReadVSchema)
+	require.NoError(t, err)
+
+	e2eTestCaseFiles := []string{
+		"select_cases.json",
+		"filter_cases.json",
+		"dml_cases.json",
+		"reference_cases.json",
+	}
 	mcmp, closer := start(t)
 	defer closer()
 	loadSampleData(t, mcmp)
@@ -34,7 +45,11 @@ func TestE2ECases(t *testing.T) {
 				if test.SkipE2E {
 					mcmp.AsT().Skip(test.Query)
 				}
-				mcmp.Exec(test.Query)
+				stmt, err := sqlparser.NewTestParser().Parse(test.Query)
+				require.NoError(mcmp.AsT(), err)
+				sqlparser.RemoveKeyspaceIgnoreSysSchema(stmt)
+
+				mcmp.ExecVitessAndMySQLDifferentQueries(test.Query, sqlparser.String(stmt))
 				pd := utils.ExecTrace(mcmp.AsT(), mcmp.VtConn, test.Query)
 				verifyTestExpectations(mcmp.AsT(), pd, test)
 				if mcmp.VtConn.IsClosed() {
diff --git a/go/test/endtoend/vtgate/queries/misc/main_test.go b/go/test/endtoend/vtgate/queries/misc/main_test.go
index ee9be542634..536dfa7500a 100644
--- a/go/test/endtoend/vtgate/queries/misc/main_test.go
+++ b/go/test/endtoend/vtgate/queries/misc/main_test.go
@@ -95,7 +95,7 @@ func TestMain(m *testing.M) {
 		vtParams = clusterInstance.GetVTParams(keyspaceName)
 
 		// create mysql instance and connection parameters
-		conn, closer, err := utils.NewMySQL(clusterInstance, keyspaceName, schemaSQL)
+		conn, closer, err := utils.NewMySQL(clusterInstance, keyspaceName, schemaSQL, uschemaSQL)
 		if err != nil {
 			fmt.Println(err)
 			return 1
diff --git a/go/test/endtoend/vtgate/queries/misc/misc_test.go b/go/test/endtoend/vtgate/queries/misc/misc_test.go
index 9f9860bd0e0..7ab0fe7ef54 100644
--- a/go/test/endtoend/vtgate/queries/misc/misc_test.go
+++ b/go/test/endtoend/vtgate/queries/misc/misc_test.go
@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/vt/sqlparser"
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/stretchr/testify/assert"
@@ -163,6 +164,8 @@ func TestSetAndGetLastInsertID(t *testing.T) {
 		"update t1 set id2 = last_insert_id(%d) where id1 = 2",
 		"update t1 set id2 = 88 where id1 = last_insert_id(%d)",
 		"delete from t1 where id1 = last_insert_id(%d)",
+		"select id2, last_insert_id(count(*)) from t1 where %d group by id2",
+		"set @x = last_insert_id(%d)",
 	}
 
 	for _, workload := range []string{"olap", "oltp"} {
@@ -175,7 +178,7 @@ func TestSetAndGetLastInsertID(t *testing.T) {
 				require.NoError(t, err)
 			}
 
-			// Insert a row for UPDATE tests
+			// Insert a few rows for UPDATE tests
 			mcmp.Exec("insert into t1 (id1, id2) values (1, 10)")
 
 			for _, query := range queries {
@@ -186,6 +189,96 @@ func TestSetAndGetLastInsertID(t *testing.T) {
 	}
 }
 
+func TestSetAndGetLastInsertIDWithInsertUnsharded(t *testing.T) {
+	mcmp, closer := start(t)
+	defer closer()
+
+	tests := []string{
+		"insert into uks.unsharded(id1, id2) values (last_insert_id(%d),12)",
+		"insert into uks.unsharded(id1, id2) select last_insert_id(%d), 453",
+	}
+
+	i := 0
+	getVal := func() int {
+		defer func() { i++ }()
+		return i
+	}
+
+	runTests := func(mcmp *utils.MySQLCompare) {
+		for _, test := range tests {
+
+			lastInsertID := getVal()
+			query := fmt.Sprintf(test, lastInsertID)
+
+			stmt, err := sqlparser.NewTestParser().Parse(query)
+			require.NoError(mcmp.AsT(), err)
+			sqlparser.RemoveKeyspaceIgnoreSysSchema(stmt)
+
+			mcmp.ExecVitessAndMySQLDifferentQueries(query, sqlparser.String(stmt))
+			mcmp.Exec("select last_insert_id()")
+		}
+	}
+
+	for _, workload := range []string{"olap", "oltp"} {
+		mcmp.Run(workload, func(mcmp *utils.MySQLCompare) {
+			_, err := mcmp.VtConn.ExecuteFetch("set workload = "+workload, 1, false)
+			require.NoError(t, err)
+			runTests(mcmp)
+
+			// run the queries again, but inside a transaction this time
+			mcmp.Exec("begin")
+			runTests(mcmp)
+			mcmp.Exec("commit")
+		})
+	}
+
+	// Now test to set the last insert id to 0, see that it has changed correctly even if the value is 0
+	mcmp.ExecVitessAndMySQLDifferentQueries(
+		"insert into uks.unsharded(id1, id2) values (last_insert_id(0),12)",
+		"insert into unsharded(id1, id2) values (last_insert_id(0),12)",
+	)
+	mcmp.Exec("select last_insert_id()")
+}
+
+func TestSetAndGetLastInsertIDWithInsert(t *testing.T) {
+	mcmp, closer := start(t)
+	defer closer()
+
+	tests := []string{
+		"insert into t1(id1, id2) values (last_insert_id(%d) ,%d)",
+		"insert into t1(id1, id2) values (%d, last_insert_id(%d))",
+		"insert into t1(id1, id2) select last_insert_id(%d), %d",
+		"insert into t1(id1, id2) select last_insert_id(id1+%d), 12 from t1 where 1 > %d",
+	}
+
+	i := 0
+	getVal := func() int {
+		defer func() { i++ }()
+		return i
+	}
+
+	runTests := func(mcmp *utils.MySQLCompare) {
+		for _, test := range tests {
+			query := fmt.Sprintf(test, getVal(), getVal())
+			mcmp.Exec(query)
+			mcmp.Exec("select last_insert_id()")
+		}
+	}
+
+	for _, workload := range []string{"olap", "oltp"} {
+		mcmp.Run(workload, func(mcmp *utils.MySQLCompare) {
+			_, err := mcmp.VtConn.ExecuteFetch("set workload = "+workload, 1, false)
+			require.NoError(t, err)
+			runTests(mcmp)
+
+			// run the queries again, but inside a transaction this time
+			mcmp.Exec("begin")
+			runTests(mcmp)
+			mcmp.Exec("commit")
+		})
+	}
+}
+
 // TestVindexHints tests that vindex hints work as intended.
 func TestVindexHints(t *testing.T) {
 	mcmp, closer := start(t)
diff --git a/go/vt/schemadiff/key.go b/go/vt/schemadiff/key.go
index 865073a5a98..97f3af1630c 100644
--- a/go/vt/schemadiff/key.go
+++ b/go/vt/schemadiff/key.go
@@ -68,6 +68,16 @@ func (i *IndexDefinitionEntity) IsUnique() bool {
 	return i.IndexDefinition.Info.IsUnique()
 }
 
+// HasExpression returns true if the index uses an expression, e.g. `KEY idx1 ((id + 1))`.
+func (i *IndexDefinitionEntity) HasExpression() bool {
+	for _, col := range i.IndexDefinition.Columns {
+		if col.Expression != nil {
+			return true
+		}
+	}
+	return false
+}
+
 // HasNullable returns true if any of the columns in the index are nullable.
 func (i *IndexDefinitionEntity) HasNullable() bool {
 	for _, col := range i.ColumnList.Entities {
diff --git a/go/vt/schemadiff/onlineddl.go b/go/vt/schemadiff/onlineddl.go
index f02ccb1224d..06f3384c8fd 100644
--- a/go/vt/schemadiff/onlineddl.go
+++ b/go/vt/schemadiff/onlineddl.go
@@ -162,6 +162,11 @@ func PrioritizedUniqueKeys(createTableEntity *CreateTableEntity) *IndexDefinitio
 		if !key.IsUnique() {
 			continue
 		}
+		if key.HasExpression() {
+			// If the key has an expression this means it unreliably covers the columns,
+			// we cannot trust it.
+			continue
+		}
 		uniqueKeys = append(uniqueKeys, key)
 	}
 	sort.SliceStable(uniqueKeys, func(i, j int) bool {
diff --git a/go/vt/schemadiff/onlineddl_test.go b/go/vt/schemadiff/onlineddl_test.go
index 834490dca1b..f5309b4f943 100644
--- a/go/vt/schemadiff/onlineddl_test.go
+++ b/go/vt/schemadiff/onlineddl_test.go
@@ -932,6 +932,11 @@ func TestRevertible(t *testing.T) {
 			toSchema:            `id int primary key, e1 set('a', 'b'), e2 set('a'), e3 set('a', 'b', 'c'), e4 set('a', 'x'), e5 set('a', 'x', 'b'), e6 set('b'), e7 varchar(1), e8 tinyint`,
 			expandedColumnNames: `e3,e4,e5,e6,e7,e8`,
 		},
+		{
+			name:       "index with expression",
+			fromSchema: "id int, primary key (id), key idx1 ((id + 1))",
+			toSchema:   "id int, primary key (id), key idx2 ((id + 2))",
+		},
 	}
 
 	var (
diff --git a/go/vt/schemadiff/table.go b/go/vt/schemadiff/table.go
index e002ef18e15..e9bb35cb3bd 100644
--- a/go/vt/schemadiff/table.go
+++ b/go/vt/schemadiff/table.go
@@ -500,9 +500,14 @@ func (c *CreateTableEntity) IndexDefinitionEntities() []*IndexDefinitionEntity {
 	keys := c.CreateTable.TableSpec.Indexes
 	entities := make([]*IndexDefinitionEntity, len(keys))
 	for i, key := range keys {
-		colEntities := make([]*ColumnDefinitionEntity, len(key.Columns))
-		for i, keyCol := range key.Columns {
-			colEntities[i] = colMap[keyCol.Column.Lowered()]
+		colEntities := []*ColumnDefinitionEntity{}
+		for _, keyCol := range key.Columns {
+			colEntity, ok := colMap[keyCol.Column.Lowered()]
+			if !ok {
+				// This can happen if the index is on an expression, e.g. `KEY idx1 ((id + 1))`.
+				continue
+			}
+			colEntities = append(colEntities, colEntity)
 		}
 		entities[i] = NewIndexDefinitionEntity(c.Env, key, NewColumnDefinitionEntityList(colEntities))
 	}
diff --git a/go/vt/schemadiff/table_test.go b/go/vt/schemadiff/table_test.go
index 84c40d769c2..ac871dbd4af 100644
--- a/go/vt/schemadiff/table_test.go
+++ b/go/vt/schemadiff/table_test.go
@@ -891,6 +891,18 @@ func TestCreateTableDiff(t *testing.T) {
 				"+	KEY `i_idx` (`i`) INVISIBLE",
 			},
 		},
+		{
+			name:  "keys with expression",
+			from:  "create table t1 (id int, primary key (id), key idx1 ((id + 1)))",
+			to:    "create table t1 (id int, primary key (id), key idx2 ((id + 2)))",
+			diff:  "alter table t1 drop key idx1, add key idx2 ((id + 2))",
+			cdiff: "ALTER TABLE `t1` DROP KEY `idx1`, ADD KEY `idx2` ((`id` + 2))",
+			textdiffs: []string{
+				"-	KEY `idx1` ((`id` + 1))",
+				"+	KEY `idx2` ((`id` + 2))",
+			},
+		},
+
 		// FULLTEXT keys
 		{
 			name:  "add one fulltext key",
@@ -2564,6 +2576,12 @@ func TestValidate(t *testing.T) {
 			alter:     "alter table t engine=innodb",
 			expectErr: &DuplicateKeyNameError{Table: "t", Key: "PRIMARY"},
 		},
+		{
+			name:  "key with expression",
+			from:  "create table t (id int, primary key (id), key idx1 ((id + 1)))",
+			alter: "alter table t add key idx2 ((id + 2))",
+			to:    "create table t (id int, primary key (id), key idx1 ((id + 1)), key idx2 ((id + 2)))",
+		},
 		// partitions
 		{
 			name:      "drop column used by partitions",
diff --git a/go/vt/sqlparser/ast_funcs.go b/go/vt/sqlparser/ast_funcs.go
index 836c824010d..2891d532d16 100644
--- a/go/vt/sqlparser/ast_funcs.go
+++ b/go/vt/sqlparser/ast_funcs.go
@@ -2440,6 +2440,25 @@ func RemoveKeyspace(in SQLNode) {
 	})
 }
 
+// RemoveKeyspaceIgnoreSysSchema removes the Qualifier.Qualifier on all ColNames and Qualifier on all TableNames in the AST
+// except for the system schema.
+func RemoveKeyspaceIgnoreSysSchema(in SQLNode) {
+	Rewrite(in, nil, func(cursor *Cursor) bool {
+		switch expr := cursor.Node().(type) {
+		case *ColName:
+			if expr.Qualifier.Qualifier.NotEmpty() && !SystemSchema(expr.Qualifier.Qualifier.String()) {
+				expr.Qualifier.Qualifier = NewIdentifierCS("")
+			}
+		case TableName:
+			if expr.Qualifier.NotEmpty() && !SystemSchema(expr.Qualifier.String()) {
+				expr.Qualifier = NewIdentifierCS("")
+				cursor.Replace(expr)
+			}
+		}
+		return true
+	})
+}
+
 func convertStringToInt(integer string) int {
 	val, _ := strconv.Atoi(integer)
 	return val
diff --git a/go/vt/sqlparser/ast_test.go b/go/vt/sqlparser/ast_test.go
index f01b47cbd7b..c1484df7cc4 100644
--- a/go/vt/sqlparser/ast_test.go
+++ b/go/vt/sqlparser/ast_test.go
@@ -917,3 +917,11 @@ func TestCloneComments(t *testing.T) {
 		assert.Equal(t, "b", val)
 	}
 }
+
+func TestRemoveKeyspace(t *testing.T) {
+	stmt, err := NewTestParser().Parse("select 1 from uks.unsharded")
+	require.NoError(t, err)
+	RemoveKeyspaceIgnoreSysSchema(stmt)
+
+	require.Equal(t, "select 1 from unsharded", String(stmt))
+}
diff --git a/go/vt/vtgate/engine/cached_size.go b/go/vt/vtgate/engine/cached_size.go
index e59832cdab5..50d3a4b6bbf 100644
--- a/go/vt/vtgate/engine/cached_size.go
+++ b/go/vt/vtgate/engine/cached_size.go
@@ -465,7 +465,7 @@ func (cached *Insert) CachedSize(alloc bool) int64 {
 	}
 	size := int64(0)
 	if alloc {
-		size += int64(240)
+		size += int64(224)
 	}
 	// field InsertCommon vitess.io/vitess/go/vt/vtgate/engine.InsertCommon
 	size += cached.InsertCommon.CachedSize(false)
diff --git a/go/vt/vtgate/engine/fake_vcursor_test.go b/go/vt/vtgate/engine/fake_vcursor_test.go
index f27ca380876..aac3e9b584c 100644
--- a/go/vt/vtgate/engine/fake_vcursor_test.go
+++ b/go/vt/vtgate/engine/fake_vcursor_test.go
@@ -400,6 +400,20 @@ func (t *noopVCursor) GetDBDDLPluginName() string {
 	panic("unimplemented")
 }
 
+func (t *noopVCursor) SetLastInsertID(uint64) {}
+func (t *noopVCursor) VExplainLogging()       {}
+func (t *noopVCursor) DisableLogging()        {}
+func (t *noopVCursor) GetVExplainLogs() []ExecuteEntry {
+	return nil
+}
+func (t *noopVCursor) GetLogs() ([]ExecuteEntry, error) {
+	return nil, nil
+}
+
+// RecordMirrorStats implements VCursor.
+func (t *noopVCursor) RecordMirrorStats(sourceExecTime, targetExecTime time.Duration, targetErr error) {
+}
+
 var (
 	_ VCursor        = (*loggingVCursor)(nil)
 	_ SessionActions = (*loggingVCursor)(nil)
@@ -893,20 +907,6 @@ func (t *loggingVCursor) RecordMirrorStats(sourceExecTime, targetExecTime time.D
 	}
 }
 
-func (t *noopVCursor) VExplainLogging() {}
-func (t *noopVCursor) DisableLogging()  {}
-func (t *noopVCursor) GetVExplainLogs() []ExecuteEntry {
-	return nil
-}
-
-func (t *noopVCursor) GetLogs() ([]ExecuteEntry, error) {
-	return nil, nil
-}
-
-// RecordMirrorStats implements VCursor.
-func (t *noopVCursor) RecordMirrorStats(sourceExecTime, targetExecTime time.Duration, targetErr error) {
-}
-
 func expectResult(t *testing.T, result, want *sqltypes.Result) {
 	t.Helper()
 	fieldsResult := fmt.Sprintf("%v", result.Fields)
diff --git a/go/vt/vtgate/engine/insert.go b/go/vt/vtgate/engine/insert.go
index 10a4048572f..5bc206f7465 100644
--- a/go/vt/vtgate/engine/insert.go
+++ b/go/vt/vtgate/engine/insert.go
@@ -58,11 +58,9 @@ type Insert struct {
 
 	// Alias represents the row alias with columns if specified in the query.
 	Alias string
-
-	FetchLastInsertID bool
 }
 
-// newQueryInsert creates an Insert with a query string.
+// newQueryInsert creates an Insert with a query string. Used in testing.
 func newQueryInsert(opcode InsertOpcode, keyspace *vindexes.Keyspace, query string) *Insert {
 	return &Insert{
 		InsertCommon: InsertCommon{
@@ -73,7 +71,7 @@ func newQueryInsert(opcode InsertOpcode, keyspace *vindexes.Keyspace, query stri
 	}
 }
 
-// newInsert creates a new Insert.
+// newInsert creates a new Insert. Used in testing.
 func newInsert(
 	opcode InsertOpcode,
 	ignore bool,
diff --git a/go/vt/vtgate/engine/insert_common.go b/go/vt/vtgate/engine/insert_common.go
index 629d848d978..d4cae045e86 100644
--- a/go/vt/vtgate/engine/insert_common.go
+++ b/go/vt/vtgate/engine/insert_common.go
@@ -161,7 +161,7 @@ func (ins *InsertCommon) executeUnshardedTableQuery(ctx context.Context, vcursor
 	if err != nil {
 		return nil, err
 	}
-	qr, err := execShard(ctx, loggingPrimitive, vcursor, query, bindVars, rss[0], true, !ins.PreventAutoCommit /* canAutocommit */, false)
+	qr, err := execShard(ctx, loggingPrimitive, vcursor, query, bindVars, rss[0], true, !ins.PreventAutoCommit /* canAutocommit */, ins.FetchLastInsertID)
 	if err != nil {
 		return nil, err
 	}
diff --git a/go/vt/vtgate/engine/insert_select.go b/go/vt/vtgate/engine/insert_select.go
index bccee5f2cf9..af834858175 100644
--- a/go/vt/vtgate/engine/insert_select.go
+++ b/go/vt/vtgate/engine/insert_select.go
@@ -51,7 +51,7 @@ type (
 	}
 )
 
-// newInsertSelect creates a new InsertSelect.
+// newInsertSelect creates a new InsertSelect. Used in testing.
 func newInsertSelect(
 	ignore bool,
 	keyspace *vindexes.Keyspace,
diff --git a/go/vt/vtgate/engine/primitive.go b/go/vt/vtgate/engine/primitive.go
index e6fa102581e..7734dd81a6b 100644
--- a/go/vt/vtgate/engine/primitive.go
+++ b/go/vt/vtgate/engine/primitive.go
@@ -147,6 +147,8 @@ type (
 
 		// RecordMirrorStats is used to record stats about a mirror query.
 		RecordMirrorStats(time.Duration, time.Duration, error)
+
+		SetLastInsertID(uint64)
 	}
 
 	// SessionActions gives primitives ability to interact with the session state
diff --git a/go/vt/vtgate/engine/set.go b/go/vt/vtgate/engine/set.go
index 95fb5c87a32..f0de330cbed 100644
--- a/go/vt/vtgate/engine/set.go
+++ b/go/vt/vtgate/engine/set.go
@@ -22,23 +22,18 @@ import (
 	"fmt"
 	"strings"
 
-	"vitess.io/vitess/go/vt/sqlparser"
-	"vitess.io/vitess/go/vt/sysvars"
-
-	vtgatepb "vitess.io/vitess/go/vt/proto/vtgate"
-
-	"vitess.io/vitess/go/vt/log"
-
-	"vitess.io/vitess/go/vt/srvtopo"
-
-	"vitess.io/vitess/go/vt/vtgate/evalengine"
-
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/key"
+	"vitess.io/vitess/go/vt/log"
 	querypb "vitess.io/vitess/go/vt/proto/query"
+	vtgatepb "vitess.io/vitess/go/vt/proto/vtgate"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/schema"
+	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/srvtopo"
+	"vitess.io/vitess/go/vt/sysvars"
 	"vitess.io/vitess/go/vt/vterrors"
+	"vitess.io/vitess/go/vt/vtgate/evalengine"
 	"vitess.io/vitess/go/vt/vtgate/vindexes"
 )
 
diff --git a/go/vt/vtgate/evalengine/arena.go b/go/vt/vtgate/evalengine/arena.go
index ccfe63f514f..c5457f076e8 100644
--- a/go/vt/vtgate/evalengine/arena.go
+++ b/go/vt/vtgate/evalengine/arena.go
@@ -71,7 +71,7 @@ func (a *Arena) newEvalEnum(raw []byte, values *EnumSetValues) *evalEnum {
 	} else {
 		a.aEnum = append(a.aEnum, evalEnum{})
 	}
-	val := &a.aEnum[len(a.aInt64)-1]
+	val := &a.aEnum[len(a.aEnum)-1]
 	s := string(raw)
 	val.string = s
 	val.value = valueIdx(values, s)
@@ -84,7 +84,7 @@ func (a *Arena) newEvalSet(raw []byte, values *EnumSetValues) *evalSet {
 	} else {
 		a.aSet = append(a.aSet, evalSet{})
 	}
-	val := &a.aSet[len(a.aInt64)-1]
+	val := &a.aSet[len(a.aSet)-1]
 	s := string(raw)
 	val.string = s
 	val.set = evalSetBits(values, s)
diff --git a/go/vt/vtgate/evalengine/cached_size.go b/go/vt/vtgate/evalengine/cached_size.go
index c1ed1f9475c..d51c65c75b4 100644
--- a/go/vt/vtgate/evalengine/cached_size.go
+++ b/go/vt/vtgate/evalengine/cached_size.go
@@ -1181,6 +1181,18 @@ func (cached *builtinLastDay) CachedSize(alloc bool) int64 {
 	size += cached.CallExpr.CachedSize(false)
 	return size
 }
+func (cached *builtinLastInsertID) CachedSize(alloc bool) int64 {
+	if cached == nil {
+		return int64(0)
+	}
+	size := int64(0)
+	if alloc {
+		size += int64(48)
+	}
+	// field CallExpr vitess.io/vitess/go/vt/vtgate/evalengine.CallExpr
+	size += cached.CallExpr.CachedSize(false)
+	return size
+}
 func (cached *builtinLeftRight) CachedSize(alloc bool) int64 {
 	if cached == nil {
 		return int64(0)
diff --git a/go/vt/vtgate/evalengine/compiler_asm.go b/go/vt/vtgate/evalengine/compiler_asm.go
index dfb1a30bffc..7dda215353f 100644
--- a/go/vt/vtgate/evalengine/compiler_asm.go
+++ b/go/vt/vtgate/evalengine/compiler_asm.go
@@ -5138,3 +5138,18 @@ func (asm *assembler) Introduce(offset int, t sqltypes.Type, col collations.Type
 		return 1
 	}, "INTRODUCE (SP-1)")
 }
+
+func (asm *assembler) Fn_LAST_INSERT_ID() {
+	asm.emit(func(env *ExpressionEnv) int {
+		arg := env.vm.stack[env.vm.sp-1]
+		if arg == nil {
+			env.VCursor().SetLastInsertID(0)
+		} else {
+			iarg := evalToInt64(arg)
+			uarg := env.vm.arena.newEvalUint64(uint64(iarg.i))
+			env.vm.stack[env.vm.sp-1] = uarg
+			env.VCursor().SetLastInsertID(uarg.u)
+		}
+		return 1
+	}, "FN LAST_INSERT_ID UINT64(SP-1)")
+}
diff --git a/go/vt/vtgate/evalengine/compiler_test.go b/go/vt/vtgate/evalengine/compiler_test.go
index 88c13a479ed..343bb0cd043 100644
--- a/go/vt/vtgate/evalengine/compiler_test.go
+++ b/go/vt/vtgate/evalengine/compiler_test.go
@@ -760,6 +760,20 @@ func TestCompilerSingle(t *testing.T) {
 			expression: `WEEK(timestamp '2024-01-01 10:34:58', 1)`,
 			result:     `INT64(1)`,
 		},
+		{
+			expression: `column0 + 1`,
+			values:     []sqltypes.Value{sqltypes.MakeTrusted(sqltypes.Enum, []byte("foo"))},
+			// Returns 0, as unknown enums evaluate here to -1. We have this test to
+			// exercise the path to push enums onto the stack.
+			result: `FLOAT64(0)`,
+		},
+		{
+			expression: `column0 + 1`,
+			values:     []sqltypes.Value{sqltypes.MakeTrusted(sqltypes.Set, []byte("foo"))},
+			// Returns 1, as unknown sets evaluate here to 0. We have this test to
+			// exercise the path to push sets onto the stack.
+			result: `FLOAT64(1)`,
+		},
 	}
 
 	tz, _ := time.LoadLocation("Europe/Madrid")
@@ -903,6 +917,99 @@ func TestBindVarLiteral(t *testing.T) {
 	}
 }
 
+type testVcursor struct {
+	lastInsertID *uint64
+	env          *vtenv.Environment
+}
+
+func (t *testVcursor) TimeZone() *time.Location {
+	return time.UTC
+}
+
+func (t *testVcursor) GetKeyspace() string {
+	return "apa"
+}
+
+func (t *testVcursor) SQLMode() string {
+	return "oltp"
+}
+
+func (t *testVcursor) Environment() *vtenv.Environment {
+	return t.env
+}
+
+func (t *testVcursor) SetLastInsertID(id uint64) {
+	t.lastInsertID = &id
+}
+
+var _ evalengine.VCursor = (*testVcursor)(nil)
+
+func TestLastInsertID(t *testing.T) {
+	var testCases = []struct {
+		expression string
+		result     uint64
+		missing    bool
+	}{
+		{
+			expression: `last_insert_id(1)`,
+			result:     1,
+		}, {
+			expression: `12`,
+			missing:    true,
+		}, {
+			expression: `last_insert_id(666)`,
+			result:     666,
+		}, {
+			expression: `last_insert_id(null)`,
+			result:     0,
+		},
+	}
+
+	venv := vtenv.NewTestEnv()
+	for _, tc := range testCases {
+		t.Run(tc.expression, func(t *testing.T) {
+			expr, err := venv.Parser().ParseExpr(tc.expression)
+			require.NoError(t, err)
+
+			cfg := &evalengine.Config{
+				Collation:         collations.CollationUtf8mb4ID,
+				NoConstantFolding: true,
+				NoCompilation:     false,
+				Environment:       venv,
+			}
+			t.Run("eval", func(t *testing.T) {
+				cfg.NoCompilation = true
+				runTest(t, expr, cfg, tc)
+			})
+			t.Run("compiled", func(t *testing.T) {
+				cfg.NoCompilation = false
+				runTest(t, expr, cfg, tc)
+			})
+		})
+	}
+}
+
+func runTest(t *testing.T, expr sqlparser.Expr, cfg *evalengine.Config, tc struct {
+	expression string
+	result     uint64
+	missing    bool
+}) {
+	converted, err := evalengine.Translate(expr, cfg)
+	require.NoError(t, err)
+
+	vc := &testVcursor{env: vtenv.NewTestEnv()}
+	env := evalengine.NewExpressionEnv(context.Background(), nil, vc)
+
+	_, err = env.Evaluate(converted)
+	require.NoError(t, err)
+	if tc.missing {
+		require.Nil(t, vc.lastInsertID)
+	} else {
+		require.NotNil(t, vc.lastInsertID)
+		require.Equal(t, tc.result, *vc.lastInsertID)
+	}
+}
+
 func TestCompilerNonConstant(t *testing.T) {
 	var testCases = []struct {
 		expression string
diff --git a/go/vt/vtgate/evalengine/expr_env.go b/go/vt/vtgate/evalengine/expr_env.go
index 38a65f9b4e0..4a7f9849ab0 100644
--- a/go/vt/vtgate/evalengine/expr_env.go
+++ b/go/vt/vtgate/evalengine/expr_env.go
@@ -35,6 +35,7 @@ type VCursor interface {
 	GetKeyspace() string
 	SQLMode() string
 	Environment() *vtenv.Environment
+	SetLastInsertID(id uint64)
 }
 
 type (
@@ -140,6 +141,7 @@ func (e *emptyVCursor) GetKeyspace() string {
 func (e *emptyVCursor) SQLMode() string {
 	return config.DefaultSQLMode
 }
+func (e *emptyVCursor) SetLastInsertID(_ uint64) {}
 
 func NewEmptyVCursor(env *vtenv.Environment, tz *time.Location) VCursor {
 	return &emptyVCursor{env: env, tz: tz}
diff --git a/go/vt/vtgate/evalengine/fn_misc.go b/go/vt/vtgate/evalengine/fn_misc.go
index 8813b62f823..2a2119ee6f4 100644
--- a/go/vt/vtgate/evalengine/fn_misc.go
+++ b/go/vt/vtgate/evalengine/fn_misc.go
@@ -81,6 +81,10 @@ type (
 	builtinUUIDToBin struct {
 		CallExpr
 	}
+
+	builtinLastInsertID struct {
+		CallExpr
+	}
 )
 
 var _ IR = (*builtinInetAton)(nil)
@@ -95,6 +99,7 @@ var _ IR = (*builtinBinToUUID)(nil)
 var _ IR = (*builtinIsUUID)(nil)
 var _ IR = (*builtinUUID)(nil)
 var _ IR = (*builtinUUIDToBin)(nil)
+var _ IR = (*builtinLastInsertID)(nil)
 
 func (call *builtinInetAton) eval(env *ExpressionEnv) (eval, error) {
 	arg, err := call.arg1(env)
@@ -194,6 +199,33 @@ func (call *builtinInet6Aton) compile(c *compiler) (ctype, error) {
 	return ctype{Type: sqltypes.VarBinary, Flag: flagNullable, Col: collationBinary}, nil
 }
 
+func (call *builtinLastInsertID) eval(env *ExpressionEnv) (eval, error) {
+	arg, err := call.arg1(env)
+	if err != nil {
+		return nil, err
+	}
+	if arg == nil {
+		env.VCursor().SetLastInsertID(0)
+		return nil, err
+	}
+	insertID := uint64(evalToInt64(arg).i)
+	env.VCursor().SetLastInsertID(insertID)
+	return newEvalUint64(insertID), nil
+}
+
+func (call *builtinLastInsertID) compile(c *compiler) (ctype, error) {
+	arg, err := call.Arguments[0].compile(c)
+	if err != nil {
+		return ctype{}, err
+	}
+	c.asm.Fn_LAST_INSERT_ID()
+	return ctype{Type: sqltypes.Uint64, Flag: arg.Flag & flagNullable, Col: collationNumeric}, nil
+}
+
+func (call *builtinLastInsertID) constant() bool {
+	return false // we don't want this function to be simplified away
+}
+
 func printIPv6AsIPv4(addr netip.Addr) (netip.Addr, bool) {
 	b := addr.AsSlice()
 	if len(b) != 16 {
diff --git a/go/vt/vtgate/evalengine/integration/comparison_test.go b/go/vt/vtgate/evalengine/integration/comparison_test.go
index ea327601975..d559cb8ab1d 100644
--- a/go/vt/vtgate/evalengine/integration/comparison_test.go
+++ b/go/vt/vtgate/evalengine/integration/comparison_test.go
@@ -209,6 +209,10 @@ type vcursor struct {
 	env *vtenv.Environment
 }
 
+func (vc *vcursor) SetLastInsertID(id uint64) {}
+
+var _ evalengine.VCursor = (*vcursor)(nil)
+
 func (vc *vcursor) GetKeyspace() string {
 	return "vttest"
 }
diff --git a/go/vt/vtgate/evalengine/translate_builtin.go b/go/vt/vtgate/evalengine/translate_builtin.go
index 476ee32483b..1f8bd7798aa 100644
--- a/go/vt/vtgate/evalengine/translate_builtin.go
+++ b/go/vt/vtgate/evalengine/translate_builtin.go
@@ -662,6 +662,11 @@ func (ast *astCompiler) translateFuncExpr(fn *sqlparser.FuncExpr) (IR, error) {
 			return nil, argError(method)
 		}
 		return &builtinReplace{CallExpr: call, collate: ast.cfg.Collation}, nil
+	case "last_insert_id":
+		if len(args) != 1 {
+			return nil, argError(method)
+		}
+		return &builtinLastInsertID{CallExpr: call}, nil
 	default:
 		return nil, translateExprNotSupported(fn)
 	}
diff --git a/go/vt/vtgate/executorcontext/vcursor_impl.go b/go/vt/vtgate/executorcontext/vcursor_impl.go
index df989fd7a67..3f8d7def797 100644
--- a/go/vt/vtgate/executorcontext/vcursor_impl.go
+++ b/go/vt/vtgate/executorcontext/vcursor_impl.go
@@ -1594,3 +1594,9 @@ func (vc *VCursorImpl) GetContextWithTimeOut(ctx context.Context) (context.Conte
 func (vc *VCursorImpl) IgnoreMaxMemoryRows() bool {
 	return vc.ignoreMaxMemoryRows
 }
+
+func (vc *VCursorImpl) SetLastInsertID(id uint64) {
+	vc.SafeSession.mu.Lock()
+	defer vc.SafeSession.mu.Unlock()
+	vc.SafeSession.LastInsertId = id
+}
diff --git a/go/vt/vtgate/planbuilder/insert.go b/go/vt/vtgate/planbuilder/insert.go
index 80516871623..d3ad5afac72 100644
--- a/go/vt/vtgate/planbuilder/insert.go
+++ b/go/vt/vtgate/planbuilder/insert.go
@@ -51,7 +51,7 @@ func gen4InsertStmtPlanner(version querypb.ExecuteOptions_PlannerVersion, insStm
 	}
 	if ks != nil {
 		if tables[0].AutoIncrement == nil && !ctx.SemTable.ForeignKeysPresent() {
-			plan := insertUnshardedShortcut(insStmt, ks, tables)
+			plan := insertUnshardedShortcut(ctx, insStmt, ks, tables)
 			setCommentDirectivesOnPlan(plan, insStmt)
 			return newPlanResult(plan, operators.QualifiedTables(ks, tables)...), nil
 		}
@@ -90,12 +90,13 @@ func errOutIfPlanCannotBeConstructed(ctx *plancontext.PlanningContext, vTbl *vin
 	return ctx.SemTable.NotUnshardedErr
 }
 
-func insertUnshardedShortcut(stmt *sqlparser.Insert, ks *vindexes.Keyspace, tables []*vindexes.Table) engine.Primitive {
+func insertUnshardedShortcut(ctx *plancontext.PlanningContext, stmt *sqlparser.Insert, ks *vindexes.Keyspace, tables []*vindexes.Table) engine.Primitive {
 	eIns := &engine.Insert{
 		InsertCommon: engine.InsertCommon{
-			Opcode:    engine.InsertUnsharded,
-			Keyspace:  ks,
-			TableName: tables[0].Name.String(),
+			Opcode:            engine.InsertUnsharded,
+			Keyspace:          ks,
+			TableName:         tables[0].Name.String(),
+			FetchLastInsertID: ctx.SemTable.ShouldFetchLastInsertID(),
 		},
 	}
 	eIns.Query = generateQuery(stmt)
diff --git a/go/vt/vtgate/planbuilder/operator_transformers.go b/go/vt/vtgate/planbuilder/operator_transformers.go
index bc71c7195b4..b51eac449fc 100644
--- a/go/vt/vtgate/planbuilder/operator_transformers.go
+++ b/go/vt/vtgate/planbuilder/operator_transformers.go
@@ -190,6 +190,7 @@ func transformInsertionSelection(ctx *plancontext.PlanningContext, op *operators
 			ForceNonStreaming: op.ForceNonStreaming,
 			Generate:          autoIncGenerate(ins.AutoIncrement),
 			ColVindexes:       ins.ColVindexes,
+			FetchLastInsertID: ctx.SemTable.ShouldFetchLastInsertID(),
 		},
 		VindexValueOffset: ins.VindexValueOffset,
 	}
@@ -659,9 +660,8 @@ func buildInsertPrimitive(
 	}
 
 	eins := &engine.Insert{
-		InsertCommon:      ic,
-		VindexValues:      ins.VindexValues,
-		FetchLastInsertID: ctx.SemTable.ShouldFetchLastInsertID(),
+		InsertCommon: ic,
+		VindexValues: ins.VindexValues,
 	}
 
 	// we would need to generate the query on the fly. The only exception here is
diff --git a/go/vt/vtgate/planbuilder/operators/cte_merging.go b/go/vt/vtgate/planbuilder/operators/cte_merging.go
index cb19e06b2a7..0c1556c81e4 100644
--- a/go/vt/vtgate/planbuilder/operators/cte_merging.go
+++ b/go/vt/vtgate/planbuilder/operators/cte_merging.go
@@ -31,7 +31,7 @@ func tryMergeRecurse(ctx *plancontext.PlanningContext, in *RecurseCTE) (Operator
 }
 
 func tryMergeCTE(ctx *plancontext.PlanningContext, seed, term Operator, in *RecurseCTE) *Route {
-	seedRoute, termRoute, routingA, routingB, a, b, sameKeyspace := prepareInputRoutes(seed, term)
+	seedRoute, termRoute, routingA, routingB, a, b, sameKeyspace := prepareInputRoutes(ctx, seed, term)
 	if seedRoute == nil {
 		return nil
 	}
diff --git a/go/vt/vtgate/planbuilder/operators/delete.go b/go/vt/vtgate/planbuilder/operators/delete.go
index 81e36d54315..015220470e0 100644
--- a/go/vt/vtgate/planbuilder/operators/delete.go
+++ b/go/vt/vtgate/planbuilder/operators/delete.go
@@ -124,7 +124,7 @@ func createDeleteWithInputOp(ctx *plancontext.PlanningContext, del *sqlparser.De
 	}
 
 	var delOps []dmlOp
-	for _, target := range ctx.SemTable.Targets.Constituents() {
+	for _, target := range ctx.SemTable.DMLTargets.Constituents() {
 		op := createDeleteOpWithTarget(ctx, target, del.Ignore)
 		delOps = append(delOps, op)
 	}
@@ -322,7 +322,7 @@ func updateQueryGraphWithSource(ctx *plancontext.PlanningContext, input Operator
 			return op, NoRewrite
 		}
 		if len(qg.Tables) > 1 {
-			panic(vterrors.VT12001("DELETE on reference table with join"))
+			panic(vterrors.VT12001("DML on reference table with join"))
 		}
 		for _, tbl := range qg.Tables {
 			if tbl.ID != tblID {
diff --git a/go/vt/vtgate/planbuilder/operators/join_merging.go b/go/vt/vtgate/planbuilder/operators/join_merging.go
index c035b7d11ed..cb3569cf79e 100644
--- a/go/vt/vtgate/planbuilder/operators/join_merging.go
+++ b/go/vt/vtgate/planbuilder/operators/join_merging.go
@@ -28,7 +28,7 @@ import (
 // If they can be merged, a new operator with the merged routing is returned
 // If they cannot be merged, nil is returned.
 func (jm *joinMerger) mergeJoinInputs(ctx *plancontext.PlanningContext, lhs, rhs Operator, joinPredicates []sqlparser.Expr) *Route {
-	lhsRoute, rhsRoute, routingA, routingB, a, b, sameKeyspace := prepareInputRoutes(lhs, rhs)
+	lhsRoute, rhsRoute, routingA, routingB, a, b, sameKeyspace := prepareInputRoutes(ctx, lhs, rhs)
 	if lhsRoute == nil {
 		return nil
 	}
@@ -102,13 +102,13 @@ func mergeAnyShardRoutings(ctx *plancontext.PlanningContext, a, b *AnyShardRouti
 	}
 }
 
-func prepareInputRoutes(lhs Operator, rhs Operator) (*Route, *Route, Routing, Routing, routingType, routingType, bool) {
+func prepareInputRoutes(ctx *plancontext.PlanningContext, lhs Operator, rhs Operator) (*Route, *Route, Routing, Routing, routingType, routingType, bool) {
 	lhsRoute, rhsRoute := operatorsToRoutes(lhs, rhs)
 	if lhsRoute == nil || rhsRoute == nil {
 		return nil, nil, nil, nil, 0, 0, false
 	}
 
-	lhsRoute, rhsRoute, routingA, routingB, sameKeyspace := getRoutesOrAlternates(lhsRoute, rhsRoute)
+	lhsRoute, rhsRoute, routingA, routingB, sameKeyspace := getRoutesOrAlternates(ctx, lhsRoute, rhsRoute)
 
 	a, b := getRoutingType(routingA), getRoutingType(routingB)
 	return lhsRoute, rhsRoute, routingA, routingB, a, b, sameKeyspace
@@ -159,7 +159,7 @@ func (rt routingType) String() string {
 
 // getRoutesOrAlternates gets the Routings from each Route. If they are from different keyspaces,
 // we check if this is a table with alternates in other keyspaces that we can use
-func getRoutesOrAlternates(lhsRoute, rhsRoute *Route) (*Route, *Route, Routing, Routing, bool) {
+func getRoutesOrAlternates(ctx *plancontext.PlanningContext, lhsRoute, rhsRoute *Route) (*Route, *Route, Routing, Routing, bool) {
 	routingA := lhsRoute.Routing
 	routingB := rhsRoute.Routing
 	sameKeyspace := routingA.Keyspace() == routingB.Keyspace()
@@ -171,13 +171,17 @@ func getRoutesOrAlternates(lhsRoute, rhsRoute *Route) (*Route, *Route, Routing,
 		return lhsRoute, rhsRoute, routingA, routingB, sameKeyspace
 	}
 
-	if refA, ok := routingA.(*AnyShardRouting); ok {
+	// If we have a reference route, we will try to find an alternate route in same keyspace as other routing keyspace.
+	// If the reference route is part of DML table update target, alternate keyspace route cannot be considered.
+	if refA, ok := routingA.(*AnyShardRouting); ok &&
+		!TableID(lhsRoute).IsOverlapping(ctx.SemTable.DMLTargets) {
 		if altARoute := refA.AlternateInKeyspace(routingB.Keyspace()); altARoute != nil {
 			return altARoute, rhsRoute, altARoute.Routing, routingB, true
 		}
 	}
 
-	if refB, ok := routingB.(*AnyShardRouting); ok {
+	if refB, ok := routingB.(*AnyShardRouting); ok &&
+		!TableID(rhsRoute).IsOverlapping(ctx.SemTable.DMLTargets) {
 		if altBRoute := refB.AlternateInKeyspace(routingA.Keyspace()); altBRoute != nil {
 			return lhsRoute, altBRoute, routingA, altBRoute.Routing, true
 		}
diff --git a/go/vt/vtgate/planbuilder/operators/subquery_planning.go b/go/vt/vtgate/planbuilder/operators/subquery_planning.go
index a2aca74fb6e..e222ae0f343 100644
--- a/go/vt/vtgate/planbuilder/operators/subquery_planning.go
+++ b/go/vt/vtgate/planbuilder/operators/subquery_planning.go
@@ -730,7 +730,7 @@ func mergeSubqueryInputs(ctx *plancontext.PlanningContext, in, out Operator, joi
 		return nil
 	}
 
-	inRoute, outRoute, inRouting, outRouting, sameKeyspace := getRoutesOrAlternates(inRoute, outRoute)
+	inRoute, outRoute, inRouting, outRouting, sameKeyspace := getRoutesOrAlternates(ctx, inRoute, outRoute)
 	inner, outer := getRoutingType(inRouting), getRoutingType(outRouting)
 
 	switch {
diff --git a/go/vt/vtgate/planbuilder/operators/union_merging.go b/go/vt/vtgate/planbuilder/operators/union_merging.go
index 000d176b61a..6173b59e0dc 100644
--- a/go/vt/vtgate/planbuilder/operators/union_merging.go
+++ b/go/vt/vtgate/planbuilder/operators/union_merging.go
@@ -108,7 +108,7 @@ func mergeUnionInputs(
 	lhsExprs, rhsExprs sqlparser.SelectExprs,
 	distinct bool,
 ) (Operator, sqlparser.SelectExprs) {
-	lhsRoute, rhsRoute, routingA, routingB, a, b, sameKeyspace := prepareInputRoutes(lhs, rhs)
+	lhsRoute, rhsRoute, routingA, routingB, a, b, sameKeyspace := prepareInputRoutes(ctx, lhs, rhs)
 	if lhsRoute == nil {
 		return nil, nil
 	}
diff --git a/go/vt/vtgate/planbuilder/operators/update.go b/go/vt/vtgate/planbuilder/operators/update.go
index dd0a86c2de2..18a81175f7b 100644
--- a/go/vt/vtgate/planbuilder/operators/update.go
+++ b/go/vt/vtgate/planbuilder/operators/update.go
@@ -164,7 +164,7 @@ func createUpdateWithInputOp(ctx *plancontext.PlanningContext, upd *sqlparser.Up
 	ueMap := prepareUpdateExpressionList(ctx, upd)
 
 	var updOps []dmlOp
-	for _, target := range ctx.SemTable.Targets.Constituents() {
+	for _, target := range ctx.SemTable.DMLTargets.Constituents() {
 		op := createUpdateOpWithTarget(ctx, upd, target, ueMap[target])
 		updOps = append(updOps, op)
 	}
@@ -308,7 +308,7 @@ func errIfUpdateNotSupported(ctx *plancontext.PlanningContext, stmt *sqlparser.U
 		}
 	}
 
-	// Now we check if any of the foreign key columns that are being udpated have dependencies on other updated columns.
+	// Now we check if any of the foreign key columns that are being updated have dependencies on other updated columns.
 	// This is unsafe, and we currently don't support this in Vitess.
 	if err := ctx.SemTable.ErrIfFkDependentColumnUpdated(stmt.Exprs); err != nil {
 		panic(err)
diff --git a/go/vt/vtgate/planbuilder/plan_test.go b/go/vt/vtgate/planbuilder/plan_test.go
index df813b04dea..f3bed93e3c8 100644
--- a/go/vt/vtgate/planbuilder/plan_test.go
+++ b/go/vt/vtgate/planbuilder/plan_test.go
@@ -84,6 +84,7 @@ func (s *planTestSuite) TestPlan() {
 	s.addPKsProvided(vschema, "user", []string{"user_extra"}, []string{"id", "user_id"})
 	s.addPKsProvided(vschema, "ordering", []string{"order"}, []string{"oid", "region_id"})
 	s.addPKsProvided(vschema, "ordering", []string{"order_event"}, []string{"oid", "ename"})
+	s.addPKsProvided(vschema, "main", []string{"source_of_ref"}, []string{"id"})
 
 	// You will notice that some tests expect user.Id instead of user.id.
 	// This is because we now pre-create vindex columns in the symbol
@@ -305,6 +306,7 @@ func (s *planTestSuite) TestOne() {
 	s.addPKsProvided(vschema, "user", []string{"user_extra"}, []string{"id", "user_id"})
 	s.addPKsProvided(vschema, "ordering", []string{"order"}, []string{"oid", "region_id"})
 	s.addPKsProvided(vschema, "ordering", []string{"order_event"}, []string{"oid", "ename"})
+	s.addPKsProvided(vschema, "main", []string{"source_of_ref"}, []string{"id"})
 
 	s.testFile("onecase.json", vw, false)
 }
@@ -666,7 +668,7 @@ func (s *planTestSuite) testFile(filename string, vschema *vschemawrapper.VSchem
 			current := PlanTest{
 				Comment: tcase.Comment,
 				Query:   tcase.Query,
-				SkipE2E: true,
+				SkipE2E: tcase.SkipE2E,
 			}
 			vschema.Version = Gen4
 			out := getPlanOutput(tcase, vschema, render)
diff --git a/go/vt/vtgate/planbuilder/testdata/aggr_cases.json b/go/vt/vtgate/planbuilder/testdata/aggr_cases.json
index 49a03a8f05a..1ecbf3d4ff9 100644
--- a/go/vt/vtgate/planbuilder/testdata/aggr_cases.json
+++ b/go/vt/vtgate/planbuilder/testdata/aggr_cases.json
@@ -6160,6 +6160,44 @@
       ]
     }
   },
+  {
+    "comment": "last_insert_id on aggregation calculated at the vtgate level",
+    "query": "select last_insert_id(count(*)) from user",
+    "plan": {
+      "QueryType": "SELECT",
+      "Original": "select last_insert_id(count(*)) from user",
+      "Instructions": {
+        "OperatorType": "Projection",
+        "Expressions": [
+          "last_insert_id(count(*)) as last_insert_id(count(*))"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "Aggregate",
+            "Variant": "Scalar",
+            "Aggregates": "sum_count_star(0) AS count(*)",
+            "Inputs": [
+              {
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FetchLastInsertID": true,
+                "FieldQuery": "select count(*) from `user` where 1 != 1",
+                "Query": "select count(*) from `user`",
+                "Table": "`user`"
+              }
+            ]
+          }
+        ]
+      },
+      "TablesUsed": [
+        "user.user"
+      ]
+    }
+  },
   {
     "comment": "aggregation on top of aggregation works fine",
     "query": "select distinct count(*) from user, (select distinct count(*) from user) X",
diff --git a/go/vt/vtgate/planbuilder/testdata/dml_cases.json b/go/vt/vtgate/planbuilder/testdata/dml_cases.json
index 8893b4df0c0..95cb14e38f5 100644
--- a/go/vt/vtgate/planbuilder/testdata/dml_cases.json
+++ b/go/vt/vtgate/planbuilder/testdata/dml_cases.json
@@ -2648,6 +2648,30 @@
     },
     "skip_e2e": true
   },
+  {
+    "comment": "insert using last_insert_id with argument (already an e2e test for this plan)",
+    "query": "insert into unsharded values(last_insert_id(789), 2)",
+    "plan": {
+      "QueryType": "INSERT",
+      "Original": "insert into unsharded values(last_insert_id(789), 2)",
+      "Instructions": {
+        "OperatorType": "Insert",
+        "Variant": "Unsharded",
+        "Keyspace": {
+          "Name": "main",
+          "Sharded": false
+        },
+        "TargetTabletType": "PRIMARY",
+        "FetchLastInsertID": true,
+        "Query": "insert into unsharded values (last_insert_id(789), 2)",
+        "TableName": "unsharded"
+      },
+      "TablesUsed": [
+        "main.unsharded"
+      ]
+    },
+    "skip_e2e": true
+  },
   {
     "comment": "update vindex value to null with multiple primary keyspace id",
     "query": "update user set name = null where id in (1, 2, 3)",
diff --git a/go/vt/vtgate/planbuilder/testdata/reference_cases.json b/go/vt/vtgate/planbuilder/testdata/reference_cases.json
index 6aa01355934..1bf893beeef 100644
--- a/go/vt/vtgate/planbuilder/testdata/reference_cases.json
+++ b/go/vt/vtgate/planbuilder/testdata/reference_cases.json
@@ -2,6 +2,7 @@
   {
     "comment": "select from unqualified ambiguous reference routes to reference source",
     "query": "select * from ambiguous_ref_with_source",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select * from ambiguous_ref_with_source",
@@ -24,6 +25,7 @@
   {
     "comment": "join with unqualified ambiguous reference table routes to optimal keyspace",
     "query": "select user.col from user join ambiguous_ref_with_source",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select user.col from user join ambiguous_ref_with_source",
@@ -47,6 +49,7 @@
   {
     "comment": "ambiguous unqualified reference table self-join routes to reference source",
     "query": "select r1.col from ambiguous_ref_with_source r1 join ambiguous_ref_with_source",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select r1.col from ambiguous_ref_with_source r1 join ambiguous_ref_with_source",
@@ -69,6 +72,7 @@
   {
     "comment": "ambiguous unqualified reference table can merge with other opcodes left to right.",
     "query": "select ambiguous_ref_with_source.col from ambiguous_ref_with_source join user",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select ambiguous_ref_with_source.col from ambiguous_ref_with_source join user",
@@ -92,6 +96,7 @@
   {
     "comment": "ambiguous unqualified reference table can merge with other opcodes left to right and vindex value is in the plan",
     "query": "select ambiguous_ref_with_source.col from ambiguous_ref_with_source join (select aa from user where user.id=1) user",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select ambiguous_ref_with_source.col from ambiguous_ref_with_source join (select aa from user where user.id=1) user",
@@ -119,6 +124,7 @@
   {
     "comment": "qualified join to reference table routes to optimal keyspace",
     "query": "select user.col from user join main.ambiguous_ref_with_source",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select user.col from user join main.ambiguous_ref_with_source",
@@ -142,6 +148,7 @@
   {
     "comment": "insert into ambiguous unqualified reference table routes to source",
     "query": "insert into ambiguous_ref_with_source(col) values(1)",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "INSERT",
       "Original": "insert into ambiguous_ref_with_source(col) values(1)",
@@ -164,6 +171,7 @@
   {
     "comment": "Reference tables using left join with a derived table having a limit clause",
     "query": "SELECT u.id FROM ( SELECT a.id, a.u_id FROM user.ref_with_source AS a WHERE a.id IN (3) ORDER BY a.d_at LIMIT 1) as u LEFT JOIN user.ref_with_source AS u0 ON u.u_id = u0.u_uid ORDER BY u.id",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "SELECT u.id FROM ( SELECT a.id, a.u_id FROM user.ref_with_source AS a WHERE a.id IN (3) ORDER BY a.d_at LIMIT 1) as u LEFT JOIN user.ref_with_source AS u0 ON u.u_id = u0.u_uid ORDER BY u.id",
@@ -208,6 +216,7 @@
   {
     "comment": "insert into qualified ambiguous reference table routes to source",
     "query": "insert into user.ambiguous_ref_with_source(col) values(1)",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "INSERT",
       "Original": "insert into user.ambiguous_ref_with_source(col) values(1)",
@@ -230,6 +239,7 @@
   {
     "comment": "update unqualified ambiguous reference table routes to source",
     "query": "update ambiguous_ref_with_source set col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "UPDATE",
       "Original": "update ambiguous_ref_with_source set col = 1",
@@ -252,6 +262,7 @@
   {
     "comment": "update qualified ambiguous reference table route to source",
     "query": "update user.ambiguous_ref_with_source set col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "UPDATE",
       "Original": "update user.ambiguous_ref_with_source set col = 1",
@@ -274,6 +285,7 @@
   {
     "comment": "delete from unqualified ambiguous reference table routes to source",
     "query": "delete from ambiguous_ref_with_source where col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "DELETE",
       "Original": "delete from ambiguous_ref_with_source where col = 1",
@@ -296,6 +308,7 @@
   {
     "comment": "delete from qualified ambiguous reference table route to source",
     "query": "delete from user.ambiguous_ref_with_source where col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "DELETE",
       "Original": "delete from user.ambiguous_ref_with_source where col = 1",
@@ -318,6 +331,7 @@
   {
     "comment": "join with unqualified unambiguous ref with source routes to requested table",
     "query": "select user.col from user join ref_with_source",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select user.col from user join ref_with_source",
@@ -341,6 +355,7 @@
   {
     "comment": "join with unqualified reference optimize routes when source & reference have different names",
     "query": "select user.col from user join source_of_ref",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select user.col from user join source_of_ref",
@@ -364,6 +379,7 @@
   {
     "comment": "join with unqualified reference respects routing rules",
     "query": "select user.col from user join rerouted_ref",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select user.col from user join rerouted_ref",
@@ -387,6 +403,7 @@
   {
     "comment": "join with reference to unqualified source routes to optimal keyspace",
     "query": "select user.col from user join global_ref",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select user.col from user join global_ref",
@@ -410,6 +427,7 @@
   {
     "comment": "insert into qualified reference with unqualified source routes to source",
     "query": "insert into user.global_ref(col) values(1)",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "INSERT",
       "Original": "insert into user.global_ref(col) values(1)",
@@ -432,6 +450,7 @@
   {
     "comment": "delete from reference table with another name - query send to source table",
     "query": "delete from user.ref_with_source where col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "DELETE",
       "Original": "delete from user.ref_with_source where col = 1",
@@ -454,6 +473,7 @@
   {
     "comment": "update from reference table with another name - query send to source table",
     "query": "update user.ref_with_source set x = 4 where col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "UPDATE",
       "Original": "update user.ref_with_source set x = 4 where col = 1",
@@ -476,6 +496,7 @@
   {
     "comment": "insert from reference table with another name - query send to source table",
     "query": "insert into user.ref_with_source(x) values(4)",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "INSERT",
       "Original": "insert into user.ref_with_source(x) values(4)",
@@ -498,6 +519,7 @@
   {
     "comment": "delete from reference table - query send to source table",
     "query": "delete from source_of_ref where col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "DELETE",
       "Original": "delete from source_of_ref where col = 1",
@@ -520,6 +542,7 @@
   {
     "comment": "update from reference table - query send to source table",
     "query": "update source_of_ref set x = 4 where col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "UPDATE",
       "Original": "update source_of_ref set x = 4 where col = 1",
@@ -542,6 +565,7 @@
   {
     "comment": "insert from reference table - query send to source table",
     "query": "insert into source_of_ref(x) values(4)",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "INSERT",
       "Original": "insert into source_of_ref(x) values(4)",
@@ -564,6 +588,7 @@
   {
     "comment": "delete from reference table qualified with unsharded - query send to source table",
     "query": "delete from main.source_of_ref where col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "DELETE",
       "Original": "delete from main.source_of_ref where col = 1",
@@ -586,6 +611,7 @@
   {
     "comment": "update from reference table qualified with unsharded - query send to source table",
     "query": "update main.source_of_ref set x = 4 where col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "UPDATE",
       "Original": "update main.source_of_ref set x = 4 where col = 1",
@@ -608,6 +634,7 @@
   {
     "comment": "insert from reference table qualified with unsharded - query send to source table",
     "query": "insert into main.source_of_ref(x) values(4)",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "INSERT",
       "Original": "insert into main.source_of_ref(x) values(4)",
@@ -630,6 +657,7 @@
   {
     "comment": "delete from reference table with another name - query send to source table",
     "query": "delete from user.ref_with_source where col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "DELETE",
       "Original": "delete from user.ref_with_source where col = 1",
@@ -652,6 +680,7 @@
   {
     "comment": "update from reference table with another name - query send to source table",
     "query": "update user.ref_with_source set x = 4 where col = 1",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "UPDATE",
       "Original": "update user.ref_with_source set x = 4 where col = 1",
@@ -674,6 +703,7 @@
   {
     "comment": "insert from reference table with another name - query send to source table",
     "query": "insert into user.ref_with_source(x) values(4)",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "INSERT",
       "Original": "insert into user.ref_with_source(x) values(4)",
@@ -696,6 +726,7 @@
   {
     "comment": "select with join to reference table in sharded keyspace: should route shard-scoped",
     "query": "select * from user.ref_with_source ref, `user`.`user` u where ref.id = u.ref_id and u.id = 2",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select * from user.ref_with_source ref, `user`.`user` u where ref.id = u.ref_id and u.id = 2",
@@ -723,6 +754,7 @@
   {
     "comment": "select with join to reference table in unsharded keyspace: should route shard-scoped",
     "query": "select * from source_of_ref ref, `user`.`user` u where ref.id = u.ref_id and u.id = 2",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select * from source_of_ref ref, `user`.`user` u where ref.id = u.ref_id and u.id = 2",
@@ -750,6 +782,7 @@
   {
     "comment": "two sharded and two unsharded reference table join - all should be merged into one route",
     "query": "select 1 from user u join user_extra ue on u.id = ue.user_id join main.source_of_ref sr on sr.foo = ue.foo join main.rerouted_ref rr on rr.bar = sr.bar",
+    "skip_e2e": true,
     "plan": {
       "QueryType": "SELECT",
       "Original": "select 1 from user u join user_extra ue on u.id = ue.user_id join main.source_of_ref sr on sr.foo = ue.foo join main.rerouted_ref rr on rr.bar = sr.bar",
@@ -771,5 +804,145 @@
         "user.user_extra"
       ]
     }
+  },
+  {
+    "comment": "update reference table with join on sharded table",
+    "query": "update main.source_of_ref as sr join main.rerouted_ref as rr on sr.id = rr.id inner join user.music as m on sr.col = m.col set sr.tt = 5 where m.user_id = 1",
+    "plan": {
+      "QueryType": "UPDATE",
+      "Original": "update main.source_of_ref as sr join main.rerouted_ref as rr on sr.id = rr.id inner join user.music as m on sr.col = m.col set sr.tt = 5 where m.user_id = 1",
+      "Instructions": {
+        "OperatorType": "DMLWithInput",
+        "TargetTabletType": "PRIMARY",
+        "Offset": [
+          "0:[0]"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "Join",
+            "Variant": "Join",
+            "JoinColumnIndexes": "R:0",
+            "JoinVars": {
+              "m_col": 0
+            },
+            "TableName": "music_rerouted_ref, source_of_ref",
+            "Inputs": [
+              {
+                "OperatorType": "Route",
+                "Variant": "EqualUnique",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select m.col from music as m where 1 != 1",
+                "Query": "select m.col from music as m where m.user_id = 1 lock in share mode",
+                "Table": "music",
+                "Values": [
+                  "1"
+                ],
+                "Vindex": "user_index"
+              },
+              {
+                "OperatorType": "Route",
+                "Variant": "Unsharded",
+                "Keyspace": {
+                  "Name": "main",
+                  "Sharded": false
+                },
+                "FieldQuery": "select sr.id from source_of_ref as sr, rerouted_ref as rr where 1 != 1",
+                "Query": "select sr.id from source_of_ref as sr, rerouted_ref as rr where sr.col = :m_col and sr.id = rr.id lock in share mode",
+                "Table": "rerouted_ref, source_of_ref"
+              }
+            ]
+          },
+          {
+            "OperatorType": "Update",
+            "Variant": "Unsharded",
+            "Keyspace": {
+              "Name": "main",
+              "Sharded": false
+            },
+            "TargetTabletType": "PRIMARY",
+            "Query": "update source_of_ref as sr set sr.tt = 5 where sr.id in ::dml_vals",
+            "Table": "source_of_ref"
+          }
+        ]
+      },
+      "TablesUsed": [
+        "main.rerouted_ref",
+        "main.source_of_ref",
+        "user.music"
+      ]
+    }
+  },
+  {
+    "comment": "delete from reference table with join on sharded table",
+    "query": "delete sr from main.source_of_ref as sr join main.rerouted_ref as rr on sr.id = rr.id inner join user.music as m on sr.col = m.col where m.user_id = 1",
+    "plan": {
+      "QueryType": "DELETE",
+      "Original": "delete sr from main.source_of_ref as sr join main.rerouted_ref as rr on sr.id = rr.id inner join user.music as m on sr.col = m.col where m.user_id = 1",
+      "Instructions": {
+        "OperatorType": "DMLWithInput",
+        "TargetTabletType": "PRIMARY",
+        "Offset": [
+          "0:[0]"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "Join",
+            "Variant": "Join",
+            "JoinColumnIndexes": "R:0",
+            "JoinVars": {
+              "m_col": 0
+            },
+            "TableName": "music_rerouted_ref, source_of_ref",
+            "Inputs": [
+              {
+                "OperatorType": "Route",
+                "Variant": "EqualUnique",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select m.col from music as m where 1 != 1",
+                "Query": "select m.col from music as m where m.user_id = 1",
+                "Table": "music",
+                "Values": [
+                  "1"
+                ],
+                "Vindex": "user_index"
+              },
+              {
+                "OperatorType": "Route",
+                "Variant": "Unsharded",
+                "Keyspace": {
+                  "Name": "main",
+                  "Sharded": false
+                },
+                "FieldQuery": "select sr.id from source_of_ref as sr, rerouted_ref as rr where 1 != 1",
+                "Query": "select sr.id from source_of_ref as sr, rerouted_ref as rr where sr.col = :m_col and sr.id = rr.id",
+                "Table": "rerouted_ref, source_of_ref"
+              }
+            ]
+          },
+          {
+            "OperatorType": "Delete",
+            "Variant": "Unsharded",
+            "Keyspace": {
+              "Name": "main",
+              "Sharded": false
+            },
+            "TargetTabletType": "PRIMARY",
+            "Query": "delete from source_of_ref as sr where sr.id in ::dml_vals",
+            "Table": "source_of_ref"
+          }
+        ]
+      },
+      "TablesUsed": [
+        "main.rerouted_ref",
+        "main.source_of_ref",
+        "user.music"
+      ]
+    }
   }
 ]
diff --git a/go/vt/vtgate/planbuilder/testdata/sampledata/user.sql b/go/vt/vtgate/planbuilder/testdata/sampledata/user.sql
index 044a1ee140d..ff1afd68fca 100644
--- a/go/vt/vtgate/planbuilder/testdata/sampledata/user.sql
+++ b/go/vt/vtgate/planbuilder/testdata/sampledata/user.sql
@@ -11,4 +11,13 @@ INSERT INTO sales_extra(colx, cola, colb, start, end)
 VALUES (13, 'a_3', 'b_3',1000, 1500);
 
 INSERT INTO sales_extra(colx, cola, colb, start, end)
-VALUES (14, 'a_4', 'b_4',1500, 2000);
\ No newline at end of file
+VALUES (14, 'a_4', 'b_4',1500, 2000);
+
+INSERT INTO music (id, user_id, col)
+VALUES (100, 1, 'foo');
+
+INSERT INTO source_of_ref (id, col, tt)
+VALUES (200, 'foo', 2);
+
+INSERT INTO rerouted_ref (id, ref_col, name)
+VALUES (200, 'bar', 'baz');
\ No newline at end of file
diff --git a/go/vt/vtgate/planbuilder/testdata/schemas/main.sql b/go/vt/vtgate/planbuilder/testdata/schemas/main.sql
index fb03b69419b..e615871cf2b 100644
--- a/go/vt/vtgate/planbuilder/testdata/schemas/main.sql
+++ b/go/vt/vtgate/planbuilder/testdata/schemas/main.sql
@@ -1,26 +1,44 @@
-CREATE TABLE `unsharded` (
-    `id`     INT NOT NULL PRIMARY KEY,
-    `col`    VARCHAR(255) DEFAULT NULL,
-    `col1`   VARCHAR(255) DEFAULT NULL,
-    `col2`   VARCHAR(255) DEFAULT NULL,
-    `name`   VARCHAR(255) DEFAULT NULL,
-    `baz`    INT
+CREATE TABLE `unsharded`
+(
+    `id`   INT NOT NULL PRIMARY KEY,
+    `col`  VARCHAR(255) DEFAULT NULL,
+    `col1` VARCHAR(255) DEFAULT NULL,
+    `col2` VARCHAR(255) DEFAULT NULL,
+    `name` VARCHAR(255) DEFAULT NULL,
+    `baz`  INT
 );
 
-CREATE TABLE `unsharded_auto` (
+CREATE TABLE `unsharded_auto`
+(
     `id`   INT NOT NULL PRIMARY KEY,
     `col1` VARCHAR(255) DEFAULT NULL,
     `col2` VARCHAR(255) DEFAULT NULL
 );
 
-CREATE TABLE `unsharded_a` (
+CREATE TABLE `unsharded_a`
+(
     `id`   INT NOT NULL PRIMARY KEY,
     `col`  VARCHAR(255) DEFAULT NULL,
     `name` VARCHAR(255) DEFAULT NULL
 );
 
-CREATE TABLE `unsharded_b` (
+CREATE TABLE `unsharded_b`
+(
     `id`   INT NOT NULL PRIMARY KEY,
     `col`  VARCHAR(255) DEFAULT NULL,
     `name` VARCHAR(255) DEFAULT NULL
+);
+
+CREATE TABLE `source_of_ref`
+(
+    `id`  INT NOT NULL PRIMARY KEY,
+    `col` VARCHAR(255) DEFAULT NULL,
+    `tt`  BIGINT       DEFAULT NULL
+);
+
+CREATE TABLE `rerouted_ref`
+(
+    `id`   INT NOT NULL PRIMARY KEY,
+    `ref_col`  VARCHAR(255) DEFAULT NULL,
+    `name` VARCHAR(255) DEFAULT NULL
 );
\ No newline at end of file
diff --git a/go/vt/vtgate/planbuilder/testdata/schemas/user.sql b/go/vt/vtgate/planbuilder/testdata/schemas/user.sql
index 818d2508069..10c1886a992 100644
--- a/go/vt/vtgate/planbuilder/testdata/schemas/user.sql
+++ b/go/vt/vtgate/planbuilder/testdata/schemas/user.sql
@@ -1,25 +1,25 @@
 CREATE TABLE user
 (
-    id          INT PRIMARY KEY,
-    col         BIGINT,
-    intcol      BIGINT,
-    user_id     INT,
-    id1         INT,
-    id2         INT,
-    id3         INT,
-    m           INT,
-    bar         INT,
-    a           INT,
-    name        VARCHAR(255),
-    col1        VARCHAR(255),
-    col2        VARCHAR(255),
-    costly      VARCHAR(255),
-    predef1     VARCHAR(255),
-    predef2     VARCHAR(255),
-    textcol1    VARCHAR(255),
-    textcol2    VARCHAR(255),
-    someColumn  VARCHAR(255),
-    foo         VARCHAR(255)
+    id         INT PRIMARY KEY,
+    col        BIGINT,
+    intcol     BIGINT,
+    user_id    INT,
+    id1        INT,
+    id2        INT,
+    id3        INT,
+    m          INT,
+    bar        INT,
+    a          INT,
+    name       VARCHAR(255),
+    col1       VARCHAR(255),
+    col2       VARCHAR(255),
+    costly     VARCHAR(255),
+    predef1    VARCHAR(255),
+    predef2    VARCHAR(255),
+    textcol1   VARCHAR(255),
+    textcol2   VARCHAR(255),
+    someColumn VARCHAR(255),
+    foo        VARCHAR(255)
 );
 
 CREATE TABLE user_metadata
@@ -34,15 +34,23 @@ CREATE TABLE user_metadata
 
 CREATE TABLE music
 (
-    user_id INT,
-    id      INT,
-    col1    VARCHAR(255),
-    col2    VARCHAR(255),
-    genre   VARCHAR(255),
+    user_id   INT,
+    id        INT,
+    col       VARCHAR(255),
+    col1      VARCHAR(255),
+    col2      VARCHAR(255),
+    genre     VARCHAR(255),
     componist VARCHAR(255),
     PRIMARY KEY (user_id)
 );
 
+CREATE TABLE name_user_vdx
+(
+    name        INT,
+    keyspace_id VARBINARY(10),
+    primary key (name)
+);
+
 CREATE TABLE samecolvin
 (
     col VARCHAR(255),
@@ -118,69 +126,63 @@ CREATE TABLE authoritative
 
 CREATE TABLE colb_colc_map
 (
-    colb  INT PRIMARY KEY,
-    colc  INT,
+    colb        INT PRIMARY KEY,
+    colc        INT,
     keyspace_id VARCHAR(255)
 );
 
 CREATE TABLE seq
 (
-    id         INT,
-    next_id    BIGINT,
-    cache      BIGINT,
+    id      INT,
+    next_id BIGINT,
+    cache   BIGINT,
     PRIMARY KEY (id)
 ) COMMENT 'vitess_sequence';
 
 CREATE TABLE user_extra
 (
-    id           INT,
-    user_id      INT,
-    extra_id     INT,
-    col          INT,
-    m2           INT,
+    id       INT,
+    user_id  INT,
+    extra_id INT,
+    col      INT,
+    m2       INT,
     PRIMARY KEY (id, extra_id)
 );
 
 CREATE TABLE name_user_map
 (
-    name          VARCHAR(255),
-    keyspace_id   VARCHAR(255)
-);
-
-CREATE TABLE name_user_vdx
-(
-    name          VARCHAR(255),
-    keyspace_id   VARCHAR(255)
+    name        VARCHAR(255),
+    keyspace_id VARCHAR(255)
 );
 
 CREATE TABLE costly_map
 (
-    costly        VARCHAR(255),
-    keyspace_id   VARCHAR(255)
+    costly      VARCHAR(255),
+    keyspace_id VARCHAR(255)
 );
 
 CREATE TABLE unq_binary_idx
 (
-    id       INT PRIMARY KEY,
-    col1     INT
+    id   INT PRIMARY KEY,
+    col1 INT
 );
 
 CREATE TABLE sales
 (
-    oid       INT PRIMARY KEY,
-    col1      VARCHAR(255)
+    oid  INT PRIMARY KEY,
+    col1 VARCHAR(255)
 );
 
 CREATE TABLE sales_extra
 (
-    colx      INT PRIMARY KEY,
-    cola      VARCHAR(255),
-    colb      VARCHAR(255),
-    start     INT,
-    end       INT
+    colx  INT PRIMARY KEY,
+    cola  VARCHAR(255),
+    colb  VARCHAR(255),
+    start INT,
+    end   INT
 );
 
 CREATE TABLE ref
 (
-    col    INT PRIMARY KEY
+    col INT PRIMARY KEY
 );
\ No newline at end of file
diff --git a/go/vt/vtgate/planbuilder/testdata/set_cases.json b/go/vt/vtgate/planbuilder/testdata/set_cases.json
index 58cb2fffa75..02c5603a03c 100644
--- a/go/vt/vtgate/planbuilder/testdata/set_cases.json
+++ b/go/vt/vtgate/planbuilder/testdata/set_cases.json
@@ -605,5 +605,28 @@
         ]
       }
     }
+  },
+  {
+    "comment": "set last_insert_id with agrument to user defined variable",
+    "query": "set @foo = last_insert_id(1)",
+    "plan": {
+      "QueryType": "SET",
+      "Original": "set @foo = last_insert_id(1)",
+      "Instructions": {
+        "OperatorType": "Set",
+        "Ops": [
+          {
+            "Type": "UserDefinedVariable",
+            "Name": "foo",
+            "Expr": "last_insert_id(1)"
+          }
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "SingleRow"
+          }
+        ]
+      }
+    }
   }
 ]
diff --git a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.json b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.json
index 9241cec595c..55329586b0e 100644
--- a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.json
+++ b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.json
@@ -342,7 +342,12 @@
   {
     "comment": "reference table delete with join",
     "query": "delete r from user u join ref_with_source r on u.col = r.col",
-    "plan": "VT12001: unsupported: DELETE on reference table with join"
+    "plan": "VT12001: unsupported: DML on reference table with join"
+  },
+  {
+    "comment": "reference table update with join",
+    "query": "update user u join ref_with_source r on u.col = r.col set r.col = 5",
+    "plan": "VT12001: unsupported: DML on reference table with join"
   },
   {
     "comment": "group_concat unsupported when needs full evaluation at vtgate with more than 1 column",
diff --git a/go/vt/vtgate/semantics/analyzer.go b/go/vt/vtgate/semantics/analyzer.go
index 988932f4414..62cdc019ddf 100644
--- a/go/vt/vtgate/semantics/analyzer.go
+++ b/go/vt/vtgate/semantics/analyzer.go
@@ -174,7 +174,7 @@ func (a *analyzer) newSemTable(
 		Direct:                    a.binder.direct,
 		ExprTypes:                 a.typer.m,
 		Tables:                    a.tables.Tables,
-		Targets:                   a.binder.targets,
+		DMLTargets:                a.binder.targets,
 		NotSingleRouteErr:         a.notSingleRouteErr,
 		NotUnshardedErr:           a.unshardedErr,
 		Warning:                   a.warning,
diff --git a/go/vt/vtgate/semantics/semantic_table.go b/go/vt/vtgate/semantics/semantic_table.go
index 492259427c5..30a41ba5f12 100644
--- a/go/vt/vtgate/semantics/semantic_table.go
+++ b/go/vt/vtgate/semantics/semantic_table.go
@@ -130,8 +130,8 @@ type (
 		// It doesn't recurse inside derived tables to find the original dependencies.
 		Direct ExprDependencies
 
-		// Targets contains the TableSet of each table getting modified by the update/delete statement.
-		Targets TableSet
+		// DMLTargets contains the TableSet of each table getting modified by the update/delete statement.
+		DMLTargets TableSet
 
 		// ColumnEqualities is used for transitive closures (e.g., if a == b and b == c, then a == c).
 		ColumnEqualities map[columnName][]sqlparser.Expr
@@ -203,7 +203,7 @@ func (st *SemTable) CopyDependencies(from, to sqlparser.Expr) {
 
 // GetChildForeignKeysForTargets gets the child foreign keys as a list for all the target tables.
 func (st *SemTable) GetChildForeignKeysForTargets() (fks []vindexes.ChildFKInfo) {
-	for _, ts := range st.Targets.Constituents() {
+	for _, ts := range st.DMLTargets.Constituents() {
 		fks = append(fks, st.childForeignKeysInvolved[ts]...)
 	}
 	return fks
@@ -211,7 +211,7 @@ func (st *SemTable) GetChildForeignKeysForTargets() (fks []vindexes.ChildFKInfo)
 
 // GetChildForeignKeysForTableSet gets the child foreign keys as a listfor the TableSet.
 func (st *SemTable) GetChildForeignKeysForTableSet(target TableSet) (fks []vindexes.ChildFKInfo) {
-	for _, ts := range st.Targets.Constituents() {
+	for _, ts := range st.DMLTargets.Constituents() {
 		if target.IsSolvedBy(ts) {
 			fks = append(fks, st.childForeignKeysInvolved[ts]...)
 		}
@@ -239,7 +239,7 @@ func (st *SemTable) GetChildForeignKeysList() []vindexes.ChildFKInfo {
 
 // GetParentForeignKeysForTargets gets the parent foreign keys as a list for all the target tables.
 func (st *SemTable) GetParentForeignKeysForTargets() (fks []vindexes.ParentFKInfo) {
-	for _, ts := range st.Targets.Constituents() {
+	for _, ts := range st.DMLTargets.Constituents() {
 		fks = append(fks, st.parentForeignKeysInvolved[ts]...)
 	}
 	return fks
@@ -247,7 +247,7 @@ func (st *SemTable) GetParentForeignKeysForTargets() (fks []vindexes.ParentFKInf
 
 // GetParentForeignKeysForTableSet gets the parent foreign keys as a list for the TableSet.
 func (st *SemTable) GetParentForeignKeysForTableSet(target TableSet) (fks []vindexes.ParentFKInfo) {
-	for _, ts := range st.Targets.Constituents() {
+	for _, ts := range st.DMLTargets.Constituents() {
 		if target.IsSolvedBy(ts) {
 			fks = append(fks, st.parentForeignKeysInvolved[ts]...)
 		}
@@ -971,7 +971,7 @@ func (st *SemTable) UpdateChildFKExpr(origUpdExpr *sqlparser.UpdateExpr, newExpr
 
 // GetTargetTableSetForTableName returns the TableSet for the given table name from the target tables.
 func (st *SemTable) GetTargetTableSetForTableName(name sqlparser.TableName) (TableSet, error) {
-	for _, target := range st.Targets.Constituents() {
+	for _, target := range st.DMLTargets.Constituents() {
 		tbl, err := st.Tables[target.TableOffset()].Name()
 		if err != nil {
 			return "", err
diff --git a/java/example/pom.xml b/java/example/pom.xml
index fa3220f51bd..fabab27dd77 100644
--- a/java/example/pom.xml
+++ b/java/example/pom.xml
@@ -32,7 +32,7 @@
     <dependency>
       <groupId>mysql</groupId>
       <artifactId>mysql-connector-java</artifactId>
-      <version>8.0.28</version>
+      <version>8.0.33</version>
       <optional>false</optional>
     </dependency>
 
diff --git a/java/grpc-client/src/test/resources/ca.config b/java/grpc-client/src/test/resources/ca.config
index e0955f28ccf..c5758831e06 100644
--- a/java/grpc-client/src/test/resources/ca.config
+++ b/java/grpc-client/src/test/resources/ca.config
@@ -2,6 +2,7 @@
  default_bits           = 1024
  default_keyfile        = keyfile.pem
  distinguished_name     = req_distinguished_name
+ x509_extensions        = v3_ca
  attributes             = req_attributes
  prompt                 = no
  output_password        = mypass
@@ -15,3 +16,5 @@
  emailAddress           = test@email.address
 [ req_attributes ]
  challengePassword      = A challenge password
+[ v3_ca ]
+ basicConstraints       = CA:TRUE
diff --git a/java/pom.xml b/java/pom.xml
index 6742258a6b8..e68bbd5574e 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -72,8 +72,8 @@
     <netty.handler.version>4.1.110.Final</netty.handler.version>
     <tcnative.boring.ssl.version>2.0.65.Final</tcnative.boring.ssl.version>
 
-    <protobuf.java.version>4.28.3</protobuf.java.version>
-    <protobuf.protoc.version>3.24.3</protobuf.protoc.version>
+    <protobuf.java.version>3.25.5</protobuf.java.version>
+    <protobuf.protoc.version>3.25.5</protobuf.protoc.version>
     <checkstyle.plugin.version>3.0.0</checkstyle.plugin.version>
     <log4j2.version>2.24.1</log4j2.version>
   </properties>
diff --git a/test/config.json b/test/config.json
index 1fe200b8dbb..dfc0910f10e 100644
--- a/test/config.json
+++ b/test/config.json
@@ -604,7 +604,7 @@
 			"Manual": false,
 			"Shard": "vtgate_queries",
 			"RetryMax": 2,
-			"Tags": ["upgrade_downgrade_query_serving_queries"]
+			"Tags": ["upgrade_downgrade_query_serving_queries_2"]
 		},
 		"vtgate_queries_subquery": {
 			"File": "unused.go",
@@ -613,7 +613,7 @@
 			"Manual": false,
 			"Shard": "vtgate_queries",
 			"RetryMax": 2,
-			"Tags": ["upgrade_downgrade_query_serving_queries"]
+			"Tags": ["upgrade_downgrade_query_serving_queries_2"]
 		},
 		"vtgate_queries_union": {
 			"File": "unused.go",
@@ -622,7 +622,7 @@
 			"Manual": false,
 			"Shard": "vtgate_queries",
 			"RetryMax": 2,
-			"Tags": ["upgrade_downgrade_query_serving_queries"]
+			"Tags": ["upgrade_downgrade_query_serving_queries_2"]
 		},
 		"vtgate_queries_insert": {
 			"File": "unused.go",
@@ -631,7 +631,7 @@
 			"Manual": false,
 			"Shard": "vtgate_queries",
 			"RetryMax": 2,
-			"Tags": ["upgrade_downgrade_query_serving_queries"]
+			"Tags": ["upgrade_downgrade_query_serving_queries_2"]
 		},
 		"vtgate_queries_vexplain": {
 			"File": "unused.go",
@@ -640,7 +640,7 @@
 			"Manual": false,
 			"Shard": "vtgate_queries",
 			"RetryMax": 2,
-			"Tags": ["upgrade_downgrade_query_serving_queries"]
+			"Tags": ["upgrade_downgrade_query_serving_queries_2"]
 		},
 		"vtgate_queries_reference": {
 			"File": "unused.go",
@@ -649,7 +649,7 @@
 			"Manual": false,
 			"Shard": "vtgate_queries",
 			"RetryMax": 1,
-			"Tags": ["upgrade_downgrade_query_serving_queries"]
+			"Tags": ["upgrade_downgrade_query_serving_queries_2"]
 		},
 		"vtgate_queries_random": {
 			"File": "unused.go",
@@ -658,7 +658,7 @@
 			"Manual": false,
 			"Shard": "vtgate_queries",
 			"RetryMax": 1,
-			"Tags": ["upgrade_downgrade_query_serving_queries"]
+			"Tags": ["upgrade_downgrade_query_serving_queries_2"]
 		},
 		"vtgate_kill": {
 			"File": "unused.go",
@@ -667,7 +667,7 @@
 			"Manual": false,
 			"Shard": "vtgate_queries",
 			"RetryMax": 1,
-			"Tags": ["upgrade_downgrade_query_serving_queries"]
+			"Tags": ["upgrade_downgrade_query_serving_queries_2"]
 		},
 		"vtgate_concurrentdml": {
 			"File": "unused.go",
diff --git a/test/local_example.sh b/test/local_example.sh
index 391e75a9224..27f512a34eb 100755
--- a/test/local_example.sh
+++ b/test/local_example.sh
@@ -98,5 +98,11 @@ mysql --table < ../common/select_customer80-_data.sql
 
 ./306_down_shard_0.sh
 
-./401_teardown.sh
+./401_backup.sh
+
+./402_list_backup.sh
+
+./403_restore_from_backup.sh
+
+./501_teardown.sh