feat(bigtable): add validation for checksum in stream processing

google/cloud/bigtable/BUILD.bazel

@@ -59,6 +59,8 @@ cc_library(
         "@com_google_googleapis//google/bigtable/v2:bigtable_cc_grpc",
         "@com_google_googleapis//google/longrunning:longrunning_cc_grpc",
         "@com_google_googleapis//google/rpc:error_details_cc_proto",
+        "@com_github_google_crc32c//:crc32c",
+        "@com_google_absl//absl/strings:cord",
         "@com_github_grpc_grpc//:grpc++",
     ] + select({
         ":metrics_enabled": [

google/cloud/bigtable/CMakeLists.txt

@@ -48,6 +48,7 @@ include(CTest)
 include(CreateBazelConfig)
 
 find_package(opentelemetry-cpp CONFIG)
+find_package(Crc32c)
 
 # the client library
 add_library(
@@ -179,8 +180,12 @@ add_library(
     internal/common_client.h
     internal/connection_refresh_state.cc
     internal/connection_refresh_state.h
+    internal/const_buffer.cc
+    internal/const_buffer.h
     internal/convert_policies.cc
     internal/convert_policies.h
+    internal/crc32c.cc
+    internal/crc32c.h
     internal/data_connection_impl.cc
     internal/data_connection_impl.h
     internal/data_tracing_connection.cc
@@ -291,6 +296,7 @@ target_link_libraries(
            google-cloud-cpp::bigtable_protos
            google-cloud-cpp::common
            google-cloud-cpp::grpc_utils
+           Crc32c::crc32c
            gRPC::grpc++
            gRPC::grpc
            protobuf::libprotobuf)
@@ -347,7 +353,9 @@ google_cloud_cpp_add_pkgconfig(
     "google_cloud_cpp_grpc_utils"
     "google_cloud_cpp_common"
     "google_cloud_cpp_bigtable_protos"
-    ${EXTRA_MODULES})
+    ${EXTRA_MODULES}
+    LIBS
+    crc32c)
 
 # Create and install the CMake configuration files.
 include(CMakePackageConfigHelpers)
@@ -477,6 +485,7 @@ if (BUILD_TESTING)
         internal/bulk_mutator_test.cc
         internal/connection_refresh_state_test.cc
         internal/convert_policies_test.cc
+        internal/crc32c_test.cc
         internal/data_connection_impl_test.cc
         internal/data_tracing_connection_test.cc
         internal/default_row_reader_test.cc

google/cloud/bigtable/bigtable_client_unit_tests.bzl

@@ -51,6 +51,7 @@ bigtable_client_unit_tests = [
     "internal/bulk_mutator_test.cc",
     "internal/connection_refresh_state_test.cc",
     "internal/convert_policies_test.cc",
+    "internal/crc32c_test.cc",
     "internal/data_connection_impl_test.cc",
     "internal/data_tracing_connection_test.cc",
     "internal/default_row_reader_test.cc",

google/cloud/bigtable/config.cmake.in

@@ -19,5 +19,6 @@ find_dependency(google_cloud_cpp_common)
 find_dependency(google_cloud_cpp_grpc_utils)
 find_dependency(google_cloud_cpp_opentelemetry)
 find_dependency(absl)
+find_dependency(Crc32c)
 
 include("${CMAKE_CURRENT_LIST_DIR}/google_cloud_cpp_bigtable-targets.cmake")

google/cloud/bigtable/google_cloud_cpp_bigtable.bzl

@@ -88,7 +88,9 @@ google_cloud_cpp_bigtable_hdrs = [
     "internal/client_options_defaults.h",
     "internal/common_client.h",
     "internal/connection_refresh_state.h",
+    "internal/const_buffer.h",
     "internal/convert_policies.h",
+    "internal/crc32c.h",
     "internal/data_connection_impl.h",
     "internal/data_tracing_connection.h",
     "internal/default_row_reader.h",
@@ -208,7 +210,9 @@ google_cloud_cpp_bigtable_srcs = [
     "internal/bigtable_tracing_stub.cc",
     "internal/bulk_mutator.cc",
     "internal/connection_refresh_state.cc",
+    "internal/const_buffer.cc",
     "internal/convert_policies.cc",
+    "internal/crc32c.cc",
     "internal/data_connection_impl.cc",
     "internal/data_tracing_connection.cc",
     "internal/default_row_reader.cc",

google/cloud/bigtable/internal/const_buffer.cc

@@ -0,0 +1,44 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "google/cloud/bigtable/internal/const_buffer.h"
+
+namespace google {
+namespace cloud {
+namespace bigtable_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+void PopFrontBytes(ConstBufferSequence& s, std::size_t count) {
+  auto i = s.begin();
+  for (; i != s.end() && i->size() <= count; ++i) {
+    count -= i->size();
+  }
+  if (i == s.end()) {
+    s.clear();
+    return;
+  }
+  // In practice this is expected to be cheap, most vectors will contain 1
+  // or 2 elements. And, if you are really lucky, your compiler turns this
+  // into a memmove():
+  //     https://godbolt.org/z/jw5VDd
+  s.erase(s.begin(), i);
+  if (count > 0 && !s.empty()) {
+    s.front() = ConstBuffer(s.front().data() + count, s.front().size() - count);
+  }
+}
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace bigtable_internal
+}  // namespace cloud
+}  // namespace google

google/cloud/bigtable/internal/const_buffer.h

@@ -0,0 +1,49 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BIGTABLE_INTERNAL_CONST_BUFFER_H
+#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BIGTABLE_INTERNAL_CONST_BUFFER_H
+
+#include "google/cloud/bigtable/version.h"
+#include "absl/types/span.h"
+#include <numeric>
+#include <vector>
+
+namespace google {
+namespace cloud {
+namespace bigtable_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+/// Represent a memory range. Use to upload with low copying
+using ConstBuffer = absl::Span<char const>;
+
+/// Represent a sequence of memory ranges. Use to upload with low copying.
+using ConstBufferSequence = std::vector<ConstBuffer>;
+
+/// The total number of bytes in the buffer sequence.
+inline std::size_t TotalBytes(ConstBufferSequence const& s) {
+  return std::accumulate(
+      s.begin(), s.end(), std::size_t{0},
+      [](std::size_t a, ConstBuffer const& b) { return a + b.size(); });
+}
+
+/// Remove @p count bytes at the start of @p s
+void PopFrontBytes(ConstBufferSequence& s, std::size_t count);
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace bigtable_internal
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BIGTABLE_INTERNAL_CONST_BUFFER_H

google/cloud/bigtable/internal/crc32c.cc

@@ -0,0 +1,95 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "google/cloud/bigtable/internal/crc32c.h"
+#include "absl/base/config.h"
+#if defined(ABSL_LTS_RELEASE_VERSION) && ABSL_LTS_RELEASE_VERSION >= 20230125
+#include "absl/crc/crc32c.h"
+#define GOOGLE_CLOUD_CPP_USE_ABSL_CRC32C 1
+#else
+#define GOOGLE_CLOUD_CPP_USE_ABSL_CRC32C 0
+#endif  // ABSL_LTS_RELEASE_VERSION
+#include <crc32c/crc32c.h>
+
+namespace google {
+namespace cloud {
+namespace bigtable_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+std::uint32_t ExtendCrc32c(std::uint32_t crc, absl::string_view data) {
+  return crc32c::Extend(crc, reinterpret_cast<uint8_t const*>(data.data()),
+                        data.size());
+}
+
+std::uint32_t ExtendCrc32c(std::uint32_t crc,
+                           bigtable_internal::ConstBufferSequence const& data) {
+  for (auto const& b : data) {
+    crc = ExtendCrc32c(crc, absl::string_view{b.data(), b.size()});
+  }
+  return crc;
+}
+
+std::uint32_t ExtendCrc32c(std::uint32_t crc, absl::Cord const& data) {
+  for (auto i = data.chunk_begin(); i != data.chunk_end(); ++i) {
+    crc = ExtendCrc32c(crc, *i);
+  }
+  return crc;
+}
+
+#if GOOGLE_CLOUD_CPP_USE_ABSL_CRC32C
+
+std::uint32_t ExtendCrc32c(std::uint32_t crc, absl::string_view data,
+                           std::uint32_t data_crc) {
+  return static_cast<std::uint32_t>(absl::ConcatCrc32c(
+      absl::crc32c_t{crc}, absl::crc32c_t{data_crc}, data.size()));
+}
+
+std::uint32_t ExtendCrc32c(std::uint32_t crc,
+                           bigtable_internal::ConstBufferSequence const& data,
+                           std::uint32_t data_crc) {
+  auto const size = bigtable_internal::TotalBytes(data);
+  return static_cast<std::uint32_t>(
+      absl::ConcatCrc32c(absl::crc32c_t{crc}, absl::crc32c_t{data_crc}, size));
+}
+
+std::uint32_t ExtendCrc32c(std::uint32_t crc, absl::Cord const& data,
+                           std::uint32_t data_crc) {
+  return static_cast<std::uint32_t>(absl::ConcatCrc32c(
+      absl::crc32c_t{crc}, absl::crc32c_t{data_crc}, data.size()));
+}
+
+#else
+
+std::uint32_t ExtendCrc32c(std::uint32_t crc, absl::string_view data,
+                           std::uint32_t /*data_crc*/) {
+  return ExtendCrc32c(crc, data);
+}
+
+std::uint32_t ExtendCrc32c(std::uint32_t crc,
+                           bigtable_internal::ConstBufferSequence const& data,
+                           std::uint32_t /*data_crc*/) {
+  return ExtendCrc32c(crc, data);
+}
+
+std::uint32_t ExtendCrc32c(std::uint32_t crc, absl::Cord const& data,
+                           std::uint32_t /*data_crc*/) {
+  return ExtendCrc32c(crc, data);
+}
+
+#endif  // GOOGLE_CLOUD_CPP_USE_ABSL_CRC32C
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace bigtable_internal
+}  // namespace cloud
+}  // namespace google

google/cloud/bigtable/internal/crc32c.h

@@ -0,0 +1,60 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BIGTABLE_INTERNAL_CRC32C_H
+#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BIGTABLE_INTERNAL_CRC32C_H
+
+#include "google/cloud/bigtable/internal/const_buffer.h"
+#include "google/cloud/bigtable/version.h"
+#include "absl/strings/cord.h"
+#include "absl/strings/string_view.h"
+#include <cstdint>
+
+namespace google {
+namespace cloud {
+namespace bigtable_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+std::uint32_t ExtendCrc32c(std::uint32_t crc, absl::string_view data);
+std::uint32_t ExtendCrc32c(std::uint32_t crc,
+                           bigtable_internal::ConstBufferSequence const& data);
+std::uint32_t ExtendCrc32c(std::uint32_t crc, absl::Cord const& data);
+
+std::uint32_t ExtendCrc32c(std::uint32_t crc, absl::string_view data,
+                           std::uint32_t data_crc);
+std::uint32_t ExtendCrc32c(std::uint32_t crc,
+                           bigtable_internal::ConstBufferSequence const& data,
+                           std::uint32_t data_crc);
+std::uint32_t ExtendCrc32c(std::uint32_t crc, absl::Cord const& data,
+                           std::uint32_t data_crc);
+
+inline std::uint32_t Crc32c(absl::string_view data) {
+  return ExtendCrc32c(0, data);
+}
+
+inline std::uint32_t Crc32c(
+    bigtable_internal::ConstBufferSequence const& data) {
+  return ExtendCrc32c(0, data);
+}
+
+inline std::uint32_t Crc32c(absl::Cord const& data) {
+  return ExtendCrc32c(0, data);
+}
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace bigtable_internal
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BIGTABLE_INTERNAL_CRC32C_H