Repository: apache/cassandra-gocql-driver

Issue:
CASSGO-41 Deadlock in refreshDebouncer when reconnection fails
### What version of Cassandra are you using?
astra-classic

### What version of Gocql are you using?
v1.6.0


### What version of Go are you using?
1.21

### What did you do?
Connection errors, I think due to overload, lead to frequent reconnection attempts and failures

### What did you expect to see?
Should retry until connection succeeds


### What did you see instead?
Deadlock

```
498297 goroutine 1324045437 [chan send, 113 minutes]:
498298 github.com/gocql/gocql.(*refreshDebouncer).stop(0xc0b826a7c0)
498299         /go/pkg/mod/github.com/gocql/gocql@v1.6.0/host_source.go:848 +0x8c
498300 github.com/gocql/gocql.(*Session).Close(0xc03efb0c00)
498301         /go/pkg/mod/github.com/gocql/gocql@v1.6.0/session.go:494 +0x105
498302 github.com/gocql/gocql.NewSession({{0xc24be58930, 0x3, 0x3}, {0x2ef55cf, 0x5}, 0x4, 0x12a05f200, 0x12a05f200, 0x0, 0x755a, ...})
498303         /go/pkg/mod/github.com/gocql/gocql@v1.6.0/session.go:180 +0x98d
498304 github.com/gocql/gocql.(*ClusterConfig).CreateSession(...)
498305         /go/pkg/mod/github.com/gocql/gocql@v1.6.0/cluster.go:289
```

It looks like a race condition between `(*refreshDebouncer).stop()` and `(*refreshDebouncer).flusher()`
1. `stop()` acquires `d.mu` and sets `d.stopped` to true
2. `flusher()` exits the `select` at the top of the loop and blocks on acquiring `d.mu`
3. `stop()` releases `d.mu` and tries to write to `d.quit`
4. `flusher()` acquires `d.mu` and returns because `d.stopped` is true
5. `stop()` is deadlocked because `d.quit` is unbuffered and the reader has stopped

