diff --git a/google/cloud/bigtable/ci/run_conformance_tests_proxy_bazel.sh b/google/cloud/bigtable/ci/run_conformance_tests_proxy_bazel.sh
index 18da4862fb76e..c2c4eb1e6402a 100755
--- a/google/cloud/bigtable/ci/run_conformance_tests_proxy_bazel.sh
+++ b/google/cloud/bigtable/ci/run_conformance_tests_proxy_bazel.sh
@@ -72,10 +72,10 @@ exit_status=$?
 #exit_status=$?
 
 # Response/Metadata mismatches b/461233335
-#go test -v \
-#  -run "FailsOnTypeMismatch|FailsOnTypeMismatchWithinMap|FailsOnTypeMismatchWithinArray|FailsOnTypeMismatchWithinStruct|FailsOnStructMissingField" \
-#  -proxy_addr=:9999
-#exit_status=$?
+go test -v \
+  -run "FailsOnTypeMismatch|FailsOnTypeMismatchWithinMap|FailsOnTypeMismatchWithinArray|FailsOnTypeMismatchWithinStruct|FailsOnStructMissingField" \
+  -proxy_addr=:9999
+exit_status=$?
 
 # QueryPlan refresh tests b/461233613
 #go test -v \
diff --git a/google/cloud/bigtable/internal/data_connection_impl.cc b/google/cloud/bigtable/internal/data_connection_impl.cc
index 7cdc49d753ded..c2678799c3264 100644
--- a/google/cloud/bigtable/internal/data_connection_impl.cc
+++ b/google/cloud/bigtable/internal/data_connection_impl.cc
@@ -111,6 +111,10 @@ bool IsStatusMetadataIndicatingRetryPolicyExhausted(Status const& status) {
                                                 "retry-policy-exhausted"));
 }
 