Discussion:
@joao-r-reis: I can work on a fix for this but it's likely that it will only get merged when this driver is donated to the ASF (see https://github.com/gocql/gocql/issues/1749#issuecomment-2104443885 and https://github.com/gocql/gocql/issues/1751 )@joao-r-reis: By the way @kevinkyyro , this won't really help with gocql not reconnecting, it will just stop the panic from happening. I believe your issue with the reconnection is related to gocql-astra adding a single contact point which causes gocql to fail after 1 connection attempt (see: https://github.com/datastax/gocql-astra/pull/24 )

In any case I've opened a PR to fix this deadlock issue: #1767 

Solution patch:
diff --git a/.github/actions/setup-environment/action.yml b/.github/actions/setup-environment/action.yml
deleted file mode 100644
index 3ed57aa..0000000
--- a/.github/actions/setup-environment/action.yml
+++ /dev/null
@@ -1,50 +0,0 @@
-name: Setup environment
-description: Setup environment for integration tests execution
-runs:
-  using: "composite"
-  steps:
-    - name: Set up cache for SDKMAN
-      uses: actions/cache@v3
-      with:
-        path: ~/.sdkman
-        key: ${{ runner.os }}-sdkman
-
-    - name: Set up cache for PIP
-      uses: actions/cache@v3
-      with:
-        path: ~/.cache/pip
-        key: ${{ runner.os }}-pip-${{ hashFiles('requirements.txt') }}
-        restore-keys: |
-          ${{ runner.os }}-pip-
-
-    - name: Install Java
-      shell: bash
-      run: |
-        echo "Installing SDKMAN..."
-        curl -s "https://get.sdkman.io" | bash
-        source "$HOME/.sdkman/bin/sdkman-init.sh"
-        echo "sdkman_auto_answer=true" >> ~/.sdkman/etc/config
-
-        echo "Installing Java versions..."
-        sdk install java 11.0.24-zulu
-        sdk install java 17.0.12-zulu
-
-        sdk default java 11.0.24-zulu
-        sdk use java 11.0.24-zulu
-
-        echo "JAVA11_HOME=$JAVA_HOME_11_X64" >> $GITHUB_ENV
-        echo "JAVA17_HOME=$JAVA_HOME_17_X64" >> $GITHUB_ENV
-        echo "JAVA_HOME=$JAVA_HOME_11_X64" >> $GITHUB_ENV
-        echo "PATH=$PATH" >> $GITHUB_ENV
-
-    - name: Install CCM
-      shell: bash
-      run: |
-        echo "Creating Python virtual environment..."
-        VENV_DIR="$HOME/venv"
-        python3 -m venv $VENV_DIR
-        source $VENV_DIR/bin/activate
-        pip install --upgrade pip setuptools
-
-        echo "Installing CCM..."
-        pip install "git+https://github.com/riptano/ccm.git@${CCM_VERSION}"
\ No newline at end of file
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index db4228d..d9ae88a 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -7,9 +7,6 @@ on:
   pull_request:
     types: [ opened, synchronize, reopened ]
 
-env:
-  CCM_VERSION: "4621dfee5ad73956b831091a8b863d100d25c610"
-
 jobs:
   build:
     name: Unit tests
@@ -22,9 +19,11 @@ jobs:
       - uses: actions/setup-go@v4
         with:
           go-version: ${{ matrix.go }}
-      - run: go vet ./...
+      - name: Run linting
+        run: make check
       - name: Run unit tests
-        run: go test -v -tags unit -race ./...
+        run: make test-unit
+
   integration-cassandra:
     timeout-minutes: 15
     needs:
@@ -45,82 +44,51 @@ jobs:
             compressor: "snappy"
     steps:
       - uses: actions/checkout@v2
-      - uses: actions/setup-go@v2
+
+      - uses: actions/setup-go@v5
         with:
           go-version: ${{ matrix.go }}
-      - uses: actions/cache@v4
-        id: gomod-cache
-        with:
-          path: ~/go/pkg/mod
-          key: ${{ runner.os }}-go-${{ hashFiles('go.mod') }}
-          restore-keys: |
-            ${{ runner.os }}-go-
-      - name: Setup environment
-        uses: ./.github/actions/setup-environment
-      - name: Start cassandra nodes
-        run: |
-          source ~/venv/bin/activate
-          VERSION=${{ matrix.cassandra_version }}
-          keypath="$(pwd)/testdata/pki"
-          conf=(
-              "client_encryption_options.enabled: true"
-              "client_encryption_options.keystore: $keypath/.keystore"
-              "client_encryption_options.keystore_password: cassandra"
-              "client_encryption_options.require_client_auth: true"
-              "client_encryption_options.truststore: $keypath/.truststore"
-              "client_encryption_options.truststore_password: cassandra"
-              "concurrent_reads: 2"
-              "concurrent_writes: 2"
-              "write_request_timeout_in_ms: 5000"
-              "read_request_timeout_in_ms: 5000"
-          )
-
-          if [[ $VERSION == 3.*.* ]]; then
-              conf+=(
-                "rpc_server_type: sync"
-                "rpc_min_threads: 2"
-                "rpc_max_threads: 2"
-                "enable_user_defined_functions: true"
-                "enable_materialized_views: true"
-              )
-          elif [[ $VERSION == 4.0.* ]]; then
-              conf+=(
-                "enable_user_defined_functions: true"
-                "enable_materialized_views: true"
-              )
-          else
-              conf+=(
-                "user_defined_functions_enabled: true"
-                "materialized_views_enabled: true"
-              )
-          fi
 
-          ccm remove test || true
-
-          ccm create test -v $VERSION -n 3 -d --vnodes --jvm_arg="-Xmx256m -XX:NewSize=100m"
-          ccm updateconf "${conf[@]}"
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.10'
+          cache: 'pip'
+          cache-dependency-path: Makefile
 
-          export JVM_EXTRA_OPTS=" -Dcassandra.test.fail_writes_ks=test -Dcassandra.custom_query_handler_class=org.apache.cassandra.cql3.CustomPayloadMirroringQueryHandler"
+      - name: Set up cache for SDKMAN
+        uses: actions/cache@v3
+        with:
+          path: ~/.sdkman
+          key: ${{ runner.os }}-sdkman
 
-          ccm start --wait-for-binary-proto --verbose
-          ccm status
-          ccm node1 nodetool status
+      - name: Set up cache for CCM
+        uses: actions/cache@v4
+        with:
+          path: ~/.ccm/repository
+          key: ${{ runner.os }}-ccm-${{ matrix.cassandra_version }}
 
-          args="-gocql.timeout=60s -runssl -proto=${{ matrix.proto_version }} -rf=3 -clusterSize=3 -autowait=2000ms -compressor=${{ matrix.compressor }} -gocql.cversion=$VERSION -cluster=$(ccm liveset) ./..."
+      - name: Start cassandra nodes
+        env:
+          CASSANDRA_VERSION: ${{ matrix.cassandra_version }}
+        run: |
+          make cassandra-start
+          sleep 30s
 
-          echo "args=$args" >> $GITHUB_ENV
-          echo "JVM_EXTRA_OPTS=$JVM_EXTRA_OPTS" >> $GITHUB_ENV
       - name: Integration tests
+        env:
+          CASSANDRA_VERSION: ${{ matrix.cassandra_version }}
+          TEST_CQL_PROTOCOL: ${{ matrix.proto_version }}
+          TEST_COMPRESSOR: ${{ matrix.compressor }}
+          TEST_INTEGRATION_TAGS: ${{ matrix.tags }}
         run: |
-          source ~/venv/bin/activate
-          export JVM_EXTRA_OPTS="${{env.JVM_EXTRA_OPTS}}"
-          go test -v -tags "${{ matrix.tags }} gocql_debug" -timeout=5m -race ${{ env.args }}
+          make test-integration
+
       - name: 'Save ccm logs'
         if: 'failure()'
         uses: actions/upload-artifact@v4
         with:
           name: ccm-cluster-cassandra-${{ matrix.cassandra_version }}-go-${{ matrix.go }}-tag-${{ matrix.tags }}-proto-version-${{ matrix.proto_version }}-compressor-${{ matrix.compressor }}
-          path: /home/runner/.ccm/test
+          path: /home/runner/.ccm/gocql_integration_test
           retention-days: 5
   integration-auth-cassandra:
     timeout-minutes: 15
@@ -144,69 +112,37 @@ jobs:
       - uses: actions/setup-go@v4
         with:
           go-version: ${{ matrix.go }}
-      - name: Setup environment
-        uses: ./.github/actions/setup-environment
-      - name: Start cassandra nodes
-        run: |
-          source ~/venv/bin/activate
-          VERSION=${{ matrix.cassandra_version }}
-          keypath="$(pwd)/testdata/pki"
-          conf=(
-              "client_encryption_options.enabled: true"
-              "client_encryption_options.keystore: $keypath/.keystore"
-              "client_encryption_options.keystore_password: cassandra"
-              "client_encryption_options.require_client_auth: true"
-              "client_encryption_options.truststore: $keypath/.truststore"
-              "client_encryption_options.truststore_password: cassandra"
-              "concurrent_reads: 2"
-              "concurrent_writes: 2"
-              "write_request_timeout_in_ms: 5000"
-              "read_request_timeout_in_ms: 5000"
-              "authenticator: PasswordAuthenticator"
-              "authorizer: CassandraAuthorizer"
-              "enable_user_defined_functions: true"
-          )
-
-          if [[ $VERSION == 3.*.* ]]; then
-              conf+=(
-                "rpc_server_type: sync"
-                "rpc_min_threads: 2"
-                "rpc_max_threads: 2"
-                "enable_user_defined_functions: true"
-                "enable_materialized_views: true"
-              )
-          elif [[ $VERSION == 4.0.* ]]; then
-              conf+=(
-                "enable_user_defined_functions: true"
-                "enable_materialized_views: true"
-              )
-          else
-              conf+=(
-                "user_defined_functions_enabled: true"
-                "materialized_views_enabled: true"
-              )
-          fi
-
-          ccm remove test || true
-
-          ccm create test -v $VERSION -n 1 -d --vnodes --jvm_arg="-Xmx256m -XX:NewSize=100m"
-          ccm updateconf "${conf[@]}"
-
-          rm -rf $HOME/.ccm/test/node1/data/system_auth
 
-          export JVM_EXTRA_OPTS=" -Dcassandra.test.fail_writes_ks=test -Dcassandra.custom_query_handler_class=org.apache.cassandra.cql3.CustomPayloadMirroringQueryHandler"
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.10'
+          cache: 'pip'
+          cache-dependency-path: Makefile
 
-          ccm start --wait-for-binary-proto --verbose
-          ccm status
-          ccm node1 nodetool status
+      - name: Set up cache for SDKMAN
+        uses: actions/cache@v3
+        with:
+          path: ~/.sdkman
+          key: ${{ runner.os }}-sdkman
 
-          args="-gocql.timeout=60s -runssl -proto=${{ matrix.proto_version }} -rf=3 -clusterSize=1 -autowait=2000ms -compressor=${{ matrix.compressor }} -gocql.cversion=$VERSION -cluster=$(ccm liveset) ./..."
+      - name: Set up cache for CCM
+        uses: actions/cache@v4
+        with:
+          path: ~/.ccm/repository
+          key: ${{ runner.os }}-ccm-${{ matrix.cassandra_version }}
 
-          echo "args=$args" >> $GITHUB_ENV
-          echo "JVM_EXTRA_OPTS=$JVM_EXTRA_OPTS" >> $GITHUB_ENV
+      - name: Start cassandra nodes
+        env:
+          CASSANDRA_VERSION: ${{ matrix.cassandra_version }}
+        run: |
+          make cassandra-start
           sleep 30s
+
       - name: Integration tests
+        env:
+          CASSANDRA_VERSION: ${{ matrix.cassandra_version }}
+          TEST_CQL_PROTOCOL: ${{ matrix.proto_version }}
+          TEST_COMPRESSOR: ${{ matrix.compressor }}
+          TEST_INTEGRATION_TAGS: ${{ matrix.tags }}
         run: |
-          source ~/venv/bin/activate
-          export JVM_EXTRA_OPTS="${{env.JVM_EXTRA_OPTS}}"
-          go test -v -run=TestAuthentication -tags "${{ matrix.tags }} gocql_debug" -timeout=15s -runauth ${{ env.args }}
+          make test-integration-auth
diff --git a/.golangci.yml b/.golangci.yml
new file mode 100644
index 0000000..f0ff7d3
--- /dev/null
+++ b/.golangci.yml
@@ -0,0 +1,33 @@
+version: "2"
+
+issues:
+  max-same-issues: 50
+
+formatters:
+  enable:
+    - goimports
+
+  settings:
+    goimports:
+      local-prefixes:
+        - github.com/gocql/gocql
+        - github.com/apache/cassandra-gocql-driver
+    golines:
+      max-len: 120
+
+linters:
+  default: none
+  enable:
+    - nolintlint
+    - govet
+  settings:
+    govet:
+      enable-all: true
+      disable:
+        - shadow
+        - fieldalignment
+
+    nolintlint:
+      allow-no-explanation: [ golines ]
+      require-explanation: true
+      require-specific: true
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ffaa662..6c6c7cc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,65 +5,53 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
+## [2.0.0] - Unreleased
 
-### Added
+### Removed
 
-- Support vector type [CASSGO-11](https://issues.apache.org/jira/browse/CASSGO-11)
+- Drop support for old CQL protocol versions: 1 and 2 (CASSGO-75)
 
-- Allow SERIAL and LOCAL_SERIAL on SELECT statements [CASSGO-26](https://issues.apache.org/jira/browse/CASSGO-26)
+### Added
 
+- Support vector type (CASSGO-11)
+- Allow SERIAL and LOCAL_SERIAL on SELECT statements (CASSGO-26)
 - Support of sending queries to the specific node with Query.SetHostID() (CASSGO-4)
-
-- Support for Native Protocol 5. Following protocol changes exposed new API 
-  Query.SetKeyspace(), Query.WithNowInSeconds(), Batch.SetKeyspace(), Batch.WithNowInSeconds() (CASSGO-1)
+- Support for Native Protocol 5 (CASSGO-1)
 
 ### Changed
 
 - Move lz4 compressor to lz4 package within the gocql module (CASSGO-32)
-
 - Don't restrict server authenticator unless PasswordAuthentictor.AllowedAuthenticators is provided (CASSGO-19)
-
 - Cleanup of deprecated elements (CASSGO-12)
-
 - Remove global NewBatch function (CASSGO-15)
-
 - Detailed description for NumConns (CASSGO-3)
-
 - Change Batch API to be consistent with Query() (CASSGO-7)
-
 - Added Cassandra 4.0 table options support(CASSGO-13)
-
 - Remove deprecated global logger (CASSGO-24)
-
 - Bumped actions/upload-artifact and actions/cache versions to v4 in CI workflow (CASSGO-48)
-
 - Keep nil slices in MapScan (CASSGO-44)
-
 - Improve error messages for marshalling (CASSGO-38)
-
 - Remove HostPoolHostPolicy from gocql package (CASSGO-21)
-
 - Standardized spelling of datacenter (CASSGO-35)
-
 - Refactor HostInfo creation and ConnectAddress() method (CASSGO-45)
-
 - gocql.Compressor interface changes to follow append-like design. Bumped Go version to 1.19 (CASSGO-1)
-
 - Refactoring hostpool package test and Expose HostInfo creation (CASSGO-59)
 
+- Move "execute batch" methods to Batch type (CASSGO-57)
+
+- Make `Session` immutable by removing setters and associated mutex (CASSGO-23)
+
 ### Fixed
 - Cassandra version unmarshal fix (CASSGO-49)
-
 - Retry policy now takes into account query idempotency (CASSGO-27)
-
 - Don't return error to caller with RetryType Ignore (CASSGO-28)
 - The marshalBigInt return 8 bytes slice in all cases except for big.Int,
   which returns a variable length slice, but should be 8 bytes slice as well (CASSGO-2)
-
 - Skip metadata only if the prepared result includes metadata (CASSGO-40)
-
 - Don't panic in MapExecuteBatchCAS if no `[applied]` column is returned (CASSGO-42)
+- Fix deadlock in refresh debouncer stop (CASSGO-41)
+
+- Endless query execution fix (CASSGO-50)
 
 ## [1.7.0] - 2024-09-23
 
diff --git a/Makefile b/Makefile
new file mode 100644
index 0000000..56ed015
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,164 @@
+SHELL := bash
+MAKEFILE_PATH := $(abspath $(dir $(abspath $(lastword $(MAKEFILE_LIST)))))
+KEY_PATH = ${MAKEFILE_PATH}/testdata/pki
+
+CASSANDRA_VERSION ?= 4.1.6
+TEST_CQL_PROTOCOL ?= 4
+TEST_COMPRESSOR ?= no-compression
+TEST_INTEGRATION_TAGS ?= integration
+
+CCM_VERSION ?= 39b8222b31a6c7afe8fe845d16981088a5a735ad
+GOLANGCI_VERSION = v2.1.6
+JVM_EXTRA_OPTS ?= -Dcassandra.test.fail_writes_ks=test -Dcassandra.custom_query_handler_class=org.apache.cassandra.cql3.CustomPayloadMirroringQueryHandler
+ifeq (${CCM_CONFIG_DIR},)
+	CCM_CONFIG_DIR = ~/.ccm
+endif
+CCM_CONFIG_DIR := $(shell readlink --canonicalize ${CCM_CONFIG_DIR})
+
+CASSANDRA_CONFIG ?= "client_encryption_options.enabled: true" \
+"client_encryption_options.keystore: ${KEY_PATH}/.keystore" \
+"client_encryption_options.keystore_password: cassandra" \
+"client_encryption_options.require_client_auth: true" \
+"client_encryption_options.truststore: ${KEY_PATH}/.truststore" \
+"client_encryption_options.truststore_password: cassandra" \
+"concurrent_reads: 2" \
+"concurrent_writes: 2" \
+"write_request_timeout_in_ms: 5000" \
+"read_request_timeout_in_ms: 5000"
+
+ifeq ($(shell echo "${CASSANDRA_VERSION}" | grep -oP "3\.[0-9]+\.[0-9]+" ),${CASSANDRA_VERSION})
+	CASSANDRA_CONFIG += "rpc_server_type: sync" \
+"rpc_min_threads: 2" \
+"rpc_max_threads: 2" \
+"enable_user_defined_functions: true" \
+"enable_materialized_views: true"
+else ifeq ($(shell echo "${CASSANDRA_VERSION}" | grep -oP "4\.0\.[0-9]+" ),${CASSANDRA_VERSION})
+	CASSANDRA_CONFIG +=	"enable_user_defined_functions: true" \
+"enable_materialized_views: true"
+else
+	CASSANDRA_CONFIG += "user_defined_functions_enabled: true" \
+"materialized_views_enabled: true"
+endif
+
+ifneq (${JAVA11_HOME},)
+	export JAVA11_HOME
+endif
+
+ifneq (${JAVA17_HOME},)
+	export JAVA17_HOME
+endif
+
+ifneq (${JAVA_HOME},)
+	export JAVA_HOME
+endif
+
+export JVM_EXTRA_OPTS
+
+.prepare-cassandra-cluster: .prepare-ccm .prepare-java
+	@if [ -d ${CCM_CONFIG_DIR}/gocql_integration_test ] && ccm switch gocql_integration_test 2>/dev/null 1>&2 && ccm status | grep UP 2>/dev/null 1>&2; then \
+		echo "Cassandra cluster is already started"; \
+  	else \
+		echo "Start cassandra ${CASSANDRA_VERSION} cluster"; \
+		ccm stop gocql_integration_test 2>/dev/null 1>&2 || true; \
+		ccm remove gocql_integration_test 2>/dev/null 1>&2 || true; \
+		rm -rf ${CCM_CONFIG_DIR}/gocql_integration_test || true; \
+		ccm create gocql_integration_test -v ${CASSANDRA_VERSION} -n 3 -d --vnodes --jvm_arg="-Xmx256m -XX:NewSize=100m" && \
+		ccm updateconf ${CASSANDRA_CONFIG} && \
+		ccm start --wait-for-binary-proto --verbose && \
+		ccm status && \
+		ccm node1 nodetool status; \
+	fi
+
+cassandra-start: .prepare-ccm .prepare-java
+	@echo "Start cassandra ${CASSANDRA_VERSION} cluster"
+	@ccm stop gocql_integration_test 2>/dev/null 1>&2 || true
+	@ccm remove gocql_integration_test 2>/dev/null 1>&2 || true
+	@rm -rf ${CCM_CONFIG_DIR}/gocql_integration_test || true
+	ccm create gocql_integration_test -v ${CASSANDRA_VERSION} -n 3 -d --vnodes --jvm_arg="-Xmx256m -XX:NewSize=100m"
+	@ccm updateconf ${CASSANDRA_CONFIG}
+	@ccm start --wait-for-binary-proto --verbose
+	@ccm status
+	@ccm node1 nodetool status
+
+cassandra-stop: .prepare-ccm
+	@echo "Stop cassandra cluster"
+	@ccm stop gocql_integration_test 2>/dev/null 1>&2 || true
+
+cassandra-remove: .prepare-ccm
+	@echo "Remove cassandra cluster"
+	@ccm remove gocql_integration_test 2>/dev/null 1>&2 || true
+	@rm -rf ${CCM_CONFIG_DIR}/gocql_integration_test || true
+
+test-integration: .prepare-cassandra-cluster
+	@echo "Run integration tests for proto ${TEST_CQL_PROTOCOL} on cassandra ${CASSANDRA_VERSION}"
+	go test -v ${TEST_OPTS} -tags "${TEST_INTEGRATION_TAGS} gocql_debug" -timeout=5m -proto=${TEST_CQL_PROTOCOL} -gocql.timeout=60s -runssl -rf=3 -clusterSize=3 -autowait=2000ms -compressor=${TEST_COMPRESSOR} -gocql.cversion=${CASSANDRA_VERSION} -cluster=$$(ccm liveset) ./...
+
+test-integration-auth: .prepare-cassandra-cluster
+	@echo "Run auth integration tests for proto ${TEST_CQL_PROTOCOL} on cassandra ${CASSANDRA_VERSION}"
+	go test -v -run=TestAuthentication -tags "${TEST_INTEGRATION_TAGS} gocql_debug" -timeout=5m -proto=${TEST_CQL_PROTOCOL} -gocql.timeout=60s -runssl -runauth -rf=3 -clusterSize=3 -autowait=2000ms -compressor=${TEST_COMPRESSOR} -gocql.cversion=${CASSANDRA_VERSION} -cluster=$$(ccm liveset) ./...
+
+test-unit:
+	@echo "Run unit tests"
+	@go clean -testcache
+	go test -tags unit -timeout=5m -race ./...
+
+check: .prepare-golangci
+	@echo "Build"
+	@go build -tags all .
+	@echo "Check linting"
+	@golangci-lint run
+
+fix: .prepare-golangci
+	@echo "Fix linting"
+	golangci-lint run --fix
+
+.prepare-java:
+ifeq ($(shell if [ -f ~/.sdkman/bin/sdkman-init.sh ]; then echo "installed"; else echo "not-installed"; fi), not-installed)
+	@$(MAKE) install-java
+else
+	@exit 0
+endif
+
+install-java:
+	@echo "Installing SDKMAN..."
+	@curl -s "https://get.sdkman.io" | bash
+	@echo "sdkman_auto_answer=true" >> ~/.sdkman/etc/config
+	@( \
+		source ~/.sdkman/bin/sdkman-init.sh; \
+		export PATH=${PATH}:~/.sdkman/bin; \
+		echo "Installing Java versions..."; \
+		sdk install java 11.0.24-zulu; \
+		sdk install java 17.0.12-zulu; \
+		sdk default java 11.0.24-zulu; \
+		sdk use java 11.0.24-zulu; \
+		if [[ -n "${GITHUB_ENV}" ]]; then \
+			echo "JAVA11_HOME=$$JAVA_HOME_11_X64" >> ${GITHUB_ENV}; \
+    		echo "JAVA17_HOME=$$JAVA_HOME_17_X64" >> ${GITHUB_ENV}; \
+    		echo "JAVA_HOME=$$JAVA_HOME_11_X64" >> ${GITHUB_ENV}; \
+    		echo "$$PATH" > ${GITHUB_PATH}; \
+		fi; \
+	)
+
+.prepare-ccm:
+	@ccm --help 2>/dev/null 1>&2; \
+	if [[ $$? -le 127 ]] && grep ${CCM_VERSION} ${CCM_CONFIG_DIR}/ccm-version 2>/dev/null 1>&2; then \
+	  	echo "CCM ${CCM_VERSION} is already installed"; \
+	else \
+	  	echo "Install CCM ${CCM_VERSION}"; \
+      	pip install "git+https://github.com/riptano/ccm.git@${CCM_VERSION}"; \
+      	mkdir ${CCM_CONFIG_DIR} 2>/dev/null 1>&2 || true; \
+      	echo ${CCM_VERSION} > ${CCM_CONFIG_DIR}/ccm-version; \
+	fi
+
+install-ccm:
+	@echo "Install CCM ${CCM_VERSION}"
+	@pip install "git+https://github.com/riptano/ccm.git@${CCM_VERSION}"
+	@mkdir ${CCM_CONFIG_DIR} 2>/dev/null 1>&2 || true
+	@echo ${CCM_VERSION} > ${CCM_CONFIG_DIR}/ccm-version
+
+.prepare-golangci:
+	@if ! golangci-lint --version 2>/dev/null | grep ${GOLANGCI_VERSION} >/dev/null; then \
+  		echo "Installing golangci-ling ${GOLANGCI_VERSION}"; \
+		go install github.com/golangci/golangci-lint/v2/cmd/golangci-lint@${GOLANGCI_VERSION}; \
+  	fi
+
diff --git a/cluster.go b/cluster.go
index 2f96191..06a271e 100644
--- a/cluster.go
+++ b/cluster.go
@@ -263,6 +263,17 @@ type ClusterConfig struct {
 	// If not specified, defaults to the gocql.defaultLogger.
 	Logger StdLogger
 
+	// Tracer will be used for all queries. Alternatively it can be set of on a
+	// per query basis.
+	// default: nil
+	Tracer Tracer
+
+	// NextPagePrefetch sets the default threshold for pre-fetching new pages. If
+	// there are only p*pageSize rows remaining, the next page will be requested
+	// automatically. This value can also be changed on a per-query basis.
+	// default: 0.25.
+	NextPagePrefetch float64
+
 	// internal config for testing
 	disableControlConn bool
 }
@@ -298,6 +309,7 @@ func NewCluster(hosts ...string) *ClusterConfig {
 		ConvictionPolicy:       &SimpleConvictionPolicy{},
 		ReconnectionPolicy:     &ConstantReconnectionPolicy{MaxRetries: 3, Interval: 1 * time.Second},
 		WriteCoalesceWaitTime:  200 * time.Microsecond,
+		NextPagePrefetch:       0.25,
 	}
 	return cfg
 }
diff --git a/conn.go b/conn.go
index 114f159..b7a3ed2 100644
--- a/conn.go
+++ b/conn.go
@@ -179,7 +179,7 @@ type Conn struct {
 	frameObserver  FrameHeaderObserver
 	streamObserver StreamObserver
 
-	headerBuf [maxFrameHeaderSize]byte
+	headerBuf [frameHeadSize]byte
 
 	streams *streams.IDGenerator
 	mu      sync.Mutex
@@ -784,12 +784,11 @@ func (c *Conn) recvSegment(ctx context.Context) error {
 		return err
 	}
 
-	const frameHeaderLength = 9
-	buf := bytes.NewBuffer(make([]byte, 0, head.length+frameHeaderLength))
+	buf := bytes.NewBuffer(make([]byte, 0, head.length+frameHeadSize))
 	buf.Write(frame)
 
 	// Computing how many bytes of message left to read
-	bytesToRead := head.length - len(frame) + frameHeaderLength
+	bytesToRead := head.length - len(frame) + frameHeadSize
 
 	err = c.recvPartialFrames(buf, bytesToRead)
 	if err != nil {
@@ -1738,10 +1737,6 @@ func (c *Conn) UseKeyspace(keyspace string) error {
 }
 
 func (c *Conn) executeBatch(ctx context.Context, batch *Batch) *Iter {
-	if c.version == protoVersion1 {
-		return &Iter{err: ErrUnsupported}
-	}
-
 	n := len(batch.Entries)
 	req := &writeBatchFrame{
 		typ:                   batch.Type,
@@ -1900,6 +1895,7 @@ func (c *Conn) awaitSchemaAgreement(ctx context.Context) (err error) {
 
 	var versions map[string]struct{}
 	var schemaVersion string
+	var rows []map[string]interface{}
 
 	endDeadline := time.Now().Add(c.session.cfg.MaxWaitSchemaAgreement)
 
@@ -1908,17 +1904,18 @@ func (c *Conn) awaitSchemaAgreement(ctx context.Context) (err error) {
 
 		versions = make(map[string]struct{})
 
-		rows, err := iter.SliceMap()
+		rows, err = iter.SliceMap()
 		if err != nil {
 			goto cont
 		}
 
 		for _, row := range rows {
-			h, err := NewHostInfo(c.host.ConnectAddress(), c.session.cfg.Port)
+			var host *HostInfo
+			host, err = NewHostInfo(c.host.ConnectAddress(), c.session.cfg.Port)
 			if err != nil {
 				goto cont
 			}
-			host, err := c.session.hostInfoFromMap(row, h)
+			host, err = c.session.hostInfoFromMap(row, host)
 			if err != nil {
 				goto cont
 			}
diff --git a/doc.go b/doc.go
index 109a850..f6e7b28 100644
--- a/doc.go
+++ b/doc.go
@@ -271,7 +271,7 @@
 //
 // # Paging
 //
-// The driver supports paging of results with automatic prefetch, see ClusterConfig.PageSize, Session.SetPrefetch,
+// The driver supports paging of results with automatic prefetch, see ClusterConfig.PageSize,
 // Query.PageSize, and Query.Prefetch.
 //
 // It is also possible to control the paging manually with Query.PageState (this disables automatic prefetch).
@@ -311,7 +311,7 @@
 //
 // The CQL protocol supports sending batches of DML statements (INSERT/UPDATE/DELETE) and so does gocql.
 // Use Session.Batch to create a new batch and then fill-in details of individual queries.
-// Then execute the batch with Session.ExecuteBatch.
+// Then execute the batch with Batch.Exec.
 //
 // Logged batches ensure atomicity, either all or none of the operations in the batch will succeed, but they have
 // overhead to ensure this property.
@@ -329,8 +329,8 @@
 // It is also possible to pass entire BEGIN BATCH .. APPLY BATCH statement to Query.Exec.
 // There are differences how those are executed.
 // BEGIN BATCH statement passed to Query.Exec is prepared as a whole in a single statement.
-// Session.ExecuteBatch prepares individual statements in the batch.
-// If you have variable-length batches using the same statement, using Session.ExecuteBatch is more efficient.
+// Batch.Exec prepares individual statements in the batch.
+// If you have variable-length batches using the same statement, using Batch.Exec is more efficient.
 //
 // See Example_batch for an example.
 //
@@ -340,9 +340,9 @@
 // INSERT/UPDATE .. IF statement) and reading its result. See example for Query.MapScanCAS.
 //
 // Multiple-statement lightweight transactions can be executed as a logged batch that contains at least one conditional
-// statement. All the conditions must return true for the batch to be applied. You can use Session.ExecuteBatchCAS and
-// Session.MapExecuteBatchCAS when executing the batch to learn about the result of the LWT. See example for
-// Session.MapExecuteBatchCAS.
+// statement. All the conditions must return true for the batch to be applied. You can use Batch.ExecCAS and
+// Batch.MapExecCAS when executing the batch to learn about the result of the LWT. See example for
+// Batch.MapExecCAS.
 //
 // # Retries and speculative execution
 //
diff --git a/frame.go b/frame.go
index 926c54b..f2f6c7c 100644
--- a/frame.go
+++ b/frame.go
@@ -307,7 +307,7 @@ var (
 	ErrFrameTooBig = errors.New("frame length is bigger than the maximum allowed")
 )
 
-const maxFrameHeaderSize = 9
+const frameHeadSize = 9
 
 func readInt(p []byte) int32 {
 	return int32(p[0])<<24 | int32(p[1])<<16 | int32(p[2])<<8 | int32(p[3])
@@ -364,9 +364,8 @@ type FrameHeaderObserver interface {
 type framer struct {
 	proto byte
 	// flags are for outgoing flags, enabling compression and tracing etc
-	flags    byte
-	compres  Compressor
-	headSize int
+	flags   byte
+	compres Compressor
 	// if this frame was read then the header will be here
 	header *frameHeader
 
@@ -398,15 +397,9 @@ func newFramer(compressor Compressor, version byte) *framer {
 
 	version &= protoVersionMask
 
-	headSize := 8
-	if version > protoVersion2 {
-		headSize = 9
-	}
-
 	f.compres = compressor
 	f.proto = version
 	f.flags = flags
-	f.headSize = headSize
 
 	f.header = nil
 	f.traceID = nil
@@ -426,43 +419,27 @@ func readHeader(r io.Reader, p []byte) (head frameHeader, err error) {
 
 	version := p[0] & protoVersionMask
 
-	if version < protoVersion1 || version > protoVersion5 {
+	if version < protoVersion3 || version > protoVersion5 {
 		return frameHeader{}, fmt.Errorf("gocql: unsupported protocol response version: %d", version)
 	}
 
-	headSize := 9
-	if version < protoVersion3 {
-		headSize = 8
-	}
-
-	_, err = io.ReadFull(r, p[1:headSize])
+	_, err = io.ReadFull(r, p[1:frameHeadSize])
 	if err != nil {
 		return frameHeader{}, err
 	}
 
-	p = p[:headSize]
+	p = p[:frameHeadSize]
 
 	head.version = protoVersion(p[0])
 	head.flags = p[1]
 
-	if version > protoVersion2 {
-		if len(p) != 9 {
-			return frameHeader{}, fmt.Errorf("not enough bytes to read header require 9 got: %d", len(p))
-		}
-
-		head.stream = int(int16(p[2])<<8 | int16(p[3]))
-		head.op = frameOp(p[4])
-		head.length = int(readInt(p[5:]))
-	} else {
-		if len(p) != 8 {
-			return frameHeader{}, fmt.Errorf("not enough bytes to read header require 8 got: %d", len(p))
-		}
-
-		head.stream = int(int8(p[2]))
-		head.op = frameOp(p[3])
-		head.length = int(readInt(p[4:]))
+	if len(p) != 9 {
+		return frameHeader{}, fmt.Errorf("not enough bytes to read header require 9 got: %d", len(p))
 	}
 
+	head.stream = int(int16(p[2])<<8 | int16(p[3]))
+	head.op = frameOp(p[4])
+	head.length = int(readInt(p[5:]))
 	return head, nil
 }
 
@@ -700,43 +677,18 @@ func (f *framer) readErrorMap() (errMap ErrorMap) {
 }
 
 func (f *framer) writeHeader(flags byte, op frameOp, stream int) {
-	f.buf = f.buf[:0]
-	f.buf = append(f.buf,
-		f.proto,
-		flags,
-	)
-
-	if f.proto > protoVersion2 {
-		f.buf = append(f.buf,
-			byte(stream>>8),
-			byte(stream),
-		)
-	} else {
-		f.buf = append(f.buf,
-			byte(stream),
-		)
-	}
-
-	// pad out length
-	f.buf = append(f.buf,
-		byte(op),
-		0,
-		0,
-		0,
-		0,
+	f.buf = append(f.buf[:0],
+		f.proto, flags, byte(stream>>8), byte(stream),
+		// pad out length
+		byte(op), 0, 0, 0, 0,
 	)
 }
 
 func (f *framer) setLength(length int) {
-	p := 4
-	if f.proto > protoVersion2 {
-		p = 5
-	}
-
-	f.buf[p+0] = byte(length >> 24)
-	f.buf[p+1] = byte(length >> 16)
-	f.buf[p+2] = byte(length >> 8)
-	f.buf[p+3] = byte(length)
+	f.buf[5] = byte(length >> 24)
+	f.buf[6] = byte(length >> 16)
+	f.buf[7] = byte(length >> 8)
+	f.buf[8] = byte(length)
 }
 
 func (f *framer) finish() error {
@@ -752,14 +704,14 @@ func (f *framer) finish() error {
 		}
 
 		// TODO: only compress frames which are big enough
-		compressed, err := f.compres.AppendCompressedWithLength(nil, f.buf[f.headSize:])
+		compressed, err := f.compres.AppendCompressedWithLength(nil, f.buf[frameHeadSize:])
 		if err != nil {
 			return err
 		}
 
-		f.buf = append(f.buf[:f.headSize], compressed...)
+		f.buf = append(f.buf[:frameHeadSize], compressed...)
 	}
-	length := len(f.buf) - f.headSize
+	length := len(f.buf) - frameHeadSize
 	f.setLength(length)
 
 	return nil
@@ -771,7 +723,14 @@ func (f *framer) writeTo(w io.Writer) error {
 }
 
 func (f *framer) readTrace() {
-	f.traceID = f.readUUID().Bytes()
+	if len(f.buf) < 16 {
+		panic(fmt.Errorf("not enough bytes in buffer to read trace uuid require 16 got: %d", len(f.buf)))
+	}
+	if len(f.traceID) != 16 {
+		f.traceID = make([]byte, 16)
+	}
+	copy(f.traceID, f.buf[:16])
+	f.buf = f.buf[16:]
 }
 
 type readyFrame struct {
@@ -1185,11 +1144,6 @@ func (f *framer) parseResultPrepared() frame {
 	}
 
 	frame.reqMeta = f.parsePreparedMetadata()
-
-	if f.proto < protoVersion2 {
-		return frame
-	}
-
 	frame.respMeta = f.parseResultMetadata()
 
 	return frame
@@ -1245,87 +1199,65 @@ type schemaChangeAggregate struct {
 }
 
 func (f *framer) parseResultSchemaChange() frame {
-	if f.proto <= protoVersion2 {
-		change := f.readString()
-		keyspace := f.readString()
-		table := f.readString()
-
-		if table != "" {
-			return &schemaChangeTable{
-				frameHeader: *f.header,
-				change:      change,
-				keyspace:    keyspace,
-				object:      table,
-			}
-		} else {
-			return &schemaChangeKeyspace{
-				frameHeader: *f.header,
-				change:      change,
-				keyspace:    keyspace,
-			}
+	change := f.readString()
+	target := f.readString()
+
+	// TODO: could just use a separate type for each target
+	switch target {
+	case "KEYSPACE":
+		frame := &schemaChangeKeyspace{
+			frameHeader: *f.header,
+			change:      change,
 		}
-	} else {
-		change := f.readString()
-		target := f.readString()
-
-		// TODO: could just use a separate type for each target
-		switch target {
-		case "KEYSPACE":
-			frame := &schemaChangeKeyspace{
-				frameHeader: *f.header,
-				change:      change,
-			}
 
-			frame.keyspace = f.readString()
+		frame.keyspace = f.readString()
 
-			return frame
-		case "TABLE":
-			frame := &schemaChangeTable{
-				frameHeader: *f.header,
-				change:      change,
-			}
+		return frame
+	case "TABLE":
+		frame := &schemaChangeTable{
+			frameHeader: *f.header,
+			change:      change,
+		}
 
-			frame.keyspace = f.readString()
-			frame.object = f.readString()
+		frame.keyspace = f.readString()
+		frame.object = f.readString()
 
-			return frame
-		case "TYPE":
-			frame := &schemaChangeType{
-				frameHeader: *f.header,
-				change:      change,
-			}
+		return frame
+	case "TYPE":
+		frame := &schemaChangeType{
+			frameHeader: *f.header,
+			change:      change,
+		}
 
-			frame.keyspace = f.readString()
-			frame.object = f.readString()
+		frame.keyspace = f.readString()
+		frame.object = f.readString()
 
-			return frame
-		case "FUNCTION":
-			frame := &schemaChangeFunction{
-				frameHeader: *f.header,
-				change:      change,
-			}
+		return frame
+	case "FUNCTION":
+		frame := &schemaChangeFunction{
+			frameHeader: *f.header,
+			change:      change,
+		}
 
-			frame.keyspace = f.readString()
-			frame.name = f.readString()
-			frame.args = f.readStringList()
+		frame.keyspace = f.readString()
+		frame.name = f.readString()
+		frame.args = f.readStringList()
 
-			return frame
-		case "AGGREGATE":
-			frame := &schemaChangeAggregate{
-				frameHeader: *f.header,
-				change:      change,
-			}
+		return frame
+	case "AGGREGATE":
+		frame := &schemaChangeAggregate{
+			frameHeader: *f.header,
+			change:      change,
+		}
 
-			frame.keyspace = f.readString()
-			frame.name = f.readString()
-			frame.args = f.readStringList()
+		frame.keyspace = f.readString()
+		frame.name = f.readString()
+		frame.args = f.readStringList()
 
-			return frame
-		default:
-			panic(fmt.Errorf("gocql: unknown SCHEMA_CHANGE target: %q change: %q", target, change))
-		}
+		return frame
+	default:
+		panic(fmt.Errorf("gocql: unknown SCHEMA_CHANGE target: %q change: %q", target, change))
 	}
-
 }
 
 type authenticateFrame struct {
@@ -1479,10 +1411,6 @@ func (q queryParams) String() string {
 func (f *framer) writeQueryParams(opts *queryParams) {
 	f.writeConsistency(opts.consistency)
 
-	if f.proto == protoVersion1 {
-		return
-	}
-
 	var flags uint32
 	names := false
 
@@ -1502,16 +1430,13 @@ func (f *framer) writeQueryParams(opts *queryParams) {
 		flags |= flagWithSerialConsistency
 	}
 
-	// protoV3 specific things
-	if f.proto > protoVersion2 {
-		if opts.defaultTimestamp {
-			flags |= flagDefaultTimestamp
-		}
+	if opts.defaultTimestamp {
+		flags |= flagDefaultTimestamp
+	}
 
-		if len(opts.values) > 0 && opts.values[0].name != "" {
-			flags |= flagWithNameValues
-			names = true
-		}
+	if len(opts.values) > 0 && opts.values[0].name != "" {
+		flags |= flagWithNameValues
+		names = true
 	}
 
 	if opts.keyspace != "" {
@@ -1561,7 +1486,7 @@ func (f *framer) writeQueryParams(opts *queryParams) {
 		f.writeConsistency(opts.serialConsistency)
 	}
 
-	if f.proto > protoVersion2 && opts.defaultTimestamp {
+	if opts.defaultTimestamp {
 		// timestamp in microseconds
 		var ts int64
 		if opts.defaultTimestampValue != 0 {
@@ -1650,20 +1575,7 @@ func (f *framer) writeExecuteFrame(streamID int, preparedID, resultMetadataID []
 		f.writeShortBytes(resultMetadataID)
 	}
 
-	if f.proto > protoVersion1 {
-		f.writeQueryParams(params)
-	} else {
-		n := len(params.values)
-		f.writeShort(uint16(n))
-		for i := 0; i < n; i++ {
-			if params.values[i].isUnset {
-				f.writeUnset()
-			} else {
-				f.writeBytes(params.values[i].value)
-			}
-		}
-		f.writeConsistency(params.consistency)
-	}
+	f.writeQueryParams(params)
 
 	return f.finish()
 }
@@ -1724,7 +1636,7 @@ func (f *framer) writeBatchFrame(streamID int, w *writeBatchFrame, customPayload
 		f.writeShort(uint16(len(b.values)))
 		for j := range b.values {
 			col := b.values[j]
-			if f.proto > protoVersion2 && col.name != "" {
+			if col.name != "" {
 				// TODO: move this check into the caller and set a flag on writeBatchFrame
 				// to indicate using named values
 				if f.proto <= protoVersion5 {
@@ -1743,13 +1655,12 @@ func (f *framer) writeBatchFrame(streamID int, w *writeBatchFrame, customPayload
 
 	f.writeConsistency(w.consistency)
 
-	if f.proto > protoVersion2 {
-		if w.serialConsistency > 0 {
-			flags |= flagWithSerialConsistency
-		}
-		if w.defaultTimestamp {
-			flags |= flagDefaultTimestamp
-		}
+	if w.serialConsistency > 0 {
+		flags |= flagWithSerialConsistency
+	}
+
+	if w.defaultTimestamp {
+		flags |= flagDefaultTimestamp
 	}
 
 	if w.keyspace != "" {
@@ -1876,17 +1787,6 @@ func (f *framer) readLongString() (s string) {
 	return
 }
 
-func (f *framer) readUUID() *UUID {
-	if len(f.buf) < 16 {
-		panic(fmt.Errorf("not enough bytes in buffer to read uuid require %d got: %d", 16, len(f.buf)))
-	}
-
-	// TODO: how to handle this error, if it is a uuid, then sureley, problems?
-	u, _ := UUIDFromBytes(f.buf[:16])
-	f.buf = f.buf[16:]
-	return &u
-}
-
 func (f *framer) readStringList() []string {
 	size := f.readShort()
 
diff --git a/host_source.go b/host_source.go
index adcf1a7..396bcfe 100644
--- a/host_source.go
+++ b/host_source.go
@@ -807,6 +807,7 @@ type refreshDebouncer struct {
 	timer        *time.Timer
 	refreshNowCh chan struct{}
 	quit         chan struct{}
+	done         chan struct{}
 	refreshFn    func() error
 }
 
@@ -816,6 +817,7 @@ func newRefreshDebouncer(interval time.Duration, refreshFn func() error) *refres
 		broadcaster:  nil,
 		refreshNowCh: make(chan struct{}, 1),
 		quit:         make(chan struct{}),
+		done:         make(chan struct{}),
 		interval:     interval,
 		timer:        time.NewTimer(interval),
 		refreshFn:    refreshFn,
@@ -851,6 +853,7 @@ func (d *refreshDebouncer) refreshNow() <-chan error {
 }
 
 func (d *refreshDebouncer) flusher() {
+	defer close(d.done)
 	for {
 		select {
 		case <-d.refreshNowCh:
@@ -899,8 +902,8 @@ func (d *refreshDebouncer) stop() {
 	}
 	d.stopped = true
 	d.mu.Unlock()
-	d.quit <- struct{}{} // sync with flusher
 	close(d.quit)
+	<-d.done
 }
 
 // broadcasts an error to multiple channels (listeners)
diff --git a/lz4/lz4.go b/lz4/lz4.go
index c836a09..f97cbd3 100644
--- a/lz4/lz4.go
+++ b/lz4/lz4.go
@@ -27,6 +27,7 @@ package lz4
 import (
 	"encoding/binary"
 	"fmt"
+
 	"github.com/pierrec/lz4/v4"
 )
 
diff --git a/marshal.go b/marshal.go
index 368c7f1..65ab656 100644
--- a/marshal.go
+++ b/marshal.go
@@ -116,7 +116,6 @@ func Marshal(info TypeInfo, value interface{}) ([]byte, error) {
 	if info.Version() < protoVersion1 {
 		panic("protocol version not set")
 	}
-
 	if valueRef := reflect.ValueOf(value); valueRef.Kind() == reflect.Ptr {
 		if valueRef.IsNil() {
 			return nil, nil
@@ -1565,25 +1564,15 @@ func encVint(v int64) []byte {
 	return buf
 }
 
-func writeCollectionSize(info CollectionType, n int, buf *bytes.Buffer) error {
-	if info.proto > protoVersion2 {
-		if n > math.MaxInt32 {
-			return marshalErrorf("marshal: collection too large")
-		}
-
-		buf.WriteByte(byte(n >> 24))
-		buf.WriteByte(byte(n >> 16))
-		buf.WriteByte(byte(n >> 8))
-		buf.WriteByte(byte(n))
-	} else {
-		if n > math.MaxUint16 {
-			return marshalErrorf("marshal: collection too large")
-		}
-
-		buf.WriteByte(byte(n >> 8))
-		buf.WriteByte(byte(n))
+func writeCollectionSize(n int, buf *bytes.Buffer) error {
+	if n > math.MaxInt32 {
+		return marshalErrorf("marshal: collection too large")
 	}
 
+	buf.WriteByte(byte(n >> 24))
+	buf.WriteByte(byte(n >> 16))
+	buf.WriteByte(byte(n >> 8))
+	buf.WriteByte(byte(n))
 	return nil
 }
 
@@ -1611,7 +1600,7 @@ func marshalList(info TypeInfo, value interface{}) ([]byte, error) {
 		buf := &bytes.Buffer{}
 		n := rv.Len()
 
-		if err := writeCollectionSize(listInfo, n, buf); err != nil {
+		if err := writeCollectionSize(n, buf); err != nil {
 			return nil, err
 		}
 
@@ -1622,10 +1611,10 @@ func marshalList(info TypeInfo, value interface{}) ([]byte, error) {
 			}
 			itemLen := len(item)
 			// Set the value to null for supported protocols
-			if item == nil && listInfo.proto > protoVersion2 {
+			if item == nil {
 				itemLen = -1
 			}
-			if err := writeCollectionSize(listInfo, itemLen, buf); err != nil {
+			if err := writeCollectionSize(itemLen, buf); err != nil {
 				return nil, err
 			}
 			buf.Write(item)
@@ -1645,20 +1634,12 @@ func marshalList(info TypeInfo, value interface{}) ([]byte, error) {
 	return nil, marshalErrorf("can not marshal %T into %s. Accepted types: slice, array, map[]struct.", value, info)
 }
 
-func readCollectionSize(info CollectionType, data []byte) (size, read int, err error) {
-	if info.proto > protoVersion2 {
-		if len(data) < 4 {
-			return 0, 0, unmarshalErrorf("unmarshal list: unexpected eof")
-		}
-		size = int(int32(data[0])<<24 | int32(data[1])<<16 | int32(data[2])<<8 | int32(data[3]))
-		read = 4
-	} else {
-		if len(data) < 2 {
-			return 0, 0, unmarshalErrorf("unmarshal list: unexpected eof")
-		}
-		size = int(data[0])<<8 | int(data[1])
-		read = 2
+func readCollectionSize(data []byte) (size, read int, err error) {
+	if len(data) < 4 {
+		return 0, 0, unmarshalErrorf("unmarshal list: unexpected eof")
 	}
+	size = int(int32(data[0])<<24 | int32(data[1])<<16 | int32(data[2])<<8 | int32(data[3]))
+	read = 4
 	return
 }
 
@@ -1688,7 +1669,7 @@ func unmarshalList(info TypeInfo, data []byte, value interface{}) error {
 			rv.Set(reflect.Zero(t))
 			return nil
 		}
-		n, p, err := readCollectionSize(listInfo, data)
+		n, p, err := readCollectionSize(data)
 		if err != nil {
 			return err
 		}
@@ -1701,7 +1682,7 @@ func unmarshalList(info TypeInfo, data []byte, value interface{}) error {
 			rv.Set(reflect.MakeSlice(t, n, n))
 		}
 		for i := 0; i < n; i++ {
-			m, p, err := readCollectionSize(listInfo, data)
+			m, p, err := readCollectionSize(data)
 			if err != nil {
 				return err
 			}
@@ -1909,7 +1890,7 @@ func marshalMap(info TypeInfo, value interface{}) ([]byte, error) {
 	buf := &bytes.Buffer{}
 	n := rv.Len()
 
-	if err := writeCollectionSize(mapInfo, n, buf); err != nil {
+	if err := writeCollectionSize(n, buf); err != nil {
 		return nil, err
 	}
 
@@ -1921,10 +1902,10 @@ func marshalMap(info TypeInfo, value interface{}) ([]byte, error) {
 		}
 		itemLen := len(item)
 		// Set the key to null for supported protocols
-		if item == nil && mapInfo.proto > protoVersion2 {
+		if item == nil {
 			itemLen = -1
 		}
-		if err := writeCollectionSize(mapInfo, itemLen, buf); err != nil {
+		if err := writeCollectionSize(itemLen, buf); err != nil {
 			return nil, err
 		}
 		buf.Write(item)
@@ -1935,10 +1916,10 @@ func marshalMap(info TypeInfo, value interface{}) ([]byte, error) {
 		}
 		itemLen = len(item)
 		// Set the value to null for supported protocols
-		if item == nil && mapInfo.proto > protoVersion2 {
+		if item == nil {
 			itemLen = -1
 		}
-		if err := writeCollectionSize(mapInfo, itemLen, buf); err != nil {
+		if err := writeCollectionSize(itemLen, buf); err != nil {
 			return nil, err
 		}
 		buf.Write(item)
@@ -1965,7 +1946,7 @@ func unmarshalMap(info TypeInfo, data []byte, value interface{}) error {
 		rv.Set(reflect.Zero(t))
 		return nil
 	}
-	n, p, err := readCollectionSize(mapInfo, data)
+	n, p, err := readCollectionSize(data)
 	if err != nil {
 		return err
 	}
@@ -1975,7 +1956,7 @@ func unmarshalMap(info TypeInfo, data []byte, value interface{}) error {
 	rv.Set(reflect.MakeMapWithSize(t, n))
 	data = data[p:]
 	for i := 0; i < n; i++ {
-		m, p, err := readCollectionSize(mapInfo, data)
+		m, p, err := readCollectionSize(data)
 		if err != nil {
 			return err
 		}
@@ -1994,7 +1975,7 @@ func unmarshalMap(info TypeInfo, data []byte, value interface{}) error {
 			return err
 		}
 
-		m, p, err = readCollectionSize(mapInfo, data)
+		m, p, err = readCollectionSize(data)
 		if err != nil {
 			return err
 		}
@@ -2603,7 +2584,7 @@ func unmarshalUDT(info TypeInfo, data []byte, value interface{}) error {
 		f, ok := fields[e.Name]
 		if !ok {
 			f = k.FieldByName(e.Name)
-			if f == emptyValue {
+			if f == emptyValue { //nolint:govet // there is no other way to compare with empty value
 				// skip fields which exist in the UDT but not in
 				// the struct passed in
 				continue
diff --git a/metadata.go b/metadata.go
index c7f8e4b..c3e6ad3 100644
--- a/metadata.go
+++ b/metadata.go
@@ -32,7 +32,6 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"strconv"
 	"strings"
 	"sync"
 )
@@ -388,122 +387,7 @@ func compileMetadata(
 		table.OrderedColumns = append(table.OrderedColumns, col.Name)
 	}
 
-	if protoVersion == protoVersion1 {
-		compileV1Metadata(tables, protoVersion, logger)
-	} else {
-		compileV2Metadata(tables, protoVersion, logger)
-	}
-}
-
-// Compiles derived information from TableMetadata which have had
-// ColumnMetadata added already. V1 protocol does not return as much
-// column metadata as V2+ (because V1 doesn't support the "type" column in the
-// system.schema_columns table) so determining PartitionKey and ClusterColumns
-// is more complex.
-func compileV1Metadata(tables []TableMetadata, protoVer int, logger StdLogger) {
-	for i := range tables {
-		table := &tables[i]
-
-		// decode the key validator
-		keyValidatorParsed := parseType(table.KeyValidator, byte(protoVer), logger)
-		// decode the comparator
-		comparatorParsed := parseType(table.Comparator, byte(protoVer), logger)
-
-		// the partition key length is the same as the number of types in the
-		// key validator
-		table.PartitionKey = make([]*ColumnMetadata, len(keyValidatorParsed.types))
-
-		// V1 protocol only returns "regular" columns from
-		// system.schema_columns (there is no type field for columns)
-		// so the alias information is used to
-		// create the partition key and clustering columns
-
-		// construct the partition key from the alias
-		for i := range table.PartitionKey {
-			var alias string
-			if len(table.KeyAliases) > i {
-				alias = table.KeyAliases[i]
-			} else if i == 0 {
-				alias = DEFAULT_KEY_ALIAS
-			} else {
-				alias = DEFAULT_KEY_ALIAS + strconv.Itoa(i+1)
-			}
-
-			column := &ColumnMetadata{
-				Keyspace:       table.Keyspace,
-				Table:          table.Name,
-				Name:           alias,
-				Type:           keyValidatorParsed.types[i],
-				Kind:           ColumnPartitionKey,
-				ComponentIndex: i,
-			}
-
-			table.PartitionKey[i] = column
-			table.Columns[alias] = column
-		}
-
-		// determine the number of clustering columns
-		size := len(comparatorParsed.types)
-		if comparatorParsed.isComposite {
-			if len(comparatorParsed.collections) != 0 ||
-				(len(table.ColumnAliases) == size-1 &&
-					comparatorParsed.types[size-1].Type() == TypeVarchar) {
-				size = size - 1
-			}
-		} else {
-			if !(len(table.ColumnAliases) != 0 || len(table.Columns) == 0) {
-				size = 0
-			}
-		}
-
-		table.ClusteringColumns = make([]*ColumnMetadata, size)
-
-		for i := range table.ClusteringColumns {
-			var alias string
-			if len(table.ColumnAliases) > i {
-				alias = table.ColumnAliases[i]
-			} else if i == 0 {
-				alias = DEFAULT_COLUMN_ALIAS
-			} else {
-				alias = DEFAULT_COLUMN_ALIAS + strconv.Itoa(i+1)
-			}
-
-			order := ASC
-			if comparatorParsed.reversed[i] {
-				order = DESC
-			}
-
-			column := &ColumnMetadata{
-				Keyspace:       table.Keyspace,
-				Table:          table.Name,
-				Name:           alias,
-				Type:           comparatorParsed.types[i],
-				Order:          order,
-				Kind:           ColumnClusteringKey,
-				ComponentIndex: i,
-			}
-
-			table.ClusteringColumns[i] = column
-			table.Columns[alias] = column
-		}
-
-		if size != len(comparatorParsed.types)-1 {
-			alias := DEFAULT_VALUE_ALIAS
-			if len(table.ValueAlias) > 0 {
-				alias = table.ValueAlias
-			}
-			// decode the default validator
-			defaultValidatorParsed := parseType(table.DefaultValidator, byte(protoVer), logger)
-			column := &ColumnMetadata{
-				Keyspace: table.Keyspace,
-				Table:    table.Name,
-				Name:     alias,
-				Type:     defaultValidatorParsed.types[0],
-				Kind:     ColumnRegular,
-			}
-			table.Columns[alias] = column
-		}
-	}
+	compileV2Metadata(tables, protoVersion, logger)
 }
 
 // The simpler compile case for V2+ protocol
@@ -611,9 +495,6 @@ func getTableMetadata(session *Session, keyspaceName string) ([]TableMetadata, e
 		iter *Iter
 		scan func(iter *Iter, table *TableMetadata) bool
 		stmt string
-
-		keyAliasesJSON    []byte
-		columnAliasesJSON []byte
 	)
 
 	if session.useSystemSchema { // Cassandra 3.x+
@@ -647,31 +528,6 @@ func getTableMetadata(session *Session, keyspaceName string) ([]TableMetadata, e
 			}
 			return r
 		}
-	} else if session.cfg.ProtoVersion == protoVersion1 {
-		// we have key aliases
-		stmt = `
-		SELECT
-			columnfamily_name,
-			key_validator,
-			comparator,
-			default_validator,
-			key_aliases,
-			column_aliases,
-			value_alias
-		FROM system.schema_columnfamilies
-		WHERE keyspace_name = ?`
-
-		scan = func(iter *Iter, table *TableMetadata) bool {
-			return iter.Scan(
-				&table.Name,
-				&table.KeyValidator,
-				&table.Comparator,
-				&table.DefaultValidator,
-				&keyAliasesJSON,
-				&columnAliasesJSON,
-				&table.ValueAlias,
-			)
-		}
 	} else {
 		stmt = `
 		SELECT
@@ -698,34 +554,6 @@ func getTableMetadata(session *Session, keyspaceName string) ([]TableMetadata, e
 	table := TableMetadata{Keyspace: keyspaceName}
 
 	for scan(iter, &table) {
-		var err error
-
-		// decode the key aliases
-		if keyAliasesJSON != nil {
-			table.KeyAliases = []string{}
-			err = json.Unmarshal(keyAliasesJSON, &table.KeyAliases)
-			if err != nil {
-				iter.Close()
-				return nil, fmt.Errorf(
-					"invalid JSON value '%s' as key_aliases for in table '%s': %v",
-					keyAliasesJSON, table.Name, err,
-				)
-			}
-		}
-
-		// decode the column aliases
-		if columnAliasesJSON != nil {
-			table.ColumnAliases = []string{}
-			err = json.Unmarshal(columnAliasesJSON, &table.ColumnAliases)
-			if err != nil {
-				iter.Close()
-				return nil, fmt.Errorf(
-					"invalid JSON value '%s' as column_aliases for in table '%s': %v",
-					columnAliasesJSON, table.Name, err,
-				)
-			}
-		}
-
 		tables = append(tables, table)
 		table = TableMetadata{Keyspace: keyspaceName}
 	}
@@ -933,9 +761,6 @@ func getTypeInfo(t string, protoVer byte, logger StdLogger) TypeInfo {
 }
 
 func getUserTypeMetadata(session *Session, keyspaceName string) ([]UserTypeMetadata, error) {
-	if session.cfg.ProtoVersion == protoVersion1 {
-		return nil, nil
-	}
 	var tableName string
 	if session.useSystemSchema {
 		tableName = "system_schema.types"
@@ -1179,7 +1004,7 @@ func getMaterializedViewsMetadata(session *Session, keyspaceName string) ([]Mate
 }
 
 func getFunctionsMetadata(session *Session, keyspaceName string) ([]FunctionMetadata, error) {
-	if session.cfg.ProtoVersion == protoVersion1 || !session.hasAggregatesAndFunctions {
+	if !session.hasAggregatesAndFunctions {
 		return nil, nil
 	}
 	var tableName string
@@ -1234,7 +1059,7 @@ func getFunctionsMetadata(session *Session, keyspaceName string) ([]FunctionMeta
 }
 
 func getAggregatesMetadata(session *Session, keyspaceName string) ([]AggregateMetadata, error) {
-	if session.cfg.ProtoVersion == protoVersion1 || !session.hasAggregatesAndFunctions {
+	if !session.hasAggregatesAndFunctions {
 		return nil, nil
 	}
 	var tableName string
diff --git a/policies.go b/policies.go
index ed0b02f..219f472 100644
--- a/policies.go
+++ b/policies.go
@@ -323,6 +323,7 @@ func (host *selectedHost) Info() *HostInfo {
 func (host *selectedHost) Mark(err error) {}
 
 // NextHost is an iteration function over picked hosts
+// Should return nil eventually to prevent endless query execution.
 type NextHost func() SelectedHost
 
 // RoundRobinHostPolicy is a round-robin load balancing policy, where each host
diff --git a/query_executor.go b/query_executor.go
index 9eaf19d..61a43c8 100644
--- a/query_executor.go
+++ b/query_executor.go
@@ -27,6 +27,7 @@ package gocql
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -89,14 +90,16 @@ func (q *queryExecutor) executeQuery(qry ExecutableQuery) (*Iter, error) {
 	// check if the host id is specified for the query,
 	// if it is, the query should be executed at the corresponding host.
 	if hostID := qry.GetHostID(); hostID != "" {
+		host, ok := q.pool.session.ring.getHost(hostID)
+		if !ok {
+			return nil, ErrNoConnections
+		}
+		var returnedHostOnce int32 = 0
 		hostIter = func() SelectedHost {
-			pool, ok := q.pool.getPoolByHostID(hostID)
-			// if the specified host is down
-			// we return nil to avoid endless query execution in queryExecutor.do()
-			if !ok || !pool.host.IsUp() {
-				return nil
+			if atomic.CompareAndSwapInt32(&returnedHostOnce, 0, 1) {
+				return (*selectedHost)(host)
 			}
-			return (*selectedHost)(pool.host)
+			return nil
 		}
 	}
 
diff --git a/ring.go b/ring.go
index 6821c0d..1dbf07c 100644
--- a/ring.go
+++ b/ring.go
@@ -67,11 +67,11 @@ func (r *ring) getHostByIP(ip string) (*HostInfo, bool) {
 	return r.hosts[hi], ok
 }
 
-func (r *ring) getHost(hostID string) *HostInfo {
+func (r *ring) getHost(hostID string) (host *HostInfo, ok bool) {
 	r.mu.RLock()
-	host := r.hosts[hostID]
+	host, ok = r.hosts[hostID]
 	r.mu.RUnlock()
-	return host
+	return
 }
 
 func (r *ring) allHosts() []*HostInfo {
diff --git a/session.go b/session.go
index ed1a078..dd972f7 100644
--- a/session.go
+++ b/session.go
@@ -75,8 +75,6 @@ type Session struct {
 	ring     ring
 	metadata clusterMetadata
 
-	mu sync.RWMutex
-
 	control *controlConn
 
 	// event handlers
@@ -153,7 +151,7 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 
 	s := &Session{
 		cons:            cfg.Consistency,
-		prefetch:        0.25,
+		prefetch:        cfg.NextPagePrefetch,
 		cfg:             cfg,
 		pageSize:        cfg.PageSize,
 		stmtsLRU:        &preparedLRU{lru: lru.New(cfg.MaxPreparedStmts)},
@@ -161,6 +159,7 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 		ctx:             ctx,
 		cancel:          cancel,
 		logger:          cfg.logger(),
+		trace:           cfg.Tracer,
 	}
 
 	s.schemaDescriber = newSchemaDescriber(s)
@@ -416,41 +415,6 @@ func (s *Session) reconnectDownedHosts(intv time.Duration) {
 	}
 }
 
-// SetConsistency sets the default consistency level for this session. This
-// setting can also be changed on a per-query basis and the default value
-// is Quorum.
-func (s *Session) SetConsistency(cons Consistency) {
-	s.mu.Lock()
-	s.cons = cons
-	s.mu.Unlock()
-}
-
-// SetPageSize sets the default page size for this session. A value <= 0 will
-// disable paging. This setting can also be changed on a per-query basis.
-func (s *Session) SetPageSize(n int) {
-	s.mu.Lock()
-	s.pageSize = n
-	s.mu.Unlock()
-}
-
-// SetPrefetch sets the default threshold for pre-fetching new pages. If
-// there are only p*pageSize rows remaining, the next page will be requested
-// automatically. This value can also be changed on a per-query basis and
-// the default value is 0.25.
-func (s *Session) SetPrefetch(p float64) {
-	s.mu.Lock()
-	s.prefetch = p
-	s.mu.Unlock()
-}
-
-// SetTrace sets the default tracer for this session. This setting can also
-// be changed on a per-query basis.
-func (s *Session) SetTrace(trace Tracer) {
-	s.mu.Lock()
-	s.trace = trace
-	s.mu.Unlock()
-}
-
 // Query generates a new query object for interacting with the database.
 // Further details of the query may be tweaked using the resulting query
 // value before the query is executed. Query is automatically prepared
@@ -775,6 +739,7 @@ func (s *Session) executeBatch(batch *Batch) *Iter {
 	return iter
 }
 
+// Deprecated: use Batch.Exec instead.
 // ExecuteBatch executes a batch operation and returns nil if successful
 // otherwise an error is returned describing the failure.
 func (s *Session) ExecuteBatch(batch *Batch) error {
@@ -782,13 +747,23 @@ func (s *Session) ExecuteBatch(batch *Batch) error {
 	return iter.Close()
 }
 
+// Deprecated: use Batch.ExecCAS instead
 // ExecuteBatchCAS executes a batch operation and returns true if successful and
 // an iterator (to scan additional rows if more than one conditional statement)
 // was sent.
 // Further scans on the interator must also remember to include
 // the applied boolean as the first argument to *Iter.Scan
 func (s *Session) ExecuteBatchCAS(batch *Batch, dest ...interface{}) (applied bool, iter *Iter, err error) {
-	iter = s.executeBatch(batch)
+	return batch.ExecCAS(dest...)
+}
+
+// ExecCAS executes a batch operation and returns true if successful and
+// an iterator (to scan additional rows if more than one conditional statement)
+// was sent.
+// Further scans on the interator must also remember to include
+// the applied boolean as the first argument to *Iter.Scan
+func (b *Batch) ExecCAS(dest ...interface{}) (applied bool, iter *Iter, err error) {
+	iter = b.session.executeBatch(b)
 	if err := iter.checkErrAndNotFound(); err != nil {
 		iter.Close()
 		return false, nil, err
@@ -804,11 +779,19 @@ func (s *Session) ExecuteBatchCAS(batch *Batch, dest ...interface{}) (applied bo
 	return applied, iter, iter.err
 }
 
+// Deprecated: use Batch.MapExecCAS instead
 // MapExecuteBatchCAS executes a batch operation much like ExecuteBatchCAS,
 // however it accepts a map rather than a list of arguments for the initial
 // scan.
 func (s *Session) MapExecuteBatchCAS(batch *Batch, dest map[string]interface{}) (applied bool, iter *Iter, err error) {
-	iter = s.executeBatch(batch)
+	return batch.MapExecCAS(dest)
+}
+
+// MapExecCAS executes a batch operation much like ExecuteBatchCAS,
+// however it accepts a map rather than a list of arguments for the initial
+// scan.
+func (b *Batch) MapExecCAS(dest map[string]interface{}) (applied bool, iter *Iter, err error) {
+	iter = b.session.executeBatch(b)
 	if err := iter.checkErrAndNotFound(); err != nil {
 		iter.Close()
 		return false, nil, err
@@ -986,7 +969,6 @@ type queryRoutingInfo struct {
 func (q *Query) defaultsFromSession() {
 	s := q.session
 
-	s.mu.RLock()
 	q.cons = s.cons
 	q.pageSize = s.pageSize
 	q.trace = s.trace
@@ -999,7 +981,6 @@ func (q *Query) defaultsFromSession() {
 	q.metrics = &queryMetrics{m: make(map[string]*hostMetrics)}
 
 	q.spec = &NonSpeculativeExecution{}
-	s.mu.RUnlock()
 }
 
 // Statement returns the statement that was used to generate this query.
@@ -1828,16 +1809,14 @@ type Batch struct {
 	routingInfo *queryRoutingInfo
 }
 
+// Deprecated: use Session.Batch instead
 // NewBatch creates a new batch operation using defaults defined in the cluster
-//
-// Deprecated: use session.Batch instead
 func (s *Session) NewBatch(typ BatchType) *Batch {
 	return s.Batch(typ)
 }
 
 // Batch creates a new batch operation using defaults defined in the cluster
 func (s *Session) Batch(typ BatchType) *Batch {
-	s.mu.RLock()
 	batch := &Batch{
 		Type:             typ,
 		rt:               s.cfg.RetryPolicy,
@@ -1853,7 +1832,6 @@ func (s *Session) Batch(typ BatchType) *Batch {
 		routingInfo:      &queryRoutingInfo{},
 	}
 
-	s.mu.RUnlock()
 	return batch
 }
 



Test patch:
diff --git a/batch_test.go b/batch_test.go
index 3a8a7e7..47adff8 100644
--- a/batch_test.go
+++ b/batch_test.go
@@ -40,10 +40,6 @@ func TestBatch_Errors(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
 
-	if session.cfg.ProtoVersion < protoVersion2 {
-		t.Skip("atomic batches not supported. Please use Cassandra >= 2.0")
-	}
-
 	if err := createTable(session, `CREATE TABLE gocql_test.batch_errors (id int primary key, val inet)`); err != nil {
 		t.Fatal(err)
 	}
@@ -59,10 +55,6 @@ func TestBatch_WithTimestamp(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
 
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("Batch timestamps are only available on protocol >= 3")
-	}
-
 	if err := createTable(session, `CREATE TABLE gocql_test.batch_ts (id int primary key, val text)`); err != nil {
 		t.Fatal(err)
 	}
diff --git a/cassandra_test.go b/cassandra_test.go
index 4690990..1a9b69f 100644
--- a/cassandra_test.go
+++ b/cassandra_test.go
@@ -152,11 +152,10 @@ func TestTracing(t *testing.T) {
 	}
 
 	// also works from session tracer
-	session.SetTrace(trace)
 	trace.mu.Lock()
 	buf.Reset()
 	trace.mu.Unlock()
-	if err := session.Query(`SELECT id FROM trace WHERE id = ?`, 42).Scan(&value); err != nil {
+	if err := session.Query(`SELECT id FROM trace WHERE id = ?`, 42).Trace(trace).Scan(&value); err != nil {
 		t.Fatal("select:", err)
 	}
 	if buf.Len() == 0 {
@@ -457,7 +456,7 @@ func TestCAS(t *testing.T) {
 
 	successBatch := session.Batch(LoggedBatch)
 	successBatch.Query("INSERT INTO cas_table (title, revid, last_modified) VALUES (?, ?, ?) IF NOT EXISTS", title, revid, modified)
-	if applied, _, err := session.ExecuteBatchCAS(successBatch, &titleCAS, &revidCAS, &modifiedCAS); err != nil {
+	if applied, _, err := successBatch.ExecCAS(&titleCAS, &revidCAS, &modifiedCAS); err != nil {
 		t.Fatal("insert:", err)
 	} else if !applied {
 		t.Fatalf("insert should have been applied: title=%v revID=%v modified=%v", titleCAS, revidCAS, modifiedCAS)
@@ -466,7 +465,7 @@ func TestCAS(t *testing.T) {
 	successBatch = session.Batch(LoggedBatch)
 	successBatch.Query("INSERT INTO cas_table (title, revid, last_modified) VALUES (?, ?, ?) IF NOT EXISTS", title+"_foo", revid, modified)
 	casMap := make(map[string]interface{})
-	if applied, _, err := session.MapExecuteBatchCAS(successBatch, casMap); err != nil {
+	if applied, _, err := successBatch.MapExecCAS(casMap); err != nil {
 		t.Fatal("insert:", err)
 	} else if !applied {
 		t.Fatal("insert should have been applied")
@@ -474,7 +473,7 @@ func TestCAS(t *testing.T) {
 
 	failBatch := session.Batch(LoggedBatch)
 	failBatch.Query("INSERT INTO cas_table (title, revid, last_modified) VALUES (?, ?, ?) IF NOT EXISTS", title, revid, modified)
-	if applied, _, err := session.ExecuteBatchCAS(successBatch, &titleCAS, &revidCAS, &modifiedCAS); err != nil {
+	if applied, _, err := successBatch.ExecCAS(&titleCAS, &revidCAS, &modifiedCAS); err != nil {
 		t.Fatal("insert:", err)
 	} else if applied {
 		t.Fatalf("insert should have not been applied: title=%v revID=%v modified=%v", titleCAS, revidCAS, modifiedCAS)
@@ -483,14 +482,14 @@ func TestCAS(t *testing.T) {
 	insertBatch := session.Batch(LoggedBatch)
 	insertBatch.Query("INSERT INTO cas_table (title, revid, last_modified) VALUES ('_foo', 2c3af400-73a4-11e5-9381-29463d90c3f0, TOTIMESTAMP(NOW()))")
 	insertBatch.Query("INSERT INTO cas_table (title, revid, last_modified) VALUES ('_foo', 3e4ad2f1-73a4-11e5-9381-29463d90c3f0, TOTIMESTAMP(NOW()))")
-	if err := session.ExecuteBatch(insertBatch); err != nil {
+	if err := insertBatch.Exec(); err != nil {
 		t.Fatal("insert:", err)
 	}
 
 	failBatch = session.Batch(LoggedBatch)
 	failBatch.Query("UPDATE cas_table SET last_modified = TOTIMESTAMP(NOW()) WHERE title='_foo' AND revid=2c3af400-73a4-11e5-9381-29463d90c3f0 IF last_modified=TOTIMESTAMP(NOW());")
 	failBatch.Query("UPDATE cas_table SET last_modified = TOTIMESTAMP(NOW()) WHERE title='_foo' AND revid=3e4ad2f1-73a4-11e5-9381-29463d90c3f0 IF last_modified=TOTIMESTAMP(NOW());")
-	if applied, iter, err := session.ExecuteBatchCAS(failBatch, &titleCAS, &revidCAS, &modifiedCAS); err != nil {
+	if applied, iter, err := failBatch.ExecCAS(&titleCAS, &revidCAS, &modifiedCAS); err != nil {
 		t.Fatal("insert:", err)
 	} else if applied {
 		t.Fatalf("insert should have not been applied: title=%v revID=%v modified=%v", titleCAS, revidCAS, modifiedCAS)
@@ -521,20 +520,20 @@ func TestCAS(t *testing.T) {
 	notCASBatch := session.Batch(LoggedBatch)
 	notCASBatch.Query("INSERT INTO cas_table (title, revid, last_modified) VALUES (?, ?, ?)", title+"_baz", revid, modified)
 	casMap = make(map[string]interface{})
-	if _, _, err := session.MapExecuteBatchCAS(notCASBatch, casMap); err != ErrNotFound {
+	if _, _, err := notCASBatch.MapExecCAS(casMap); err != ErrNotFound {
 		t.Fatal("insert should have returned not found:", err)
 	}
 
 	notCASBatch = session.Batch(LoggedBatch)
 	notCASBatch.Query("INSERT INTO cas_table (title, revid, last_modified) VALUES (?, ?, ?)", title+"_baz", revid, modified)
 	casMap = make(map[string]interface{})
-	if _, _, err := session.ExecuteBatchCAS(notCASBatch, &revidCAS); err != ErrNotFound {
+	if _, _, err := notCASBatch.ExecCAS(&revidCAS); err != ErrNotFound {
 		t.Fatal("insert should have returned not found:", err)
 	}
 
 	failBatch = session.Batch(LoggedBatch)
 	failBatch.Query("UPDATE cas_table SET last_modified = TOTIMESTAMP(NOW()) WHERE title='_foo' AND revid=3e4ad2f1-73a4-11e5-9381-29463d90c3f0 IF last_modified = ?", modified)
-	if _, _, err := session.ExecuteBatchCAS(failBatch, new(bool)); err == nil {
+	if _, _, err := failBatch.ExecCAS(new(bool)); err == nil {
 		t.Fatal("update should have errored")
 	}
 	// make sure MapScanCAS does not panic when MapScan fails
@@ -550,7 +549,7 @@ func TestCAS(t *testing.T) {
 	failBatch.Query("UPDATE cas_table SET last_modified = TOTIMESTAMP(NOW()) WHERE title='_foo' AND revid=3e4ad2f1-73a4-11e5-9381-29463d90c3f0 IF last_modified = ?", modified)
 	casMap = make(map[string]interface{})
 	casMap["last_modified"] = false
-	if _, _, err := session.MapExecuteBatchCAS(failBatch, casMap); err == nil {
+	if _, _, err := failBatch.MapExecCAS(casMap); err == nil {
 		t.Fatal("update should have errored")
 	}
 }
@@ -752,7 +751,7 @@ func TestBatch(t *testing.T) {
 		batch.Query(`INSERT INTO batch_table (id) VALUES (?)`, i)
 	}
 
-	if err := session.ExecuteBatch(batch); err != nil {
+	if err := batch.Exec(); err != nil {
 		t.Fatal("execute batch:", err)
 	}
 
@@ -788,7 +787,7 @@ func TestUnpreparedBatch(t *testing.T) {
 		batch.Query(`UPDATE batch_unprepared SET c = c + 1 WHERE id = 1`)
 	}
 
-	if err := session.ExecuteBatch(batch); err != nil {
+	if err := batch.Exec(); err != nil {
 		t.Fatal("execute batch:", err)
 	}
 
@@ -824,8 +823,8 @@ func TestBatchLimit(t *testing.T) {
 	for i := 0; i < 65537; i++ {
 		batch.Query(`INSERT INTO batch_table2 (id) VALUES (?)`, i)
 	}
-	if err := session.ExecuteBatch(batch); err != ErrTooManyStmts {
-		t.Fatal("gocql attempted to execute a batch larger than the support limit of statements.")
+	if err := batch.Exec(); err != ErrTooManyStmts {
+		t.Fatalf("gocql attempted to execute a batch larger than the support limit of statements: expected %v, got %v", ErrTooManyStmts, err)
 	}
 
 }
@@ -876,7 +875,7 @@ func TestTooManyQueryArgs(t *testing.T) {
 
 	batch := session.Batch(UnloggedBatch)
 	batch.Query("INSERT INTO too_many_query_args (id, value) VALUES (?, ?)", 1, 2, 3)
-	err = session.ExecuteBatch(batch)
+	err = batch.Exec()
 
 	if err == nil {
 		t.Fatal("'`INSERT INTO too_many_query_args (id, value) VALUES (?, ?)`, 1, 2, 3' should return an error")
@@ -908,7 +907,7 @@ func TestNotEnoughQueryArgs(t *testing.T) {
 
 	batch := session.Batch(UnloggedBatch)
 	batch.Query("INSERT INTO not_enough_query_args (id, cluster, value) VALUES (?, ?, ?)", 1, 2)
-	err = session.ExecuteBatch(batch)
+	err = batch.Exec()
 
 	if err == nil {
 		t.Fatal("'`INSERT INTO not_enough_query_args (id, cluster, value) VALUES (?, ?, ?)`, 1, 2' should return an error")
@@ -1534,7 +1533,7 @@ func TestBatchQueryInfo(t *testing.T) {
 	batch := session.Batch(LoggedBatch)
 	batch.Bind("INSERT INTO batch_query_info (id, cluster, value) VALUES (?, ?,?)", write)
 
-	if err := session.ExecuteBatch(batch); err != nil {
+	if err := batch.Exec(); err != nil {
 		t.Fatalf("batch insert into batch_query_info failed, err '%v'", err)
 	}
 
@@ -2047,7 +2046,7 @@ func TestBatchStats(t *testing.T) {
 	b.Query("INSERT INTO batchStats (id) VALUES (?)", 1)
 	b.Query("INSERT INTO batchStats (id) VALUES (?)", 2)
 
-	if err := session.ExecuteBatch(b); err != nil {
+	if err := b.Exec(); err != nil {
 		t.Fatalf("query failed. %v", err)
 	} else {
 		if b.Attempts() < 1 {
@@ -2104,7 +2103,7 @@ func TestBatchObserve(t *testing.T) {
 		batch.Query(fmt.Sprintf(`INSERT INTO batch_observe_table (id,other) VALUES (?,%d)`, i), i)
 	}
 
-	if err := session.ExecuteBatch(batch); err != nil {
+	if err := batch.Exec(); err != nil {
 		t.Fatal("execute batch:", err)
 	}
 	if observedBatch == nil {
@@ -2285,25 +2284,7 @@ func TestGetTableMetadata(t *testing.T) {
 	if testTable == nil {
 		t.Fatal("Expected table metadata for name 'test_table_metadata'")
 	}
-	if session.cfg.ProtoVersion == protoVersion1 {
-		if testTable.KeyValidator != "org.apache.cassandra.db.marshal.Int32Type" {
-			t.Errorf("Expected test_table_metadata key validator to be 'org.apache.cassandra.db.marshal.Int32Type' but was '%s'", testTable.KeyValidator)
-		}
-		if testTable.Comparator != "org.apache.cassandra.db.marshal.CompositeType(org.apache.cassandra.db.marshal.Int32Type,org.apache.cassandra.db.marshal.UTF8Type)" {
-			t.Errorf("Expected test_table_metadata key validator to be 'org.apache.cassandra.db.marshal.CompositeType(org.apache.cassandra.db.marshal.Int32Type,org.apache.cassandra.db.marshal.UTF8Type)' but was '%s'", testTable.Comparator)
-		}
-		if testTable.DefaultValidator != "org.apache.cassandra.db.marshal.BytesType" {
-			t.Errorf("Expected test_table_metadata key validator to be 'org.apache.cassandra.db.marshal.BytesType' but was '%s'", testTable.DefaultValidator)
-		}
-		expectedKeyAliases := []string{"first_id"}
-		if !reflect.DeepEqual(testTable.KeyAliases, expectedKeyAliases) {
-			t.Errorf("Expected key aliases %v but was %v", expectedKeyAliases, testTable.KeyAliases)
-		}
-		expectedColumnAliases := []string{"second_id"}
-		if !reflect.DeepEqual(testTable.ColumnAliases, expectedColumnAliases) {
-			t.Errorf("Expected key aliases %v but was %v", expectedColumnAliases, testTable.ColumnAliases)
-		}
-	}
+
 	if testTable.ValueAlias != "" {
 		t.Errorf("Expected value alias '' but was '%s'", testTable.ValueAlias)
 	}
@@ -3229,10 +3210,6 @@ func TestUnmarshallNestedTypes(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
 
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("can not have frozen types in cassandra < 2.1.3")
-	}
-
 	if err := createTable(session, `CREATE TABLE gocql_test.test_557 (
 		    id text PRIMARY KEY,
 		    val list<frozen<map<text, text> > >
@@ -3408,7 +3385,7 @@ func TestUnsetColBatch(t *testing.T) {
 	b.Query("INSERT INTO gocql_test.batchUnsetInsert(id, my_int, my_text) VALUES (?,?,?)", 1, UnsetValue, "")
 	b.Query("INSERT INTO gocql_test.batchUnsetInsert(id, my_int, my_text) VALUES (?,?,?)", 2, 2, UnsetValue)
 
-	if err := session.ExecuteBatch(b); err != nil {
+	if err := b.Exec(); err != nil {
 		t.Fatalf("query failed. %v", err)
 	} else {
 		if b.Attempts() < 1 {
diff --git a/conn_test.go b/conn_test.go
index e9b33fe..cb76d90 100644
--- a/conn_test.go
+++ b/conn_test.go
@@ -53,7 +53,7 @@ import (
 )
 
 const (
-	defaultProto = protoVersion2
+	defaultProto = protoVersion4
 )
 
 func TestApprove(t *testing.T) {
@@ -333,17 +333,17 @@ func TestCancel(t *testing.T) {
 	// Make sure we finish the query without leftovers
 	var wg sync.WaitGroup
 	wg.Add(1)
-
 	go func() {
-		if err := qry.Exec(); err != context.Canceled {
-			t.Fatalf("expected to get context cancel error: '%v', got '%v'", context.Canceled, err)
-		}
+		err = qry.Exec()
 		wg.Done()
 	}()
 
 	// The query will timeout after about 1 seconds, so cancel it after a short pause
 	time.AfterFunc(20*time.Millisecond, cancel)
 	wg.Wait()
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected to get context cancel error: '%v', got '%v'", context.Canceled, err)
+	}
 }
 
 type testQueryObserver struct {
@@ -1056,18 +1056,13 @@ func (nts newTestServerOpts) newServer(t testing.TB, ctx context.Context) *TestS
 		t.Fatal(err)
 	}
 
-	headerSize := 8
-	if nts.protocol > protoVersion2 {
-		headerSize = 9
-	}
-
 	ctx, cancel := context.WithCancel(ctx)
 	srv := &TestServer{
 		Address:    listen.Addr().String(),
 		listen:     listen,
 		t:          t,
 		protocol:   nts.protocol,
-		headerSize: headerSize,
+		headerSize: 9,
 		ctx:        ctx,
 		cancel:     cancel,
 
@@ -1103,18 +1098,13 @@ func NewSSLTestServer(t testing.TB, protocol uint8, ctx context.Context) *TestSe
 		t.Fatal(err)
 	}
 
-	headerSize := 8
-	if protocol > protoVersion2 {
-		headerSize = 9
-	}
-
 	ctx, cancel := context.WithCancel(ctx)
 	srv := &TestServer{
 		Address:    listen.Addr().String(),
 		listen:     listen,
 		t:          t,
 		protocol:   protocol,
-		headerSize: headerSize,
+		headerSize: 9,
 		ctx:        ctx,
 		cancel:     cancel,
 	}
diff --git a/events_ccm_test.go b/events_ccm_test.go
index a105985..6c605d7 100644
--- a/events_ccm_test.go
+++ b/events_ccm_test.go
@@ -104,7 +104,7 @@ func TestEventNodeDownControl(t *testing.T) {
 	}
 	session.pool.mu.RUnlock()
 
-	host := session.ring.getHost(node.Addr)
+	host, _ := session.ring.getHost(node.Addr)
 	if host == nil {
 		t.Fatal("node not in metadata ring")
 	} else if host.IsUp() {
@@ -146,7 +146,7 @@ func TestEventNodeDown(t *testing.T) {
 		t.Fatal("node not removed after remove event")
 	}
 
-	host := session.ring.getHost(node.Addr)
+	host, _ := session.ring.getHost(node.Addr)
 	if host == nil {
 		t.Fatal("node not in metadata ring")
 	} else if host.IsUp() {
@@ -203,7 +203,7 @@ func TestEventNodeUp(t *testing.T) {
 		t.Fatal("node not added after node added event")
 	}
 
-	host := session.ring.getHost(node.Addr)
+	host, _ := session.ring.getHost(node.Addr)
 	if host == nil {
 		t.Fatal("node not in metadata ring")
 	} else if !host.IsUp() {
diff --git a/example_batch_test.go b/example_batch_test.go
index b27085c..b5de6be 100644
--- a/example_batch_test.go
+++ b/example_batch_test.go
@@ -61,7 +61,7 @@ func Example_batch() {
 		Idempotent: true,
 	})
 
-	err = session.ExecuteBatch(b)
+	err = b.Exec()
 	if err != nil {
 		log.Fatal(err)
 	}
diff --git a/example_lwt_batch_test.go b/example_lwt_batch_test.go
index c3cc838..8f6399a 100644
--- a/example_lwt_batch_test.go
+++ b/example_lwt_batch_test.go
@@ -72,7 +72,7 @@ func ExampleSession_MapExecuteBatchCAS() {
 			Args: []interface{}{"B", "pk1", "ck2", ck2Version},
 		})
 		m := make(map[string]interface{})
-		applied, iter, err := session.MapExecuteBatchCAS(b.WithContext(ctx), m)
+		applied, iter, err := b.WithContext(ctx).MapExecCAS(m)
 		if err != nil {
 			log.Fatal(err)
 		}
diff --git a/frame_test.go b/frame_test.go
index 8cb9024..6bf4922 100644
--- a/frame_test.go
+++ b/frame_test.go
@@ -108,14 +108,14 @@ func TestFrameReadTooLong(t *testing.T) {
 	r := &bytes.Buffer{}
 	r.Write(make([]byte, maxFrameSize+1))
 	// write a new header right after this frame to verify that we can read it
-	r.Write([]byte{0x02, 0x00, 0x00, byte(opReady), 0x00, 0x00, 0x00, 0x00})
+	r.Write([]byte{protoVersionMask & protoVersion3, 0x00, 0x00, 0x00, byte(opReady), 0x00, 0x00, 0x00, 0x00})
 
-	framer := newFramer(nil, 2)
+	framer := newFramer(nil, 3)
 
 	head := frameHeader{
-		version: 2,
+		version: protoVersion3,
 		op:      opReady,
-		length:  r.Len() - 8,
+		length:  r.Len() - frameHeadSize,
 	}
 
 	err := framer.readFrame(r, &head)
@@ -123,7 +123,7 @@ func TestFrameReadTooLong(t *testing.T) {
 		t.Fatalf("expected to get %v got %v", ErrFrameTooBig, err)
 	}
 
-	head, err = readHeader(r, make([]byte, 8))
+	head, err = readHeader(r, make([]byte, frameHeadSize))
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/host_source_test.go b/host_source_test.go
index cd4dfc0..c8f93c7 100644
--- a/host_source_test.go
+++ b/host_source_test.go
@@ -307,6 +307,32 @@ func TestRefreshDebouncer_EventsAfterRefreshNow(t *testing.T) {
 	}
 }
 
+// https://github.com/gocql/gocql/issues/1752
+func TestRefreshDebouncer_DeadlockOnStop(t *testing.T) {
+	// there's no way to guarantee this bug manifests because it depends on which `case` is picked from the `select`
+	// with 4 iterations of this test the deadlock would be hit pretty consistently
+	const iterations = 4
+	for i := 0; i < iterations; i++ {
+		refreshCalledCh := make(chan int, 5)
+		refreshDuration := 500 * time.Millisecond
+		fn := func() error {
+			refreshCalledCh <- 0
+			time.Sleep(refreshDuration)
+			return nil
+		}
+		d := newRefreshDebouncer(50*time.Millisecond, fn)
+		timeBeforeRefresh := time.Now()
+		_ = d.refreshNow()
+		<-refreshCalledCh
+		d.debounce()
+		d.stop()
+		timeAfterRefresh := time.Now()
+		if timeAfterRefresh.Sub(timeBeforeRefresh) < refreshDuration {
+			t.Errorf("refresh debouncer stop() didn't wait until flusher stopped")
+		}
+	}
+}
+
 func TestErrorBroadcaster_MultipleListeners(t *testing.T) {
 	b := newErrorBroadcaster()
 	defer b.stop()
diff --git a/integration_test.go b/integration_test.go
index 61ffbf5..ccc5939 100644
--- a/integration_test.go
+++ b/integration_test.go
@@ -221,7 +221,7 @@ func TestCustomPayloadMessages(t *testing.T) {
 	b := session.Batch(LoggedBatch)
 	b.CustomPayload = customPayload
 	b.Query("INSERT INTO testCustomPayloadMessages(id,value) VALUES(1, 1)")
-	if err := session.ExecuteBatch(b); err != nil {
+	if err := b.Exec(); err != nil {
 		t.Fatalf("query failed. %v", err)
 	}
 }
diff --git a/marshal_test.go b/marshal_test.go
index 969d8e5..cdbc646 100644
--- a/marshal_test.go
+++ b/marshal_test.go
@@ -59,56 +59,56 @@ var marshalTests = []struct {
 	UnmarshalError error
 }{
 	{
-		NativeType{proto: 2, typ: TypeVarchar},
+		NativeType{proto: protoVersion3, typ: TypeVarchar},
 		[]byte("hello world"),
 		[]byte("hello world"),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarchar},
+		NativeType{proto: protoVersion3, typ: TypeVarchar},
 		[]byte("hello world"),
 		"hello world",
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarchar},
+		NativeType{proto: protoVersion3, typ: TypeVarchar},
 		[]byte(nil),
 		[]byte(nil),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarchar},
+		NativeType{proto: protoVersion3, typ: TypeVarchar},
 		[]byte("hello world"),
 		MyString("hello world"),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarchar},
+		NativeType{proto: protoVersion3, typ: TypeVarchar},
 		[]byte("HELLO WORLD"),
 		CustomString("hello world"),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBlob},
+		NativeType{proto: protoVersion3, typ: TypeBlob},
 		[]byte("hello\x00"),
 		[]byte("hello\x00"),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBlob},
+		NativeType{proto: protoVersion3, typ: TypeBlob},
 		[]byte(nil),
 		[]byte(nil),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTimeUUID},
+		NativeType{proto: protoVersion3, typ: TypeTimeUUID},
 		[]byte{0x3d, 0xcd, 0x98, 0x0, 0xf3, 0xd9, 0x11, 0xbf, 0x86, 0xd4, 0xb8, 0xe8, 0x56, 0x2c, 0xc, 0xd0},
 		func() UUID {
 			x, _ := UUIDFromBytes([]byte{0x3d, 0xcd, 0x98, 0x0, 0xf3, 0xd9, 0x11, 0xbf, 0x86, 0xd4, 0xb8, 0xe8, 0x56, 0x2c, 0xc, 0xd0})
@@ -118,287 +118,287 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTimeUUID},
+		NativeType{proto: protoVersion3, typ: TypeTimeUUID},
 		[]byte{0xb8, 0xe8, 0x56, 0x2c, 0xc, 0xd0},
 		[]byte{0xb8, 0xe8, 0x56, 0x2c, 0xc, 0xd0},
 		MarshalError("can not marshal []byte 6 bytes long into timeuuid, must be exactly 16 bytes long"),
 		UnmarshalError("unable to parse UUID: UUIDs must be exactly 16 bytes long"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeTimeUUID},
+		NativeType{proto: protoVersion3, typ: TypeTimeUUID},
 		[]byte{0x3d, 0xcd, 0x98, 0x0, 0xf3, 0xd9, 0x11, 0xbf, 0x86, 0xd4, 0xb8, 0xe8, 0x56, 0x2c, 0xc, 0xd0},
 		[16]byte{0x3d, 0xcd, 0x98, 0x0, 0xf3, 0xd9, 0x11, 0xbf, 0x86, 0xd4, 0xb8, 0xe8, 0x56, 0x2c, 0xc, 0xd0},
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x00\x00\x00\x00"),
 		0,
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x01\x02\x03\x04"),
 		int(16909060),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x01\x02\x03\x04"),
 		AliasInt(16909060),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x80\x00\x00\x00"),
 		int32(math.MinInt32),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x7f\xff\xff\xff"),
 		int32(math.MaxInt32),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x00\x00\x00\x00"),
 		"0",
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x01\x02\x03\x04"),
 		"16909060",
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x80\x00\x00\x00"),
 		"-2147483648", // math.MinInt32
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x7f\xff\xff\xff"),
 		"2147483647", // math.MaxInt32
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x00\x00\x00\x00\x00"),
 		0,
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x01\x02\x03\x04\x05\x06\x07\x08"),
 		72623859790382856,
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x80\x00\x00\x00\x00\x00\x00\x00"),
 		int64(math.MinInt64),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x7f\xff\xff\xff\xff\xff\xff\xff"),
 		int64(math.MaxInt64),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x00\x00\x00\x00\x00"),
 		"0",
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x01\x02\x03\x04\x05\x06\x07\x08"),
 		"72623859790382856",
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x80\x00\x00\x00\x00\x00\x00\x00"),
 		"-9223372036854775808", // math.MinInt64
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x7f\xff\xff\xff\xff\xff\xff\xff"),
 		"9223372036854775807", // math.MaxInt64
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBoolean},
+		NativeType{proto: protoVersion3, typ: TypeBoolean},
 		[]byte("\x00"),
 		false,
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBoolean},
+		NativeType{proto: protoVersion3, typ: TypeBoolean},
 		[]byte("\x01"),
 		true,
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeFloat},
+		NativeType{proto: protoVersion3, typ: TypeFloat},
 		[]byte("\x40\x49\x0f\xdb"),
 		float32(3.14159265),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDouble},
+		NativeType{proto: protoVersion3, typ: TypeDouble},
 		[]byte("\x40\x09\x21\xfb\x53\xc8\xd4\xf1"),
 		float64(3.14159265),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\x00\x00\x00\x00\x00"),
 		inf.NewDec(0, 0),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\x00\x00\x00\x00\x64"),
 		inf.NewDec(100, 0),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\x00\x00\x00\x02\x19"),
 		decimalize("0.25"),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\x00\x00\x00\x13\xD5\a;\x20\x14\xA2\x91"),
 		decimalize("-0.0012095473475870063"), // From the iconara/cql-rb test suite
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\x00\x00\x00\x13*\xF8\xC4\xDF\xEB]o"),
 		decimalize("0.0012095473475870063"), // From the iconara/cql-rb test suite
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\x00\x00\x00\x12\xF2\xD8\x02\xB6R\x7F\x99\xEE\x98#\x99\xA9V"),
 		decimalize("-1042342234234.123423435647768234"), // From the iconara/cql-rb test suite
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\x00\x00\x00\r\nJ\x04\"^\x91\x04\x8a\xb1\x18\xfe"),
 		decimalize("1243878957943.1234124191998"), // From the datastax/python-driver test suite
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\x00\x00\x00\x06\xe5\xde]\x98Y"),
 		decimalize("-112233.441191"), // From the datastax/python-driver test suite
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\x00\x00\x00\x14\x00\xfa\xce"),
 		decimalize("0.00000000000000064206"), // From the datastax/python-driver test suite
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\x00\x00\x00\x14\xff\x052"),
 		decimalize("-0.00000000000000064206"), // From the datastax/python-driver test suite
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\xff\xff\xff\x9c\x00\xfa\xce"),
 		inf.NewDec(64206, -100), // From the datastax/python-driver test suite
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 4, typ: TypeTime},
+		NativeType{proto: protoVersion4, typ: TypeTime},
 		[]byte("\x00\x00\x01\x40\x77\x16\xe1\xb8"),
 		time.Duration(int64(1376387523000)),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 4, typ: TypeTime},
+		NativeType{proto: protoVersion4, typ: TypeTime},
 		[]byte("\x00\x00\x01\x40\x77\x16\xe1\xb8"),
 		int64(1376387523000),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTimestamp},
+		NativeType{proto: protoVersion3, typ: TypeTimestamp},
 		[]byte("\x00\x00\x01\x40\x77\x16\xe1\xb8"),
 		time.Date(2013, time.August, 13, 9, 52, 3, 0, time.UTC),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTimestamp},
+		NativeType{proto: protoVersion3, typ: TypeTimestamp},
 		[]byte("\x00\x00\x01\x40\x77\x16\xe1\xb8"),
 		int64(1376387523000),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 5, typ: TypeDuration},
+		NativeType{proto: protoVersion5, typ: TypeDuration},
 		[]byte("\x89\xa2\xc3\xc2\x9a\xe0F\x91\x06"),
 		Duration{Months: 1233, Days: 123213, Nanoseconds: 2312323},
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 5, typ: TypeDuration},
+		NativeType{proto: protoVersion5, typ: TypeDuration},
 		[]byte("\x89\xa1\xc3\xc2\x99\xe0F\x91\x05"),
 		Duration{Months: -1233, Days: -123213, Nanoseconds: -2312323},
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 5, typ: TypeDuration},
+		NativeType{proto: protoVersion5, typ: TypeDuration},
 		[]byte("\x02\x04\x80\xe6"),
 		Duration{Months: 1, Days: 2, Nanoseconds: 115},
 		nil,
@@ -406,73 +406,73 @@ var marshalTests = []struct {
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeList},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeList},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
-		[]byte("\x00\x02\x00\x04\x00\x00\x00\x01\x00\x04\x00\x00\x00\x02"),
+		[]byte("\x00\x00\x00\x02\x00\x00\x00\x04\x00\x00\x00\x01\x00\x00\x00\x04\x00\x00\x00\x02"),
 		[]int{1, 2},
 		nil,
 		nil,
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeList},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeList},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
-		[]byte("\x00\x02\x00\x04\x00\x00\x00\x01\x00\x04\x00\x00\x00\x02"),
+		[]byte("\x00\x00\x00\x02\x00\x00\x00\x04\x00\x00\x00\x01\x00\x00\x00\x04\x00\x00\x00\x02"),
 		[2]int{1, 2},
 		nil,
 		nil,
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeSet},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeSet},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
-		[]byte("\x00\x02\x00\x04\x00\x00\x00\x01\x00\x04\x00\x00\x00\x02"),
+		[]byte("\x00\x00\x00\x02\x00\x00\x00\x04\x00\x00\x00\x01\x00\x00\x00\x04\x00\x00\x00\x02"),
 		[]int{1, 2},
 		nil,
 		nil,
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeSet},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeSet},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
-		[]byte{0, 0}, // encoding of a list should always include the size of the collection
+		[]byte{0, 0, 0, 0}, // encoding of a list should always include the size of the collection
 		[]int{},
 		nil,
 		nil,
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeMap},
-			Key:        NativeType{proto: 2, typ: TypeVarchar},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeMap},
+			Key:        NativeType{proto: protoVersion3, typ: TypeVarchar},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
-		[]byte("\x00\x01\x00\x03foo\x00\x04\x00\x00\x00\x01"),
+		[]byte("\x00\x00\x00\x01\x00\x00\x00\x03foo\x00\x00\x00\x04\x00\x00\x00\x01"),
 		map[string]int{"foo": 1},
 		nil,
 		nil,
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeMap},
-			Key:        NativeType{proto: 2, typ: TypeVarchar},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeMap},
+			Key:        NativeType{proto: protoVersion3, typ: TypeVarchar},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
-		[]byte{0, 0},
+		[]byte{0, 0, 0, 0},
 		map[string]int{},
 		nil,
 		nil,
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeList},
-			Elem:       NativeType{proto: 2, typ: TypeVarchar},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeList},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeVarchar},
 		},
 		bytes.Join([][]byte{
-			[]byte("\x00\x01\xFF\xFF"),
+			[]byte("\x00\x00\x00\x01\x00\x00\xff\xff"),
 			bytes.Repeat([]byte("X"), math.MaxUint16)}, []byte("")),
 		[]string{strings.Repeat("X", math.MaxUint16)},
 		nil,
@@ -480,14 +480,14 @@ var marshalTests = []struct {
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeMap},
-			Key:        NativeType{proto: 2, typ: TypeVarchar},
-			Elem:       NativeType{proto: 2, typ: TypeVarchar},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeMap},
+			Key:        NativeType{proto: protoVersion3, typ: TypeVarchar},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeVarchar},
 		},
 		bytes.Join([][]byte{
-			[]byte("\x00\x01\xFF\xFF"),
+			[]byte("\x00\x00\x00\x01\x00\x00\xff\xff"),
 			bytes.Repeat([]byte("X"), math.MaxUint16),
-			[]byte("\xFF\xFF"),
+			[]byte("\x00\x00\xFF\xFF"),
 			bytes.Repeat([]byte("Y"), math.MaxUint16)}, []byte("")),
 		map[string]string{
 			strings.Repeat("X", math.MaxUint16): strings.Repeat("Y", math.MaxUint16),
@@ -496,119 +496,119 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarint},
+		NativeType{proto: protoVersion3, typ: TypeVarint},
 		[]byte("\x00"),
 		0,
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarint},
+		NativeType{proto: protoVersion3, typ: TypeVarint},
 		[]byte("\x37\xE2\x3C\xEC"),
 		int32(937573612),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarint},
+		NativeType{proto: protoVersion3, typ: TypeVarint},
 		[]byte("\x37\xE2\x3C\xEC"),
 		big.NewInt(937573612),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarint},
+		NativeType{proto: protoVersion3, typ: TypeVarint},
 		[]byte("\x03\x9EV \x15\f\x03\x9DK\x18\xCDI\\$?\a["),
 		bigintize("1231312312331283012830129382342342412123"), // From the iconara/cql-rb test suite
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarint},
+		NativeType{proto: protoVersion3, typ: TypeVarint},
 		[]byte("\xC9v\x8D:\x86"),
 		big.NewInt(-234234234234), // From the iconara/cql-rb test suite
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarint},
+		NativeType{proto: protoVersion3, typ: TypeVarint},
 		[]byte("f\x1e\xfd\xf2\xe3\xb1\x9f|\x04_\x15"),
 		bigintize("123456789123456789123456789"), // From the datastax/python-driver test suite
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarint},
+		NativeType{proto: protoVersion3, typ: TypeVarint},
 		[]byte(nil),
 		nil,
 		nil,
 		UnmarshalError("can not unmarshal into non-pointer <nil>"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeInet},
+		NativeType{proto: protoVersion3, typ: TypeInet},
 		[]byte("\x7F\x00\x00\x01"),
 		net.ParseIP("127.0.0.1").To4(),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInet},
+		NativeType{proto: protoVersion3, typ: TypeInet},
 		[]byte("\xFF\xFF\xFF\xFF"),
 		net.ParseIP("255.255.255.255").To4(),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInet},
+		NativeType{proto: protoVersion3, typ: TypeInet},
 		[]byte("\x7F\x00\x00\x01"),
 		"127.0.0.1",
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInet},
+		NativeType{proto: protoVersion3, typ: TypeInet},
 		[]byte("\xFF\xFF\xFF\xFF"),
 		"255.255.255.255",
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInet},
+		NativeType{proto: protoVersion3, typ: TypeInet},
 		[]byte("\x21\xDA\x00\xd3\x00\x00\x2f\x3b\x02\xaa\x00\xff\xfe\x28\x9c\x5a"),
 		"21da:d3:0:2f3b:2aa:ff:fe28:9c5a",
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInet},
+		NativeType{proto: protoVersion3, typ: TypeInet},
 		[]byte("\xfe\x80\x00\x00\x00\x00\x00\x00\x02\x02\xb3\xff\xfe\x1e\x83\x29"),
 		"fe80::202:b3ff:fe1e:8329",
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInet},
+		NativeType{proto: protoVersion3, typ: TypeInet},
 		[]byte("\x21\xDA\x00\xd3\x00\x00\x2f\x3b\x02\xaa\x00\xff\xfe\x28\x9c\x5a"),
 		net.ParseIP("21da:d3:0:2f3b:2aa:ff:fe28:9c5a"),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInet},
+		NativeType{proto: protoVersion3, typ: TypeInet},
 		[]byte("\xfe\x80\x00\x00\x00\x00\x00\x00\x02\x02\xb3\xff\xfe\x1e\x83\x29"),
 		net.ParseIP("fe80::202:b3ff:fe1e:8329"),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte(nil),
 		nil,
 		nil,
 		UnmarshalError("can not unmarshal into non-pointer <nil>"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarchar},
+		NativeType{proto: protoVersion3, typ: TypeVarchar},
 		[]byte("nullable string"),
 		func() *string {
 			value := "nullable string"
@@ -618,14 +618,14 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarchar},
+		NativeType{proto: protoVersion3, typ: TypeVarchar},
 		[]byte(nil),
 		(*string)(nil),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x7f\xff\xff\xff"),
 		func() *int {
 			var value int = math.MaxInt32
@@ -635,28 +635,28 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte(nil),
 		(*int)(nil),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTimeUUID},
+		NativeType{proto: protoVersion3, typ: TypeTimeUUID},
 		[]byte{0x3d, 0xcd, 0x98, 0x0, 0xf3, 0xd9, 0x11, 0xbf, 0x86, 0xd4, 0xb8, 0xe8, 0x56, 0x2c, 0xc, 0xd0},
 		&UUID{0x3d, 0xcd, 0x98, 0x0, 0xf3, 0xd9, 0x11, 0xbf, 0x86, 0xd4, 0xb8, 0xe8, 0x56, 0x2c, 0xc, 0xd0},
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTimeUUID},
+		NativeType{proto: protoVersion3, typ: TypeTimeUUID},
 		[]byte(nil),
 		(*UUID)(nil),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTimestamp},
+		NativeType{proto: protoVersion3, typ: TypeTimestamp},
 		[]byte("\x00\x00\x01\x40\x77\x16\xe1\xb8"),
 		func() *time.Time {
 			t := time.Date(2013, time.August, 13, 9, 52, 3, 0, time.UTC)
@@ -666,14 +666,14 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTimestamp},
+		NativeType{proto: protoVersion3, typ: TypeTimestamp},
 		[]byte(nil),
 		(*time.Time)(nil),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBoolean},
+		NativeType{proto: protoVersion3, typ: TypeBoolean},
 		[]byte("\x00"),
 		func() *bool {
 			b := false
@@ -683,7 +683,7 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBoolean},
+		NativeType{proto: protoVersion3, typ: TypeBoolean},
 		[]byte("\x01"),
 		func() *bool {
 			b := true
@@ -693,14 +693,14 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBoolean},
+		NativeType{proto: protoVersion3, typ: TypeBoolean},
 		[]byte(nil),
 		(*bool)(nil),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeFloat},
+		NativeType{proto: protoVersion3, typ: TypeFloat},
 		[]byte("\x40\x49\x0f\xdb"),
 		func() *float32 {
 			f := float32(3.14159265)
@@ -710,14 +710,14 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeFloat},
+		NativeType{proto: protoVersion3, typ: TypeFloat},
 		[]byte(nil),
 		(*float32)(nil),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDouble},
+		NativeType{proto: protoVersion3, typ: TypeDouble},
 		[]byte("\x40\x09\x21\xfb\x53\xc8\xd4\xf1"),
 		func() *float64 {
 			d := float64(3.14159265)
@@ -727,14 +727,14 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeDouble},
+		NativeType{proto: protoVersion3, typ: TypeDouble},
 		[]byte(nil),
 		(*float64)(nil),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInet},
+		NativeType{proto: protoVersion3, typ: TypeInet},
 		[]byte("\x7F\x00\x00\x01"),
 		func() *net.IP {
 			ip := net.ParseIP("127.0.0.1").To4()
@@ -744,7 +744,7 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInet},
+		NativeType{proto: protoVersion3, typ: TypeInet},
 		[]byte(nil),
 		(*net.IP)(nil),
 		nil,
@@ -752,10 +752,10 @@ var marshalTests = []struct {
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeList},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeList},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
-		[]byte("\x00\x02\x00\x04\x00\x00\x00\x01\x00\x04\x00\x00\x00\x02"),
+		[]byte("\x00\x00\x00\x02\x00\x00\x00\x04\x00\x00\x00\x01\x00\x00\x00\x04\x00\x00\x00\x02"),
 		func() *[]int {
 			l := []int{1, 2}
 			return &l
@@ -765,8 +765,8 @@ var marshalTests = []struct {
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 3, typ: TypeList},
-			Elem:       NativeType{proto: 3, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeList},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
 		[]byte("\x00\x00\x00\x02\x00\x00\x00\x04\x00\x00\x00\x01\x00\x00\x00\x04\x00\x00\x00\x02"),
 		func() *[]int {
@@ -778,8 +778,8 @@ var marshalTests = []struct {
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeList},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeList},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
 		[]byte(nil),
 		(*[]int)(nil),
@@ -788,11 +788,11 @@ var marshalTests = []struct {
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeMap},
-			Key:        NativeType{proto: 2, typ: TypeVarchar},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeMap},
+			Key:        NativeType{proto: protoVersion3, typ: TypeVarchar},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
-		[]byte("\x00\x01\x00\x03foo\x00\x04\x00\x00\x00\x01"),
+		[]byte("\x00\x00\x00\x01\x00\x00\x00\x03foo\x00\x00\x00\x04\x00\x00\x00\x01"),
 		func() *map[string]int {
 			m := map[string]int{"foo": 1}
 			return &m
@@ -802,9 +802,9 @@ var marshalTests = []struct {
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeMap},
-			Key:        NativeType{proto: 2, typ: TypeVarchar},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeMap},
+			Key:        NativeType{proto: protoVersion3, typ: TypeVarchar},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
 		[]byte(nil),
 		(*map[string]int)(nil),
@@ -812,7 +812,7 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarchar},
+		NativeType{proto: protoVersion3, typ: TypeVarchar},
 		[]byte("HELLO WORLD"),
 		func() *CustomString {
 			customString := CustomString("hello world")
@@ -822,252 +822,252 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarchar},
+		NativeType{proto: protoVersion3, typ: TypeVarchar},
 		[]byte(nil),
 		(*CustomString)(nil),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\x7f\xff"),
 		32767, // math.MaxInt16
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\x7f\xff"),
 		"32767", // math.MaxInt16
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\x00\x01"),
 		int16(1),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\xff\xff"),
 		int16(-1),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\x00\xff"),
 		uint8(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\xff\xff"),
 		uint16(65535),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\xff\xff"),
 		uint32(65535),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\xff\xff"),
 		uint64(65535),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\x00\xff"),
 		AliasUint8(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\xff\xff"),
 		AliasUint16(65535),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\xff\xff"),
 		AliasUint32(65535),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\xff\xff"),
 		AliasUint64(65535),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\xff\xff"),
 		AliasUint(65535),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\x7f"),
 		127, // math.MaxInt8
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\x7f"),
 		"127", // math.MaxInt8
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\x01"),
 		int16(1),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\xff"),
 		int16(-1),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\xff"),
 		uint8(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\xff"),
 		uint64(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\xff"),
 		uint32(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\xff"),
 		uint16(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\xff"),
 		uint(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\xff"),
 		AliasUint8(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\xff"),
 		AliasUint64(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\xff"),
 		AliasUint32(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\xff"),
 		AliasUint16(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTinyInt},
+		NativeType{proto: protoVersion3, typ: TypeTinyInt},
 		[]byte("\xff"),
 		AliasUint(255),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x00\x00\x00\x00\xff"),
 		uint8(math.MaxUint8),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x00\x00\x00\xff\xff"),
 		uint64(math.MaxUint16),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x00\xff\xff\xff\xff"),
 		uint64(math.MaxUint32),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\xff\xff\xff\xff\xff\xff\xff\xff"),
 		uint64(math.MaxUint64),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\xff\xff\xff\xff"),
 		uint32(math.MaxUint32),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\xff\xff\xff\xff"),
 		uint64(math.MaxUint32),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeBlob},
+		NativeType{proto: protoVersion3, typ: TypeBlob},
 		[]byte(nil),
 		([]byte)(nil),
 		nil,
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeVarchar},
+		NativeType{proto: protoVersion3, typ: TypeVarchar},
 		[]byte{},
 		func() interface{} {
 			var s string
@@ -1077,7 +1077,7 @@ var marshalTests = []struct {
 		nil,
 	},
 	{
-		NativeType{proto: 2, typ: TypeTime},
+		NativeType{proto: protoVersion3, typ: TypeTime},
 		encBigInt(1000),
 		time.Duration(1000),
 		nil,
@@ -1092,177 +1092,177 @@ var unmarshalTests = []struct {
 	UnmarshalError error
 }{
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\xff\xff"),
 		uint8(0),
 		UnmarshalError("unmarshal int: value -1 out of range for uint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\x01\x00"),
 		uint8(0),
 		UnmarshalError("unmarshal int: value 256 out of range for uint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\xff\xff\xff\xff"),
 		uint8(0),
 		UnmarshalError("unmarshal int: value -1 out of range for uint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x00\x00\x01\x00"),
 		uint8(0),
 		UnmarshalError("unmarshal int: value 256 out of range for uint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\xff\xff\xff\xff"),
 		uint16(0),
 		UnmarshalError("unmarshal int: value -1 out of range for uint16"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x00\x01\x00\x00"),
 		uint16(0),
 		UnmarshalError("unmarshal int: value 65536 out of range for uint16"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\xff\xff\xff\xff\xff\xff\xff\xff"),
 		uint8(0),
 		UnmarshalError("unmarshal int: value -1 out of range for uint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x00\x00\x00\x01\x00"),
 		uint8(0),
 		UnmarshalError("unmarshal int: value 256 out of range for uint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\xff\xff\xff\xff\xff\xff\xff\xff"),
 		uint8(0),
 		UnmarshalError("unmarshal int: value -1 out of range for uint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x00\x00\x00\x01\x00"),
 		uint8(0),
 		UnmarshalError("unmarshal int: value 256 out of range for uint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\xff\xff\xff\xff\xff\xff\xff\xff"),
 		uint16(0),
 		UnmarshalError("unmarshal int: value -1 out of range for uint16"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x00\x00\x01\x00\x00"),
 		uint16(0),
 		UnmarshalError("unmarshal int: value 65536 out of range for uint16"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\xff\xff\xff\xff\xff\xff\xff\xff"),
 		uint32(0),
 		UnmarshalError("unmarshal int: value -1 out of range for uint32"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x01\x00\x00\x00\x00"),
 		uint32(0),
 		UnmarshalError("unmarshal int: value 4294967296 out of range for uint32"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\xff\xff"),
 		AliasUint8(0),
 		UnmarshalError("unmarshal int: value -1 out of range for gocql.AliasUint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeSmallInt},
+		NativeType{proto: protoVersion3, typ: TypeSmallInt},
 		[]byte("\x01\x00"),
 		AliasUint8(0),
 		UnmarshalError("unmarshal int: value 256 out of range for gocql.AliasUint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\xff\xff\xff\xff"),
 		AliasUint8(0),
 		UnmarshalError("unmarshal int: value -1 out of range for gocql.AliasUint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x00\x00\x01\x00"),
 		AliasUint8(0),
 		UnmarshalError("unmarshal int: value 256 out of range for gocql.AliasUint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\xff\xff\xff\xff"),
 		AliasUint16(0),
 		UnmarshalError("unmarshal int: value -1 out of range for gocql.AliasUint16"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeInt},
+		NativeType{proto: protoVersion3, typ: TypeInt},
 		[]byte("\x00\x01\x00\x00"),
 		AliasUint16(0),
 		UnmarshalError("unmarshal int: value 65536 out of range for gocql.AliasUint16"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\xff\xff\xff\xff\xff\xff\xff\xff"),
 		AliasUint8(0),
 		UnmarshalError("unmarshal int: value -1 out of range for gocql.AliasUint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x00\x00\x00\x01\x00"),
 		AliasUint8(0),
 		UnmarshalError("unmarshal int: value 256 out of range for gocql.AliasUint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\xff\xff\xff\xff\xff\xff\xff\xff"),
 		AliasUint8(0),
 		UnmarshalError("unmarshal int: value -1 out of range for gocql.AliasUint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x00\x00\x00\x01\x00"),
 		AliasUint8(0),
 		UnmarshalError("unmarshal int: value 256 out of range for gocql.AliasUint8"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\xff\xff\xff\xff\xff\xff\xff\xff"),
 		AliasUint16(0),
 		UnmarshalError("unmarshal int: value -1 out of range for gocql.AliasUint16"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x00\x00\x01\x00\x00"),
 		AliasUint16(0),
 		UnmarshalError("unmarshal int: value 65536 out of range for gocql.AliasUint16"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\xff\xff\xff\xff\xff\xff\xff\xff"),
 		AliasUint32(0),
 		UnmarshalError("unmarshal int: value -1 out of range for gocql.AliasUint32"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeBigInt},
+		NativeType{proto: protoVersion3, typ: TypeBigInt},
 		[]byte("\x00\x00\x00\x01\x00\x00\x00\x00"),
 		AliasUint32(0),
 		UnmarshalError("unmarshal int: value 4294967296 out of range for gocql.AliasUint32"),
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 3, typ: TypeList},
-			Elem:       NativeType{proto: 3, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeList},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
 		[]byte("\x00\x00\x00\x02\x00\x00\x00\x04\x00\x00"), // truncated data
 		func() *[]int {
@@ -1273,56 +1273,56 @@ var unmarshalTests = []struct {
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeMap},
-			Key:        NativeType{proto: 2, typ: TypeVarchar},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeMap},
+			Key:        NativeType{proto: protoVersion3, typ: TypeVarchar},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
-		[]byte("\x00\x01\x00\x03fo"),
+		[]byte("\x00\x00\x00\x01\x00\x00\x00\x03fo"),
 		map[string]int{"foo": 1},
 		UnmarshalError("unmarshal map: unexpected eof"),
 	},
 	{
 		CollectionType{
-			NativeType: NativeType{proto: 2, typ: TypeMap},
-			Key:        NativeType{proto: 2, typ: TypeVarchar},
-			Elem:       NativeType{proto: 2, typ: TypeInt},
+			NativeType: NativeType{proto: protoVersion3, typ: TypeMap},
+			Key:        NativeType{proto: protoVersion3, typ: TypeVarchar},
+			Elem:       NativeType{proto: protoVersion3, typ: TypeInt},
 		},
-		[]byte("\x00\x01\x00\x03foo\x00\x04\x00\x00"),
+		[]byte("\x00\x00\x00\x01\x00\x00\x00\x03foo\x00\x04\x00\x00"),
 		map[string]int{"foo": 1},
 		UnmarshalError("unmarshal map: unexpected eof"),
 	},
 	{
-		NativeType{proto: 2, typ: TypeDecimal},
+		NativeType{proto: protoVersion3, typ: TypeDecimal},
 		[]byte("\xff\xff\xff"),
 		inf.NewDec(0, 0), // From the datastax/python-driver test suite
 		UnmarshalError("inf.Dec needs at least 4 bytes, while value has only 3"),
 	},
 	{
-		NativeType{proto: 5, typ: TypeDuration},
+		NativeType{proto: protoVersion5, typ: TypeDuration},
 		[]byte("\x89\xa2\xc3\xc2\x9a\xe0F\x91"),
 		Duration{},
 		UnmarshalError("failed to unmarshal duration into *gocql.Duration: failed to extract nanoseconds: data expect to have 9 bytes, but it has only 8"),
 	},
 	{
-		NativeType{proto: 5, typ: TypeDuration},
+		NativeType{proto: protoVersion5, typ: TypeDuration},
 		[]byte("\x89\xa2\xc3\xc2\x9a"),
 		Duration{},
 		UnmarshalError("failed to unmarshal duration into *gocql.Duration: failed to extract nanoseconds: unexpected eof"),
 	},
 	{
-		NativeType{proto: 5, typ: TypeDuration},
+		NativeType{proto: protoVersion5, typ: TypeDuration},
 		[]byte("\x89\xa2\xc3\xc2"),
 		Duration{},
 		UnmarshalError("failed to unmarshal duration into *gocql.Duration: failed to extract days: data expect to have 5 bytes, but it has only 4"),
 	},
 	{
-		NativeType{proto: 5, typ: TypeDuration},
+		NativeType{proto: protoVersion5, typ: TypeDuration},
 		[]byte("\x89\xa2"),
 		Duration{},
 		UnmarshalError("failed to unmarshal duration into *gocql.Duration: failed to extract days: unexpected eof"),
 	},
 	{
-		NativeType{proto: 5, typ: TypeDuration},
+		NativeType{proto: protoVersion5, typ: TypeDuration},
 		[]byte("\x89"),
 		Duration{},
 		UnmarshalError("failed to unmarshal duration into *gocql.Duration: failed to extract month: data expect to have 2 bytes, but it has only 1"),
@@ -1459,7 +1459,7 @@ func TestMarshalVarint(t *testing.T) {
 	}
 
 	for i, test := range varintTests {
-		data, err := Marshal(NativeType{proto: 2, typ: TypeVarint}, test.Value)
+		data, err := Marshal(NativeType{proto: protoVersion3, typ: TypeVarint}, test.Value)
 		if err != nil {
 			t.Errorf("error marshaling varint: %v (test #%d)", err, i)
 		}
@@ -1469,7 +1469,7 @@ func TestMarshalVarint(t *testing.T) {
 		}
 
 		binder := new(big.Int)
-		err = Unmarshal(NativeType{proto: 2, typ: TypeVarint}, test.Marshaled, binder)
+		err = Unmarshal(NativeType{proto: protoVersion3, typ: TypeVarint}, test.Marshaled, binder)
 		if err != nil {
 			t.Errorf("error unmarshaling varint: %v (test #%d)", err, i)
 		}
@@ -1517,7 +1517,7 @@ func TestMarshalVarint(t *testing.T) {
 	}
 
 	for i, test := range varintUint64Tests {
-		data, err := Marshal(NativeType{proto: 2, typ: TypeVarint}, test.Value)
+		data, err := Marshal(NativeType{proto: protoVersion3, typ: TypeVarint}, test.Value)
 		if err != nil {
 			t.Errorf("error marshaling varint: %v (test #%d)", err, i)
 		}
@@ -1527,7 +1527,7 @@ func TestMarshalVarint(t *testing.T) {
 		}
 
 		var binder uint64
-		err = Unmarshal(NativeType{proto: 2, typ: TypeVarint}, test.Marshaled, &binder)
+		err = Unmarshal(NativeType{proto: protoVersion3, typ: TypeVarint}, test.Marshaled, &binder)
 		if err != nil {
 			t.Errorf("error unmarshaling varint to uint64: %v (test #%d)", err, i)
 		}
@@ -1545,12 +1545,12 @@ func TestMarshalBigInt(t *testing.T) {
 		MarshalError error
 	}{
 		{
-			NativeType{proto: 2, typ: TypeBigInt},
+			NativeType{proto: protoVersion3, typ: TypeBigInt},
 			"-78635384813432117863538481343211",
 			MarshalError("can not marshal string to bigint: strconv.ParseInt: parsing \"-78635384813432117863538481343211\": value out of range"),
 		},
 		{
-			NativeType{proto: 2, typ: TypeBigInt},
+			NativeType{proto: protoVersion3, typ: TypeBigInt},
 			"922337203685477692259749625974294",
 			MarshalError("can not marshal string to bigint: strconv.ParseInt: parsing \"922337203685477692259749625974294\": value out of range"),
 		},
@@ -1577,13 +1577,9 @@ func equalStringPointerSlice(leftList, rightList []*string) bool {
 }
 
 func TestMarshalList(t *testing.T) {
-	typeInfoV2 := CollectionType{
-		NativeType: NativeType{proto: 2, typ: TypeList},
-		Elem:       NativeType{proto: 2, typ: TypeVarchar},
-	}
 	typeInfoV3 := CollectionType{
-		NativeType: NativeType{proto: 3, typ: TypeList},
-		Elem:       NativeType{proto: 3, typ: TypeVarchar},
+		NativeType: NativeType{proto: protoVersion3, typ: TypeList},
+		Elem:       NativeType{proto: protoVersion3, typ: TypeVarchar},
 	}
 
 	type tc struct {
@@ -1596,37 +1592,6 @@ func TestMarshalList(t *testing.T) {
 	valueB := "valueB"
 	valueEmpty := ""
 	testCases := []tc{
-		{
-			typeInfo: typeInfoV2,
-			input:    []*string{&valueA},
-			expected: []*string{&valueA},
-		},
-		{
-			typeInfo: typeInfoV2,
-			input:    []*string{&valueA, &valueB},
-			expected: []*string{&valueA, &valueB},
-		},
-		{
-			typeInfo: typeInfoV2,
-			input:    []*string{&valueA, &valueEmpty, &valueB},
-			expected: []*string{&valueA, &valueEmpty, &valueB},
-		},
-		{
-			typeInfo: typeInfoV2,
-			input:    []*string{&valueEmpty},
-			expected: []*string{&valueEmpty},
-		},
-		{
-			// nil values are marshalled to empty values for protocol < 3
-			typeInfo: typeInfoV2,
-			input:    []*string{nil},
-			expected: []*string{&valueEmpty},
-		},
-		{
-			typeInfo: typeInfoV2,
-			input:    []*string{&valueA, nil, &valueB},
-			expected: []*string{&valueA, &valueEmpty, &valueB},
-		},
 		{
 			typeInfo: typeInfoV3,
 			input:    []*string{&valueEmpty},
@@ -1740,7 +1705,7 @@ func (m *MyPointerMarshaler) MarshalCQL(_ TypeInfo) ([]byte, error) {
 
 func TestMarshalPointer(t *testing.T) {
 	m := &MyPointerMarshaler{}
-	typ := NativeType{proto: 2, typ: TypeInt}
+	typ := NativeType{proto: protoVersion3, typ: TypeInt}
 
 	data, err := Marshal(typ, m)
 
@@ -1762,17 +1727,17 @@ func TestMarshalTime(t *testing.T) {
 		Value interface{}
 	}{
 		{
-			NativeType{proto: 4, typ: TypeTime},
+			NativeType{proto: protoVersion4, typ: TypeTime},
 			expectedData,
 			duration.Nanoseconds(),
 		},
 		{
-			NativeType{proto: 4, typ: TypeTime},
+			NativeType{proto: protoVersion4, typ: TypeTime},
 			expectedData,
 			duration,
 		},
 		{
-			NativeType{proto: 4, typ: TypeTime},
+			NativeType{proto: protoVersion4, typ: TypeTime},
 			expectedData,
 			&duration,
 		},
@@ -1799,46 +1764,46 @@ func TestMarshalTimestamp(t *testing.T) {
 		Value interface{}
 	}{
 		{
-			NativeType{proto: 2, typ: TypeTimestamp},
+			NativeType{proto: protoVersion3, typ: TypeTimestamp},
 			[]byte("\x00\x00\x01\x40\x77\x16\xe1\xb8"),
 			time.Date(2013, time.August, 13, 9, 52, 3, 0, time.UTC),
 		},
 		{
-			NativeType{proto: 2, typ: TypeTimestamp},
+			NativeType{proto: protoVersion3, typ: TypeTimestamp},
 			[]byte("\x00\x00\x01\x40\x77\x16\xe1\xb8"),
 			int64(1376387523000),
 		},
 		{
 			// 9223372036854 is the maximum time representable in ms since the epoch
 			// with int64 if using UnixNano to convert
-			NativeType{proto: 2, typ: TypeTimestamp},
+			NativeType{proto: protoVersion3, typ: TypeTimestamp},
 			[]byte("\x00\x00\x08\x63\x7b\xd0\x5a\xf6"),
 			time.Date(2262, time.April, 11, 23, 47, 16, 854775807, time.UTC),
 		},
 		{
 			// One nanosecond after causes overflow when using UnixNano
 			// Instead it should resolve to the same time in ms
-			NativeType{proto: 2, typ: TypeTimestamp},
+			NativeType{proto: protoVersion3, typ: TypeTimestamp},
 			[]byte("\x00\x00\x08\x63\x7b\xd0\x5a\xf6"),
 			time.Date(2262, time.April, 11, 23, 47, 16, 854775808, time.UTC),
 		},
 		{
 			// -9223372036855 is the minimum time representable in ms since the epoch
 			// with int64 if using UnixNano to convert
-			NativeType{proto: 2, typ: TypeTimestamp},
+			NativeType{proto: protoVersion3, typ: TypeTimestamp},
 			[]byte("\xff\xff\xf7\x9c\x84\x2f\xa5\x09"),
 			time.Date(1677, time.September, 21, 00, 12, 43, 145224192, time.UTC),
 		},
 		{
 			// One nanosecond earlier causes overflow when using UnixNano
 			// it should resolve to the same time in ms
-			NativeType{proto: 2, typ: TypeTimestamp},
+			NativeType{proto: protoVersion3, typ: TypeTimestamp},
 			[]byte("\xff\xff\xf7\x9c\x84\x2f\xa5\x09"),
 			time.Date(1677, time.September, 21, 00, 12, 43, 145224191, time.UTC),
 		},
 		{
 			// Store the zero time as a blank slice
-			NativeType{proto: 2, typ: TypeTimestamp},
+			NativeType{proto: protoVersion3, typ: TypeTimestamp},
 			[]byte{},
 			time.Time{},
 		},
@@ -1860,10 +1825,10 @@ func TestMarshalTimestamp(t *testing.T) {
 
 func TestMarshalTuple(t *testing.T) {
 	info := TupleTypeInfo{
-		NativeType: NativeType{proto: 3, typ: TypeTuple},
+		NativeType: NativeType{proto: protoVersion3, typ: TypeTuple},
 		Elems: []TypeInfo{
-			NativeType{proto: 3, typ: TypeVarchar},
-			NativeType{proto: 3, typ: TypeVarchar},
+			NativeType{proto: protoVersion3, typ: TypeVarchar},
+			NativeType{proto: protoVersion3, typ: TypeVarchar},
 		},
 	}
 
@@ -2007,10 +1972,10 @@ func TestMarshalTuple(t *testing.T) {
 
 func TestUnmarshalTuple(t *testing.T) {
 	info := TupleTypeInfo{
-		NativeType: NativeType{proto: 3, typ: TypeTuple},
+		NativeType: NativeType{proto: protoVersion3, typ: TypeTuple},
 		Elems: []TypeInfo{
-			NativeType{proto: 3, typ: TypeVarchar},
-			NativeType{proto: 3, typ: TypeVarchar},
+			NativeType{proto: protoVersion3, typ: TypeVarchar},
+			NativeType{proto: protoVersion3, typ: TypeVarchar},
 		},
 	}
 
@@ -2081,10 +2046,10 @@ func TestUnmarshalTuple(t *testing.T) {
 }
 
 func TestMarshalUDTMap(t *testing.T) {
-	typeInfo := UDTTypeInfo{NativeType{proto: 3, typ: TypeUDT}, "", "xyz", []UDTField{
-		{Name: "x", Type: NativeType{proto: 3, typ: TypeInt}},
-		{Name: "y", Type: NativeType{proto: 3, typ: TypeInt}},
-		{Name: "z", Type: NativeType{proto: 3, typ: TypeInt}},
+	typeInfo := UDTTypeInfo{NativeType{proto: protoVersion3, typ: TypeUDT}, "", "xyz", []UDTField{
+		{Name: "x", Type: NativeType{proto: protoVersion3, typ: TypeInt}},
+		{Name: "y", Type: NativeType{proto: protoVersion3, typ: TypeInt}},
+		{Name: "z", Type: NativeType{proto: protoVersion3, typ: TypeInt}},
 	}}
 
 	t.Run("partially bound", func(t *testing.T) {
@@ -2136,10 +2101,10 @@ func TestMarshalUDTMap(t *testing.T) {
 }
 
 func TestMarshalUDTStruct(t *testing.T) {
-	typeInfo := UDTTypeInfo{NativeType{proto: 3, typ: TypeUDT}, "", "xyz", []UDTField{
-		{Name: "x", Type: NativeType{proto: 3, typ: TypeInt}},
-		{Name: "y", Type: NativeType{proto: 3, typ: TypeInt}},
-		{Name: "z", Type: NativeType{proto: 3, typ: TypeInt}},
+	typeInfo := UDTTypeInfo{NativeType{proto: protoVersion3, typ: TypeUDT}, "", "xyz", []UDTField{
+		{Name: "x", Type: NativeType{proto: protoVersion3, typ: TypeInt}},
+		{Name: "y", Type: NativeType{proto: protoVersion3, typ: TypeInt}},
+		{Name: "z", Type: NativeType{proto: protoVersion3, typ: TypeInt}},
 	}}
 
 	type xyzStruct struct {
@@ -2224,7 +2189,7 @@ func TestMarshalNil(t *testing.T) {
 	}
 
 	for _, typ := range types {
-		data, err := Marshal(NativeType{proto: 3, typ: typ}, nil)
+		data, err := Marshal(NativeType{proto: protoVersion3, typ: typ}, nil)
 		if err != nil {
 			t.Errorf("unable to marshal nil %v: %v\n", typ, err)
 		} else if data != nil {
@@ -2236,7 +2201,7 @@ func TestMarshalNil(t *testing.T) {
 func TestUnmarshalInetCopyBytes(t *testing.T) {
 	data := []byte{127, 0, 0, 1}
 	var ip net.IP
-	if err := unmarshalInet(NativeType{proto: 2, typ: TypeInet}, data, &ip); err != nil {
+	if err := unmarshalInet(NativeType{proto: protoVersion3, typ: TypeInet}, data, &ip); err != nil {
 		t.Fatal(err)
 	}
 
@@ -2250,7 +2215,7 @@ func TestUnmarshalInetCopyBytes(t *testing.T) {
 func TestUnmarshalDate(t *testing.T) {
 	data := []uint8{0x80, 0x0, 0x43, 0x31}
 	var date time.Time
-	if err := unmarshalDate(NativeType{proto: 2, typ: TypeDate}, data, &date); err != nil {
+	if err := unmarshalDate(NativeType{proto: protoVersion3, typ: TypeDate}, data, &date); err != nil {
 		t.Fatal(err)
 	}
 
@@ -2261,7 +2226,7 @@ func TestUnmarshalDate(t *testing.T) {
 		return
 	}
 	var stringDate string
-	if err2 := unmarshalDate(NativeType{proto: 2, typ: TypeDate}, data, &stringDate); err2 != nil {
+	if err2 := unmarshalDate(NativeType{proto: protoVersion3, typ: TypeDate}, data, &stringDate); err2 != nil {
 		t.Fatal(err2)
 	}
 	if expectedDate != stringDate {
@@ -2281,22 +2246,22 @@ func TestMarshalDate(t *testing.T) {
 		Value interface{}
 	}{
 		{
-			NativeType{proto: 4, typ: TypeDate},
+			NativeType{proto: protoVersion4, typ: TypeDate},
 			expectedData,
 			timestamp,
 		},
 		{
-			NativeType{proto: 4, typ: TypeDate},
+			NativeType{proto: protoVersion4, typ: TypeDate},
 			expectedData,
 			now,
 		},
 		{
-			NativeType{proto: 4, typ: TypeDate},
+			NativeType{proto: protoVersion4, typ: TypeDate},
 			expectedData,
 			&now,
 		},
 		{
-			NativeType{proto: 4, typ: TypeDate},
+			NativeType{proto: protoVersion4, typ: TypeDate},
 			expectedData,
 			now.Format("2006-01-02"),
 		},
@@ -2340,7 +2305,7 @@ func TestLargeDate(t *testing.T) {
 		},
 	}
 
-	nativeType := NativeType{proto: 4, typ: TypeDate}
+	nativeType := NativeType{proto: protoVersion4, typ: TypeDate}
 
 	for i, test := range marshalDateTests {
 		t.Log(i, test)
@@ -2390,22 +2355,22 @@ func TestMarshalDuration(t *testing.T) {
 		Value interface{}
 	}{
 		{
-			NativeType{proto: 5, typ: TypeDuration},
+			NativeType{proto: protoVersion5, typ: TypeDuration},
 			expectedData,
 			duration.Nanoseconds(),
 		},
 		{
-			NativeType{proto: 5, typ: TypeDuration},
+			NativeType{proto: protoVersion5, typ: TypeDuration},
 			expectedData,
 			duration,
 		},
 		{
-			NativeType{proto: 5, typ: TypeDuration},
+			NativeType{proto: protoVersion5, typ: TypeDuration},
 			expectedData,
 			durationS,
 		},
 		{
-			NativeType{proto: 5, typ: TypeDuration},
+			NativeType{proto: protoVersion5, typ: TypeDuration},
 			expectedData,
 			&duration,
 		},
@@ -2426,13 +2391,9 @@ func TestMarshalDuration(t *testing.T) {
 }
 
 func TestReadCollectionSize(t *testing.T) {
-	listV2 := CollectionType{
-		NativeType: NativeType{proto: 2, typ: TypeList},
-		Elem:       NativeType{proto: 2, typ: TypeVarchar},
-	}
 	listV3 := CollectionType{
-		NativeType: NativeType{proto: 3, typ: TypeList},
-		Elem:       NativeType{proto: 3, typ: TypeVarchar},
+		NativeType: NativeType{proto: protoVersion3, typ: TypeList},
+		Elem:       NativeType{proto: protoVersion3, typ: TypeVarchar},
 	}
 
 	tests := []struct {
@@ -2442,24 +2403,6 @@ func TestReadCollectionSize(t *testing.T) {
 		isError      bool
 		expectedSize int
 	}{
-		{
-			name:    "short read 0 proto 2",
-			info:    listV2,
-			data:    []byte{},
-			isError: true,
-		},
-		{
-			name:    "short read 1 proto 2",
-			info:    listV2,
-			data:    []byte{0x01},
-			isError: true,
-		},
-		{
-			name:         "good read proto 2",
-			info:         listV2,
-			data:         []byte{0x01, 0x38},
-			expectedSize: 0x0138,
-		},
 		{
 			name:    "short read 0 proto 3",
 			info:    listV3,
@@ -2493,7 +2436,7 @@ func TestReadCollectionSize(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			size, _, err := readCollectionSize(test.info, test.data)
+			size, _, err := readCollectionSize(test.data)
 			if test.isError {
 				if err == nil {
 					t.Fatal("Expected error, but it was nil")
@@ -2557,17 +2500,17 @@ func BenchmarkUnmarshalUUID(b *testing.B) {
 
 func TestUnmarshalUDT(t *testing.T) {
 	info := UDTTypeInfo{
-		NativeType: NativeType{proto: 4, typ: TypeUDT},
+		NativeType: NativeType{proto: protoVersion4, typ: TypeUDT},
 		Name:       "myudt",
 		KeySpace:   "myks",
 		Elements: []UDTField{
 			{
 				Name: "first",
-				Type: NativeType{proto: 4, typ: TypeAscii},
+				Type: NativeType{proto: protoVersion4, typ: TypeAscii},
 			},
 			{
 				Name: "second",
-				Type: NativeType{proto: 4, typ: TypeSmallInt},
+				Type: NativeType{proto: protoVersion4, typ: TypeSmallInt},
 			},
 		},
 	}
diff --git a/metadata_test.go b/metadata_test.go
index 6b3d119..21829b3 100644
--- a/metadata_test.go
+++ b/metadata_test.go
@@ -38,310 +38,11 @@ import (
 func TestCompileMetadata(t *testing.T) {
 	// V1 tests - these are all based on real examples from the integration test ccm cluster
 	log := &defaultLogger{}
-	keyspace := &KeyspaceMetadata{
-		Name: "V1Keyspace",
-	}
-	tables := []TableMetadata{
-		{
-			// This table, found in the system keyspace, has no key aliases or column aliases
-			Keyspace:         "V1Keyspace",
-			Name:             "Schema",
-			KeyValidator:     "org.apache.cassandra.db.marshal.BytesType",
-			Comparator:       "org.apache.cassandra.db.marshal.UTF8Type",
-			DefaultValidator: "org.apache.cassandra.db.marshal.BytesType",
-			KeyAliases:       []string{},
-			ColumnAliases:    []string{},
-			ValueAlias:       "",
-		},
-		{
-			// This table, found in the system keyspace, has key aliases, column aliases, and a value alias.
-			Keyspace:         "V1Keyspace",
-			Name:             "hints",
-			KeyValidator:     "org.apache.cassandra.db.marshal.UUIDType",
-			Comparator:       "org.apache.cassandra.db.marshal.CompositeType(org.apache.cassandra.db.marshal.TimeUUIDType,org.apache.cassandra.db.marshal.Int32Type)",
-			DefaultValidator: "org.apache.cassandra.db.marshal.BytesType",
-			KeyAliases:       []string{"target_id"},
-			ColumnAliases:    []string{"hint_id", "message_version"},
-			ValueAlias:       "mutation",
-		},
-		{
-			// This table, found in the system keyspace, has a comparator with collections, but no column aliases
-			Keyspace:         "V1Keyspace",
-			Name:             "peers",
-			KeyValidator:     "org.apache.cassandra.db.marshal.InetAddressType",
-			Comparator:       "org.apache.cassandra.db.marshal.CompositeType(org.apache.cassandra.db.marshal.UTF8Type,org.apache.cassandra.db.marshal.ColumnToCollectionType(746f6b656e73:org.apache.cassandra.db.marshal.SetType(org.apache.cassandra.db.marshal.UTF8Type)))",
-			DefaultValidator: "org.apache.cassandra.db.marshal.BytesType",
-			KeyAliases:       []string{"peer"},
-			ColumnAliases:    []string{},
-			ValueAlias:       "",
-		},
-		{
-			// This table, found in the system keyspace, has a column alias, but not a composite comparator
-			Keyspace:         "V1Keyspace",
-			Name:             "IndexInfo",
-			KeyValidator:     "org.apache.cassandra.db.marshal.UTF8Type",
-			Comparator:       "org.apache.cassandra.db.marshal.ReversedType(org.apache.cassandra.db.marshal.UTF8Type)",
-			DefaultValidator: "org.apache.cassandra.db.marshal.BytesType",
-			KeyAliases:       []string{"table_name"},
-			ColumnAliases:    []string{"index_name"},
-			ValueAlias:       "",
-		},
-		{
-			// This table, found in the gocql_test keyspace following an integration test run, has a composite comparator with collections as well as a column alias
-			Keyspace:         "V1Keyspace",
-			Name:             "wiki_page",
-			KeyValidator:     "org.apache.cassandra.db.marshal.UTF8Type",
-			Comparator:       "org.apache.cassandra.db.marshal.CompositeType(org.apache.cassandra.db.marshal.TimeUUIDType,org.apache.cassandra.db.marshal.UTF8Type,org.apache.cassandra.db.marshal.ColumnToCollectionType(74616773:org.apache.cassandra.db.marshal.SetType(org.apache.cassandra.db.marshal.UTF8Type),6174746163686d656e7473:org.apache.cassandra.db.marshal.MapType(org.apache.cassandra.db.marshal.UTF8Type,org.apache.cassandra.db.marshal.BytesType)))",
-			DefaultValidator: "org.apache.cassandra.db.marshal.BytesType",
-			KeyAliases:       []string{"title"},
-			ColumnAliases:    []string{"revid"},
-			ValueAlias:       "",
-		},
-		{
-			// This is a made up example with multiple unnamed aliases
-			Keyspace:         "V1Keyspace",
-			Name:             "no_names",
-			KeyValidator:     "org.apache.cassandra.db.marshal.CompositeType(org.apache.cassandra.db.marshal.UUIDType,org.apache.cassandra.db.marshal.UUIDType)",
-			Comparator:       "org.apache.cassandra.db.marshal.CompositeType(org.apache.cassandra.db.marshal.Int32Type,org.apache.cassandra.db.marshal.Int32Type,org.apache.cassandra.db.marshal.Int32Type)",
-			DefaultValidator: "org.apache.cassandra.db.marshal.BytesType",
-			KeyAliases:       []string{},
-			ColumnAliases:    []string{},
-			ValueAlias:       "",
-		},
-	}
-	columns := []ColumnMetadata{
-		// Here are the regular columns from the peers table for testing regular columns
-		{Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "data_center", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UTF8Type"},
-		{Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "host_id", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UUIDType"},
-		{Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "rack", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UTF8Type"},
-		{Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "release_version", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UTF8Type"},
-		{Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "rpc_address", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.InetAddressType"},
-		{Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "schema_version", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UUIDType"},
-		{Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "tokens", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.SetType(org.apache.cassandra.db.marshal.UTF8Type)"},
-	}
-	compileMetadata(1, keyspace, tables, columns, nil, nil, nil, nil, log)
-	assertKeyspaceMetadata(
-		t,
-		keyspace,
-		&KeyspaceMetadata{
-			Name: "V1Keyspace",
-			Tables: map[string]*TableMetadata{
-				"Schema": {
-					PartitionKey: []*ColumnMetadata{
-						{
-							Name: "key",
-							Type: NativeType{typ: TypeBlob},
-						},
-					},
-					ClusteringColumns: []*ColumnMetadata{},
-					Columns: map[string]*ColumnMetadata{
-						"key": {
-							Name: "key",
-							Type: NativeType{typ: TypeBlob},
-							Kind: ColumnPartitionKey,
-						},
-					},
-				},
-				"hints": {
-					PartitionKey: []*ColumnMetadata{
-						{
-							Name: "target_id",
-							Type: NativeType{typ: TypeUUID},
-						},
-					},
-					ClusteringColumns: []*ColumnMetadata{
-						{
-							Name:  "hint_id",
-							Type:  NativeType{typ: TypeTimeUUID},
-							Order: ASC,
-						},
-						{
-							Name:  "message_version",
-							Type:  NativeType{typ: TypeInt},
-							Order: ASC,
-						},
-					},
-					Columns: map[string]*ColumnMetadata{
-						"target_id": {
-							Name: "target_id",
-							Type: NativeType{typ: TypeUUID},
-							Kind: ColumnPartitionKey,
-						},
-						"hint_id": {
-							Name:  "hint_id",
-							Type:  NativeType{typ: TypeTimeUUID},
-							Order: ASC,
-							Kind:  ColumnClusteringKey,
-						},
-						"message_version": {
-							Name:  "message_version",
-							Type:  NativeType{typ: TypeInt},
-							Order: ASC,
-							Kind:  ColumnClusteringKey,
-						},
-						"mutation": {
-							Name: "mutation",
-							Type: NativeType{typ: TypeBlob},
-							Kind: ColumnRegular,
-						},
-					},
-				},
-				"peers": {
-					PartitionKey: []*ColumnMetadata{
-						{
-							Name: "peer",
-							Type: NativeType{typ: TypeInet},
-						},
-					},
-					ClusteringColumns: []*ColumnMetadata{},
-					Columns: map[string]*ColumnMetadata{
-						"peer": {
-							Name: "peer",
-							Type: NativeType{typ: TypeInet},
-							Kind: ColumnPartitionKey,
-						},
-						"data_center":     {Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "data_center", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UTF8Type", Type: NativeType{typ: TypeVarchar}},
-						"host_id":         {Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "host_id", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UUIDType", Type: NativeType{typ: TypeUUID}},
-						"rack":            {Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "rack", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UTF8Type", Type: NativeType{typ: TypeVarchar}},
-						"release_version": {Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "release_version", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UTF8Type", Type: NativeType{typ: TypeVarchar}},
-						"rpc_address":     {Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "rpc_address", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.InetAddressType", Type: NativeType{typ: TypeInet}},
-						"schema_version":  {Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "schema_version", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UUIDType", Type: NativeType{typ: TypeUUID}},
-						"tokens":          {Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "tokens", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.SetType(org.apache.cassandra.db.marshal.UTF8Type)", Type: CollectionType{NativeType: NativeType{typ: TypeSet}}},
-					},
-				},
-				"IndexInfo": {
-					PartitionKey: []*ColumnMetadata{
-						{
-							Name: "table_name",
-							Type: NativeType{typ: TypeVarchar},
-						},
-					},
-					ClusteringColumns: []*ColumnMetadata{
-						{
-							Name:  "index_name",
-							Type:  NativeType{typ: TypeVarchar},
-							Order: DESC,
-						},
-					},
-					Columns: map[string]*ColumnMetadata{
-						"table_name": {
-							Name: "table_name",
-							Type: NativeType{typ: TypeVarchar},
-							Kind: ColumnPartitionKey,
-						},
-						"index_name": {
-							Name:  "index_name",
-							Type:  NativeType{typ: TypeVarchar},
-							Order: DESC,
-							Kind:  ColumnClusteringKey,
-						},
-						"value": {
-							Name: "value",
-							Type: NativeType{typ: TypeBlob},
-							Kind: ColumnRegular,
-						},
-					},
-				},
-				"wiki_page": {
-					PartitionKey: []*ColumnMetadata{
-						{
-							Name: "title",
-							Type: NativeType{typ: TypeVarchar},
-						},
-					},
-					ClusteringColumns: []*ColumnMetadata{
-						{
-							Name:  "revid",
-							Type:  NativeType{typ: TypeTimeUUID},
-							Order: ASC,
-						},
-					},
-					Columns: map[string]*ColumnMetadata{
-						"title": {
-							Name: "title",
-							Type: NativeType{typ: TypeVarchar},
-							Kind: ColumnPartitionKey,
-						},
-						"revid": {
-							Name: "revid",
-							Type: NativeType{typ: TypeTimeUUID},
-							Kind: ColumnClusteringKey,
-						},
-					},
-				},
-				"no_names": {
-					PartitionKey: []*ColumnMetadata{
-						{
-							Name: "key",
-							Type: NativeType{typ: TypeUUID},
-						},
-						{
-							Name: "key2",
-							Type: NativeType{typ: TypeUUID},
-						},
-					},
-					ClusteringColumns: []*ColumnMetadata{
-						{
-							Name:  "column",
-							Type:  NativeType{typ: TypeInt},
-							Order: ASC,
-						},
-						{
-							Name:  "column2",
-							Type:  NativeType{typ: TypeInt},
-							Order: ASC,
-						},
-						{
-							Name:  "column3",
-							Type:  NativeType{typ: TypeInt},
-							Order: ASC,
-						},
-					},
-					Columns: map[string]*ColumnMetadata{
-						"key": {
-							Name: "key",
-							Type: NativeType{typ: TypeUUID},
-							Kind: ColumnPartitionKey,
-						},
-						"key2": {
-							Name: "key2",
-							Type: NativeType{typ: TypeUUID},
-							Kind: ColumnPartitionKey,
-						},
-						"column": {
-							Name:  "column",
-							Type:  NativeType{typ: TypeInt},
-							Order: ASC,
-							Kind:  ColumnClusteringKey,
-						},
-						"column2": {
-							Name:  "column2",
-							Type:  NativeType{typ: TypeInt},
-							Order: ASC,
-							Kind:  ColumnClusteringKey,
-						},
-						"column3": {
-							Name:  "column3",
-							Type:  NativeType{typ: TypeInt},
-							Order: ASC,
-							Kind:  ColumnClusteringKey,
-						},
-						"value": {
-							Name: "value",
-							Type: NativeType{typ: TypeBlob},
-							Kind: ColumnRegular,
-						},
-					},
-				},
-			},
-		},
-	)
-
 	// V2 test - V2+ protocol is simpler so here are some toy examples to verify that the mapping works
-	keyspace = &KeyspaceMetadata{
+	keyspace := &KeyspaceMetadata{
 		Name: "V2Keyspace",
 	}
-	tables = []TableMetadata{
+	tables := []TableMetadata{
 		{
 			Keyspace: "V2Keyspace",
 			Name:     "Table1",
@@ -351,7 +52,7 @@ func TestCompileMetadata(t *testing.T) {
 			Name:     "Table2",
 		},
 	}
-	columns = []ColumnMetadata{
+	columns := []ColumnMetadata{
 		{
 			Keyspace:       "V2Keyspace",
 			Table:          "Table1",
diff --git a/session_test.go b/session_test.go
index 8633f99..48f7fe7 100644
--- a/session_test.go
+++ b/session_test.go
@@ -35,7 +35,7 @@ import (
 )
 
 func TestSessionAPI(t *testing.T) {
-	cfg := &ClusterConfig{}
+	cfg := NewCluster()
 
 	s := &Session{
 		cfg:    *cfg,
@@ -43,35 +43,13 @@ func TestSessionAPI(t *testing.T) {
 		policy: RoundRobinHostPolicy(),
 		logger: cfg.logger(),
 	}
+	defer s.Close()
 
 	s.pool = cfg.PoolConfig.buildPool(s)
 	s.executor = &queryExecutor{
 		pool:   s.pool,
 		policy: s.policy,
 	}
-	defer s.Close()
-
-	s.SetConsistency(All)
-	if s.cons != All {
-		t.Fatalf("expected consistency 'All', got '%v'", s.cons)
-	}
-
-	s.SetPageSize(100)
-	if s.pageSize != 100 {
-		t.Fatalf("expected pageSize 100, got %v", s.pageSize)
-	}
-
-	s.SetPrefetch(0.75)
-	if s.prefetch != 0.75 {
-		t.Fatalf("expceted prefetch 0.75, got %v", s.prefetch)
-	}
-
-	trace := &traceWriter{}
-
-	s.SetTrace(trace)
-	if s.trace != trace {
-		t.Fatalf("expected traceWriter '%v',got '%v'", trace, s.trace)
-	}
 
 	qry := s.Query("test", 1)
 	if v, ok := qry.values[0].(int); !ok {
@@ -98,7 +76,7 @@ func TestSessionAPI(t *testing.T) {
 
 	testBatch := s.Batch(LoggedBatch)
 	testBatch.Query("test")
-	err := s.ExecuteBatch(testBatch)
+	err := testBatch.Exec()
 
 	if err != ErrNoConnections {
 		t.Fatalf("expected session.ExecuteBatch to return '%v', got '%v'", ErrNoConnections, err)
@@ -111,7 +89,7 @@ func TestSessionAPI(t *testing.T) {
 	//Should just return cleanly
 	s.Close()
 
-	err = s.ExecuteBatch(testBatch)
+	err = testBatch.Exec()
 	if err != ErrSessionClosed {
 		t.Fatalf("expected session.ExecuteBatch to return '%v', got '%v'", ErrSessionClosed, err)
 	}
diff --git a/tuple_test.go b/tuple_test.go
index 296c56f..de5317b 100644
--- a/tuple_test.go
+++ b/tuple_test.go
@@ -35,9 +35,6 @@ import (
 func TestTupleSimple(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("tuple types are only available of proto>=3")
-	}
 
 	err := createTable(session, `CREATE TABLE gocql_test.tuple_test(
 		id int,
@@ -79,9 +76,6 @@ func TestTupleSimple(t *testing.T) {
 func TestTuple_NullTuple(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("tuple types are only available of proto>=3")
-	}
 
 	err := createTable(session, `CREATE TABLE gocql_test.tuple_nil_test(
 		id int,
@@ -117,9 +111,6 @@ func TestTuple_NullTuple(t *testing.T) {
 func TestTuple_TupleNotSet(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("tuple types are only available of proto>=3")
-	}
 
 	err := createTable(session, `CREATE TABLE gocql_test.tuple_not_set_test(
 		id int,
@@ -170,9 +161,6 @@ func TestTuple_TupleNotSet(t *testing.T) {
 func TestTupleMapScan(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("tuple types are only available of proto>=3")
-	}
 
 	err := createTable(session, `CREATE TABLE gocql_test.tuple_map_scan(
 		id int,
@@ -203,9 +191,7 @@ func TestTupleMapScan(t *testing.T) {
 func TestTupleMapScanNil(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("tuple types are only available of proto>=3")
-	}
+
 	err := createTable(session, `CREATE TABLE gocql_test.tuple_map_scan_nil(
 			id int,
 			val frozen<tuple<int, int>>,
@@ -234,9 +220,7 @@ func TestTupleMapScanNil(t *testing.T) {
 func TestTupleMapScanNotSet(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("tuple types are only available of proto>=3")
-	}
+
 	err := createTable(session, `CREATE TABLE gocql_test.tuple_map_scan_not_set(
 			id int,
 			val frozen<tuple<int, int>>,
@@ -266,9 +250,7 @@ func TestTupleLastFieldEmpty(t *testing.T) {
 	// Regression test - empty value used to be treated as NULL value in the last tuple field
 	session := createSession(t)
 	defer session.Close()
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("tuple types are only available of proto>=3")
-	}
+
 	err := createTable(session, `CREATE TABLE gocql_test.tuple_last_field_empty(
 			id int,
 			val frozen<tuple<text, text>>,
@@ -304,9 +286,6 @@ func TestTupleLastFieldEmpty(t *testing.T) {
 func TestTuple_NestedCollection(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("tuple types are only available of proto>=3")
-	}
 
 	err := createTable(session, `CREATE TABLE gocql_test.nested_tuples(
 		id int,
@@ -356,9 +335,6 @@ func TestTuple_NestedCollection(t *testing.T) {
 func TestTuple_NullableNestedCollection(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("tuple types are only available of proto>=3")
-	}
 
 	err := createTable(session, `CREATE TABLE gocql_test.nested_tuples_with_nulls(
 		id int,
diff --git a/udt_test.go b/udt_test.go
index f1980f2..65ee10b 100644
--- a/udt_test.go
+++ b/udt_test.go
@@ -72,10 +72,6 @@ func TestUDT_Marshaler(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
 
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("UDT are only available on protocol >= 3")
-	}
-
 	err := createTable(session, `CREATE TYPE gocql_test.position(
 		lat int,
 		lon int,
@@ -129,10 +125,6 @@ func TestUDT_Reflect(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
 
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("UDT are only available on protocol >= 3")
-	}
-
 	err := createTable(session, `CREATE TYPE gocql_test.horse(
 		name text,
 		owner text);`)
@@ -176,22 +168,10 @@ func TestUDT_Reflect(t *testing.T) {
 	}
 }
 
-func TestUDT_Proto2error(t *testing.T) {
-	// TODO(zariel): move this to marshal test?
-	_, err := Marshal(NativeType{custom: "org.apache.cassandra.db.marshal.UserType.Type", proto: 2}, 1)
-	if err != ErrorUDTUnavailable {
-		t.Fatalf("expected %v got %v", ErrUnavailable, err)
-	}
-}
-
 func TestUDT_NullObject(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
 
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("UDT are only available on protocol >= 3")
-	}
-
 	err := createTable(session, `CREATE TYPE gocql_test.udt_null_type(
 		name text,
 		owner text);`)
@@ -242,10 +222,6 @@ func TestMapScanUDT(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
 
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("UDT are only available on protocol >= 3")
-	}
-
 	err := createTable(session, `CREATE TYPE gocql_test.log_entry (
 		created_timestamp timestamp,
 		message text
@@ -329,10 +305,6 @@ func TestUDT_MissingField(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
 
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("UDT are only available on protocol >= 3")
-	}
-
 	err := createTable(session, `CREATE TYPE gocql_test.missing_field(
 		name text,
 		owner text);`)
@@ -379,10 +351,6 @@ func TestUDT_EmptyCollections(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
 
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("UDT are only available on protocol >= 3")
-	}
-
 	err := createTable(session, `CREATE TYPE gocql_test.nil_collections(
 		a list<text>,
 		b map<text, text>,
@@ -435,10 +403,6 @@ func TestUDT_UpdateField(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
 
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("UDT are only available on protocol >= 3")
-	}
-
 	err := createTable(session, `CREATE TYPE gocql_test.update_field_udt(
 		name text,
 		owner text);`)
@@ -492,10 +456,6 @@ func TestUDT_ScanNullUDT(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
 
-	if session.cfg.ProtoVersion < protoVersion3 {
-		t.Skip("UDT are only available on protocol >= 3")
-	}
-
 	err := createTable(session, `CREATE TYPE gocql_test.scan_null_udt_position(
 		lat int,
 		lon int,