+bool IsStatusIndicatingInternalError(Status const& status) {
+  return status.code() == StatusCode::kInternal;
+}
+
 class DefaultPartialResultSetReader
     : public bigtable_internal::PartialResultSetReader {
  public:
@@ -962,6 +966,11 @@ bigtable::RowStream DataConnectionImpl::ExecuteQuery(
       }
       last_status = source.status();
 
+      if (IsStatusIndicatingInternalError(source.status())) {
+        return bigtable::RowStream(std::make_unique<StatusOnlyResultSetSource>(
+            std::move(last_status)));
+      }
+
       if (QueryPlanRefreshRetry::IsQueryPlanExpired(source.status())) {
         query_plan->Invalidate(source.status(),
                                query_plan_data->prepared_query());
diff --git a/google/cloud/bigtable/internal/partial_result_set_source.cc b/google/cloud/bigtable/internal/partial_result_set_source.cc
index efe3d9c4999fa..991ef44a9407e 100644
--- a/google/cloud/bigtable/internal/partial_result_set_source.cc
+++ b/google/cloud/bigtable/internal/partial_result_set_source.cc
@@ -52,9 +52,10 @@ PartialResultSetSource::Create(
   // Do an initial read from the stream to determine the fate of the factory.
   auto status = source->ReadFromStream();
 
-  // If the initial read finished the stream, and `Finish()` failed, then
-  // creating the `PartialResultSetSource` should fail with the same error.
-  if (source->state_ == State::kFinished && !status.ok()) return status;
+  // Any error during parsing will be returned.
+  if (!status.ok()) {
+    return status;
+  }
 
   return {std::move(source)};
 }
@@ -228,6 +229,11 @@ Status PartialResultSetSource::BufferProtoRows() {
 
     while (parsed_value != proto_values.end()) {
       for (auto const& column : proto_schema.columns()) {
+        auto type_value_match_result =
+            bigtable::Value::TypeAndValuesMatch(column.type(), *parsed_value);
+        if (!type_value_match_result.ok()) {
+          return type_value_match_result;
+        }
         auto value = FromProto(column.type(), *parsed_value);
         values.push_back(std::move(value));
         ++parsed_value;
diff --git a/google/cloud/bigtable/value.cc b/google/cloud/bigtable/value.cc
index 47e88ddfc8441..9b1ddf95d914e 100644
--- a/google/cloud/bigtable/value.cc
+++ b/google/cloud/bigtable/value.cc
@@ -17,6 +17,7 @@
 #include "google/cloud/internal/throw_delegate.h"
 #include "absl/container/flat_hash_set.h"
 #include "absl/strings/cord.h"
+#include "absl/strings/substitute.h"
 #include <google/bigtable/v2/types.pb.h>
 #include <google/protobuf/descriptor.h>
 #include <google/protobuf/message.h>
@@ -207,12 +208,6 @@ bool MapEqual(  // NOLINT(misc-no-recursion)
                              comparison_function);
 }
 
-// From the proto description, `NULL` values are represented by having a kind
-// equal to KIND_NOT_SET
-bool IsNullValue(google::bigtable::v2::Value const& value) {
-  return value.kind_case() == google::bigtable::v2::Value::KIND_NOT_SET;
-}
-
 // A helper to escape all double quotes in the given string `s`. For example,
 // if given `"foo"`, outputs `\"foo\"`. This is useful when a caller needs to
 // wrap `s` itself in double quotes.
@@ -238,7 +233,7 @@ std::ostream& StreamHelper(std::ostream& os,  // NOLINT(misc-no-recursion)
                            google::bigtable::v2::Value const& v,
                            google::bigtable::v2::Type const& t,
                            StreamMode mode) {
-  if (IsNullValue(v)) {
+  if (Value::IsNullValue(v)) {
     return os << "NULL";
   }
 
@@ -344,6 +339,161 @@ std::ostream& operator<<(std::ostream& os, Value const& v) {
   return StreamHelper(os, v.value_, v.type_, StreamMode::kScalar);
 }
 
+// NOLINTNEXTLINE(misc-no-recursion)
+Status Value::TypeAndArrayValuesMatch(
+    google::bigtable::v2::Type const& type,
+    google::bigtable::v2::Value const& value) {
+  if (!value.has_array_value()) {
+    return internal::InternalError(
+        "Value kind must be ARRAY_VALUE for columns of type: MAP");
+  }
+  auto const& vals = value.array_value().values();
+  for (auto const& val : vals) {
+    auto const element_match_result =
+        TypeAndValuesMatch(type.array_type().element_type(), val);
+    if (!element_match_result.ok()) {
+      return element_match_result;
+    }
+  }
+  return Status{};
+}
+
+// NOLINTNEXTLINE(misc-no-recursion)
+Status Value::TypeAndMapValuesMatch(google::bigtable::v2::Type const& type,
+                                    google::bigtable::v2::Value const& value) {
+  if (!value.has_array_value()) {
+    return internal::InternalError(
+        "Value kind must be ARRAY_VALUE for columns of type: MAP");
+  }
+  auto key_type = type.map_type().key_type();
+  auto value_type = type.map_type().value_type();
+  auto const& vals = value.array_value().values();
+  for (auto const& val : vals) {
+    if (!val.has_array_value() || val.array_value().values_size() != 2) {
+      return internal::InternalError(
+          "ARRAY_VALUE must contain entries of 2 values");
+    }
+    auto map_key = val.array_value().values(0);
+    auto map_value = val.array_value().values(1);
+    // NOLINTNEXTLINE(misc-no-recursion)
+    auto key_match_result = TypeAndValuesMatch(key_type, map_key);
+    if (!key_match_result.ok()) {
+      return key_match_result;
+    }
+    // NOLINTNEXTLINE(misc-no-recursion)
+    auto value_match_result = TypeAndValuesMatch(value_type, map_value);
+    if (!value_match_result.ok()) {
+      return value_match_result;
+    }
+  }
+  return Status{};
+}
+
+// NOLINTNEXTLINE(misc-no-recursion)
+Status Value::TypeAndStructValuesMatch(
+    google::bigtable::v2::Type const& type,
+    google::bigtable::v2::Value const& value) {
+  if (!value.has_array_value()) {
+    return internal::InternalError(
+        "Value kind must be ARRAY_VALUE for columns of type: STRUCT");
+  }
+  auto fields = type.struct_type().fields();
+  auto values = value.array_value().values();
+  if (fields.size() != values.size()) {
+    auto const message = absl::Substitute(
+        "received Struct with $0 values, but metadata has $1 fields",
+        values.size(), fields.size());
+    return internal::InternalError(message);
+  }
+  for (int i = 0; i < fields.size(); ++i) {
+    auto const& f1 = fields.Get(i);
+    auto const& v = values[i];
+    auto match_result = TypeAndValuesMatch(f1.type(), v);
+    if (!match_result.ok()) {
+      return match_result;
+    }
+  }
+  return Status{};
+}
+
+/**
+ * Checks whether the declared type in column matches the value's contents.
+ * Since the received values may or may not have type() set, we check against
+ * the value contents themselves
+ */
+// NOLINTNEXTLINE(misc-no-recursion)
+Status Value::TypeAndValuesMatch(google::bigtable::v2::Type const& type,
+                                 google::bigtable::v2::Value const& value) {
+  using google::bigtable::v2::Type;
+  auto make_mismatch_metadata_status = [&](std::string const& value_kind,
+                                           std::string const& type_name) {
+    auto const message = absl::Substitute(
+        "Value kind must be $0 for columns of type: $1", value_kind, type_name);
+    return internal::InternalError(message);
+  };
+  // Null values are allowed by default
+  if (IsNullValue(value)) {
+    return Status{};
+  }
+  Status result;
+  switch (type.kind_case()) {
+    case Type::kArrayType:
+      result = TypeAndArrayValuesMatch(type, value);
+      break;
+    case Type::kMapType:
+      result = TypeAndMapValuesMatch(type, value);
+      break;
+    case Type::kStructType:
+      result = TypeAndStructValuesMatch(type, value);
+      break;
+    case Type::kBoolType:
+      if (!value.has_bool_value()) {
+        result = make_mismatch_metadata_status("BOOL_VALUE", "BOOL");
+      }
+      break;
+    case Type::kBytesType:
+      if (!value.has_bytes_value()) {
+        result = make_mismatch_metadata_status("BYTES_VALUE", "BYTES");
+      }
+      break;
+    case Type::kDateType:
+      if (!value.has_date_value()) {
+        result = make_mismatch_metadata_status("DATE_VALUE", "DATE");
+      }
+      break;
+    case Type::kFloat32Type:
+      if (!value.has_float_value()) {
+        result = make_mismatch_metadata_status("FLOAT_VALUE", "FLOAT32");
+      }
+      break;
+    case Type::kFloat64Type:
+      if (!value.has_float_value()) {
+        result = make_mismatch_metadata_status("FLOAT_VALUE", "FLOAT64");
+      }
+      break;
+    case Type::kInt64Type:
+      if (!value.has_int_value()) {
+        result = make_mismatch_metadata_status("INT_VALUE", "INT64");
+      }
+      break;
+    case Type::kStringType:
+      if (!value.has_string_value()) {
+        result = make_mismatch_metadata_status("STRING_VALUE", "STRING");
+      }
+      break;
+    case Type::kTimestampType:
+      if (!value.has_timestamp_value()) {
+        result = make_mismatch_metadata_status("TIMESTAMP_VALUE", "TIMESTAMP");
+      }
+      break;
+    default:
+      result = internal::InternalError("Unsupported type");
+      break;
+  }
+  // Nulls are allowed;
+  return result;
+}
+
 //
 // Value::TypeProtoIs
 //
diff --git a/google/cloud/bigtable/value.h b/google/cloud/bigtable/value.h
index 4f986dc2dfae2..d7c16679645e7 100644
--- a/google/cloud/bigtable/value.h
+++ b/google/cloud/bigtable/value.h
@@ -310,7 +310,24 @@ class Value {
    */
   friend std::ostream& operator<<(std::ostream& os, Value const& v);
 
+  // `NULL` values are represented by having a kind equal to KIND_NOT_SET
+  static bool IsNullValue(google::bigtable::v2::Value const& value) {
+    return value.kind_case() == google::bigtable::v2::Value::KIND_NOT_SET;
+  }
+
+  static Status TypeAndValuesMatch(google::bigtable::v2::Type const& type,
+                                   google::bigtable::v2::Value const& value);
+
  private:
+  static Status TypeAndArrayValuesMatch(
+      google::bigtable::v2::Type const& type,
+      google::bigtable::v2::Value const& value);
+  static Status TypeAndMapValuesMatch(google::bigtable::v2::Type const& type,
+                                      google::bigtable::v2::Value const& value);
+  static Status TypeAndStructValuesMatch(
+      google::bigtable::v2::Type const& type,
+      google::bigtable::v2::Value const& value);
+
   // Metafunction that returns true if `T` is an `absl::optional<U>`
   template <typename T>
   struct IsOptional : std::false_type {};
diff --git a/google/cloud/bigtable/value_test.cc b/google/cloud/bigtable/value_test.cc
index 21d459a404ace..6da6a52d510dd 100644
--- a/google/cloud/bigtable/value_test.cc
+++ b/google/cloud/bigtable/value_test.cc
@@ -1748,6 +1748,275 @@ TEST(Value, OutputStreamMatchesT) {
   // Not included, because a raw map cannot be streamed.
 }
 
+void TestTypeAndValuesMatch(std::string const& type_text,
+                            std::string const& value_text, bool expected) {
+  google::bigtable::v2::Type type;
+  ASSERT_TRUE(TextFormat::ParseFromString(type_text, &type));
+  google::bigtable::v2::Value value;
+  ASSERT_TRUE(TextFormat::ParseFromString(value_text, &value));
+  auto result = Value::TypeAndValuesMatch(type, value);
+  if (expected) {
+    EXPECT_STATUS_OK(result);
+  } else {
+    EXPECT_THAT(result, Not(IsOk()));
+  }
+}
+
+TEST(Value, TypeAndValuesMatchScalar) {
+  TestTypeAndValuesMatch("int64_type {}", "int_value: 123", true);
+  TestTypeAndValuesMatch("string_type {}", "string_value: 'hello'", true);
+  TestTypeAndValuesMatch("bool_type {}", "bool_value: true", true);
+  TestTypeAndValuesMatch("float64_type {}", "float_value: 3.14", true);
+  TestTypeAndValuesMatch("float32_type {}", "float_value: 3.14", true);
+  TestTypeAndValuesMatch("bytes_type {}", "bytes_value: 'bytes'", true);
+  TestTypeAndValuesMatch("timestamp_type {}",
+                         "timestamp_value: { seconds: 123 }", true);
+  TestTypeAndValuesMatch("date_type {}",
+                         "date_value: { year: 2025, month: 1, day: 1 }", true);
+}
+
+TEST(Value, TypeAndValuesMatchScalarMismatch) {
+  TestTypeAndValuesMatch("int64_type {}", "string_value: 'mismatch'", false);
+  TestTypeAndValuesMatch("string_type {}", "int_value: 123", false);
+}
+
+TEST(Value, TypeAndValuesMatchNullScalar) {
+  TestTypeAndValuesMatch("int64_type {}", "", true);
+  TestTypeAndValuesMatch("string_type {}", "", true);
+}
+
+TEST(Value, TypeAndValuesMatchArray) {
+  auto const* type = R"pb(
+    array_type { element_type { int64_type {} } }
+  )pb";
+  auto const* matching_value = R"pb(
+    array_value {
+      values { int_value: 1 }
+      values { int_value: 2 }
+    }
+  )pb";
+  TestTypeAndValuesMatch(type, matching_value, true);
+}
+
+TEST(Value, TypeAndValuesMatchArrayMismatchElementType) {
+  auto const* type = R"pb(
+    array_type { element_type { int64_type {} } }
+  )pb";
+  auto const* mismatched_value = R"pb(
+    array_value {
+      values { int_value: 1 }
+      values { string_value: "2" }
+    }
+  )pb";
+  TestTypeAndValuesMatch(type, mismatched_value, false);
+}
+
+TEST(Value, TypeAndValuesMatchArrayMismatchScalar) {
+  auto const* type = R"pb(
+    array_type { element_type { int64_type {} } }
+  )pb";
+  TestTypeAndValuesMatch(type, "int_value: 123", false);
+}
+
+TEST(Value, TypeAndValuesMatchArrayWithNull) {
+  auto const* type = R"pb(
+    array_type { element_type { int64_type {} } }
+  )pb";
+  auto const* value_with_null = R"pb(
+    array_value {
+      values { int_value: 1 }
+      values {}  # null
+      values { int_value: 3 }
+    }
+  )pb";
+  TestTypeAndValuesMatch(type, value_with_null, true);
+}
+
+TEST(Value, TypeAndValuesMatchStruct) {
+  auto const* type = R"pb(
+    struct_type {
+      fields {
+        field_name: "name"
+        type { string_type {} }
+      }
+      fields {
+        field_name: "age"
+        type { int64_type {} }
+      }
+    }
+  )pb";
+  auto const* matching_value = R"pb(
+    array_value {
+      values { string_value: "John" }
+      values { int_value: 42 }
+    }
+  )pb";
+  TestTypeAndValuesMatch(type, matching_value, true);
+}
+
+TEST(Value, TypeAndValuesMatchStructMismatchFieldType) {
+  auto const* type = R"pb(
+    struct_type {
+      fields {
+        field_name: "name"
+        type { string_type {} }
+      }
+      fields {
+        field_name: "age"
+        type { int64_type {} }
+      }
+    }
+  )pb";
+  auto const* mismatched_value = R"pb(
+    array_value {
+      values { string_value: "John" }
+      values { string_value: "42" }
+    }
+  )pb";
+  TestTypeAndValuesMatch(type, mismatched_value, false);
+}
+
+TEST(Value, TypeAndValuesMatchStructMismatchFieldCount) {
+  auto const* type = R"pb(
+    struct_type {
+      fields { type { string_type {} } }
+      fields { type { int64_type {} } }
+    }
+  )pb";
+  auto const* mismatched_value = R"pb(
+    array_value { values { string_value: "John" } }
+  )pb";
+  // The current implementation has a bug and will return true here.
+  // This test will fail until the bug is fixed.
+  TestTypeAndValuesMatch(type, mismatched_value, false);
+}
+
+TEST(Value, TypeAndValuesMatchStructMismatchScalar) {
+  auto const* type = R"pb(
+    struct_type { fields { type { string_type {} } } }
+  )pb";
+  TestTypeAndValuesMatch(type, "string_value: 'John'", false);
+}
+
+TEST(Value, TypeAndValuesMatchStructWithNull) {
+  auto const* type = R"pb(
+    struct_type {
+      fields { type { string_type {} } }
+      fields { type { int64_type {} } }
+    }
+  )pb";
+  auto const* value_with_null = R"pb(
+    array_value {
+      values { string_value: "John" }
+      values {}
+    }
+  )pb";
+  TestTypeAndValuesMatch(type, value_with_null, true);
+}
+
+TEST(Value, TypeAndValuesMatchMap) {
+  auto const* type = R"pb(
+    map_type {
+      key_type { string_type {} }
+      value_type { int64_type {} }
+    }
+  )pb";
+  auto const* matching_value = R"pb(
+    array_value {
+      values {
+        array_value {
+          values { string_value: "key1" }
+          values { int_value: 1 }
+        }
+      }
+    }
+  )pb";
+  TestTypeAndValuesMatch(type, matching_value, true);
+}
+
+TEST(Value, TypeAndValuesMatchMapMismatchKeyType) {
+  auto const* type = R"pb(
+    map_type {
+      key_type { string_type {} }
+      value_type { int64_type {} }
+    }
+  )pb";
+  auto const* mismatched_value = R"pb(
+    array_value {
+      values {
+        array_value {
+          values { int_value: 1 }
+          values { int_value: 1 }
+        }
+      }
+    }
+  )pb";
+  TestTypeAndValuesMatch(type, mismatched_value, false);
+}
+
+TEST(Value, TypeAndValuesMatchMapMismatchValueType) {
+  auto const* type = R"pb(
+    map_type {
+      key_type { string_type {} }
+      value_type { int64_type {} }
+    }
+  )pb";
+  auto const* mismatched_value = R"pb(
+    array_value {
+      values {
+        array_value {
+          values { string_value: "key1" }
+          values { string_value: "1" }
+        }
+      }
+    }
+  )pb";
+  TestTypeAndValuesMatch(type, mismatched_value, false);
+}
+
+TEST(Value, TypeAndValuesMatchMapMismatchScalar) {
+  auto const* type = R"pb(
+    map_type {
+      key_type { string_type {} }
+      value_type { int64_type {} }
+    }
+  )pb";
+  TestTypeAndValuesMatch(type, "string_value: 'foo'", false);
+}
+
+TEST(Value, TypeAndValuesMatchMapMalformedEntry) {
+  auto const* type = R"pb(
+    map_type {
+      key_type { string_type {} }
+      value_type { int64_type {} }
+    }
+  )pb";
+  auto const* malformed_value = R"pb(
+    array_value { values { array_value { values { string_value: "key1" } } } }
+  )pb";
+  TestTypeAndValuesMatch(type, malformed_value, false);
+}
+
+TEST(Value, TypeAndValuesMatchMapWithNullValue) {
+  auto const* type = R"pb(
+    map_type {
+      key_type { string_type {} }
+      value_type { int64_type {} }
+    }
+  )pb";
+  auto const* value_with_null = R"pb(
+    array_value {
+      values {
+        array_value {
+          values { string_value: "key1" }
+          values {}
+        }
+      }
+    }
+  )pb";
+  TestTypeAndValuesMatch(type, value_with_null, true);
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace bigtable