CPP-815 - Correcting schema agreement with SNI/DBaaS clusters

Author: Michael Fero

cpp-driver/gtests/src/integration/tests/test_dbaas.cpp

@@ -302,6 +302,140 @@ CASSANDRA_INTEGRATION_TEST_F(DbaasTests, QueryEachNode) {
   EXPECT_EQ(3u, server_names.size()); // Ensure all three nodes were queried
 }
 
+/**
+ * Create function and aggregate definitions and ensure the schema metadata is reflected when
+ * execute against the DBaaS SNI single endpoint docker image.
+ *
+ * This test will perform a connection and execute create function/aggregate queries to ensure
+ * schema metadata using a DBaaS SNI single endpoint is handled properly.
+ *
+ * @jira_ticket CPP-815
+ * @test_category dbaas
+ * @test_category queries:schema_metadata:udf
+ * @since 2.14.0
+ * @expected_result Function/Aggregate definitions schema metadata are validated.
+ */
+CASSANDRA_INTEGRATION_TEST_F(DbaasTests, SchemaMetadata) {
+  CHECK_FAILURE;
+
+  Cluster cluster = default_cluster(false);
+  cass_cluster_set_cloud_secure_connection_bundle_no_ssl_lib_init(cluster.get(),
+                                                                  creds_v1().c_str());
+  connect(cluster);
+
+  // clang-format off
+  session_.execute("CREATE OR REPLACE FUNCTION avg_state(state tuple<int, bigint>, val int) "
+                   "CALLED ON NULL INPUT RETURNS tuple<int, bigint> "
+                   "LANGUAGE java AS "
+                     "'if (val != null) {"
+                       "state.setInt(0, state.getInt(0) + 1);"
+                       "state.setLong(1, state.getLong(1) + val.intValue());"
+                     "};"
+                     "return state;'"
+                   ";");
+  session_.execute("CREATE OR REPLACE FUNCTION avg_final (state tuple<int, bigint>) "
+                   "CALLED ON NULL INPUT RETURNS double "
+                   "LANGUAGE java AS "
+                     "'double r = 0;"
+                     "if (state.getInt(0) == 0) return null;"
+                     "r = state.getLong(1);"
+                     "r /= state.getInt(0);"
+                     "return Double.valueOf(r);'"
+                   ";");
+  session_.execute("CREATE OR REPLACE AGGREGATE average(int) "
+                   "SFUNC avg_state STYPE tuple<int, bigint> FINALFUNC avg_final "
+                   "INITCOND(0, 0);");
+  // clang-format on
+
+  const CassSchemaMeta* schema_meta = cass_session_get_schema_meta(session_.get());
+  ASSERT_TRUE(schema_meta != NULL);
+  const CassKeyspaceMeta* keyspace_meta =
+      cass_schema_meta_keyspace_by_name(schema_meta, default_keyspace().c_str());
+  ASSERT_TRUE(keyspace_meta != NULL);
+
+  { // Function `avg_state`
+    const char* data = NULL;
+    size_t length = 0;
+    const CassDataType* datatype = NULL;
+
+    const CassFunctionMeta* function_meta =
+        cass_keyspace_meta_function_by_name(keyspace_meta, "avg_state", "tuple<int,bigint>,int");
+    ASSERT_TRUE(function_meta != NULL);
+    cass_function_meta_name(function_meta, &data, &length);
+    EXPECT_EQ("avg_state", std::string(data, length));
+    cass_function_meta_full_name(function_meta, &data, &length);
+    EXPECT_EQ("avg_state(tuple<int,bigint>,int)", std::string(data, length));
+    cass_function_meta_body(function_meta, &data, &length);
+    EXPECT_EQ("if (val != null) {state.setInt(0, state.getInt(0) + 1);state.setLong(1, "
+              "state.getLong(1) + val.intValue());};return state;",
+              std::string(data, length));
+    cass_function_meta_language(function_meta, &data, &length);
+    EXPECT_EQ("java", std::string(data, length));
+    EXPECT_TRUE(cass_function_meta_called_on_null_input(function_meta));
+    ASSERT_EQ(2u, cass_function_meta_argument_count(function_meta));
+    cass_function_meta_argument(function_meta, 0, &data, &length, &datatype);
+    EXPECT_EQ("state", std::string(data, length));
+    EXPECT_EQ(CASS_VALUE_TYPE_TUPLE, cass_data_type_type(datatype));
+    ASSERT_EQ(2u, cass_data_type_sub_type_count(datatype));
+    EXPECT_EQ(CASS_VALUE_TYPE_INT, cass_data_type_type(cass_data_type_sub_data_type(datatype, 0)));
+    EXPECT_EQ(CASS_VALUE_TYPE_BIGINT,
+              cass_data_type_type(cass_data_type_sub_data_type(datatype, 1)));
+    cass_function_meta_argument(function_meta, 1, &data, &length, &datatype);
+    EXPECT_EQ("val", std::string(data, length));
+    EXPECT_EQ(CASS_VALUE_TYPE_INT, cass_data_type_type(datatype));
+    datatype = cass_function_meta_argument_type_by_name(function_meta, "state");
+    EXPECT_EQ(CASS_VALUE_TYPE_TUPLE, cass_data_type_type(datatype));
+    ASSERT_EQ(2u, cass_data_type_sub_type_count(datatype));
+    EXPECT_EQ(CASS_VALUE_TYPE_INT, cass_data_type_type(cass_data_type_sub_data_type(datatype, 0)));
+    EXPECT_EQ(CASS_VALUE_TYPE_BIGINT,
+              cass_data_type_type(cass_data_type_sub_data_type(datatype, 1)));
+    datatype = cass_function_meta_argument_type_by_name(function_meta, "val");
+    EXPECT_EQ(CASS_VALUE_TYPE_INT, cass_data_type_type(datatype));
+    datatype = cass_function_meta_return_type(function_meta);
+    EXPECT_EQ(CASS_VALUE_TYPE_TUPLE, cass_data_type_type(datatype));
+    ASSERT_EQ(2u, cass_data_type_sub_type_count(datatype));
+    EXPECT_EQ(CASS_VALUE_TYPE_INT, cass_data_type_type(cass_data_type_sub_data_type(datatype, 0)));
+    EXPECT_EQ(CASS_VALUE_TYPE_BIGINT,
+              cass_data_type_type(cass_data_type_sub_data_type(datatype, 1)));
+  }
+
+  { // Aggregate `average`
+    const char* data = NULL;
+    size_t length = 0;
+    const CassDataType* datatype = NULL;
+
+    const CassAggregateMeta* aggregate_meta =
+        cass_keyspace_meta_aggregate_by_name(keyspace_meta, "average", "int");
+    ASSERT_TRUE(aggregate_meta != NULL);
+    cass_aggregate_meta_name(aggregate_meta, &data, &length);
+    EXPECT_EQ("average", std::string(data, length));
+    cass_aggregate_meta_full_name(aggregate_meta, &data, &length);
+    EXPECT_EQ("average(int)", std::string(data, length));
+    size_t count = cass_aggregate_meta_argument_count(aggregate_meta);
+    ASSERT_EQ(1u, cass_aggregate_meta_argument_count(aggregate_meta));
+    datatype = cass_aggregate_meta_argument_type(aggregate_meta, 0);
+    EXPECT_EQ(CASS_VALUE_TYPE_INT, cass_data_type_type(datatype));
+    datatype = cass_aggregate_meta_return_type(aggregate_meta);
+    EXPECT_EQ(CASS_VALUE_TYPE_DOUBLE, cass_data_type_type(datatype));
+    datatype = cass_aggregate_meta_state_type(aggregate_meta);
+    EXPECT_EQ(CASS_VALUE_TYPE_TUPLE, cass_data_type_type(datatype));
+    ASSERT_EQ(2u, cass_data_type_sub_type_count(datatype));
+    EXPECT_EQ(CASS_VALUE_TYPE_INT, cass_data_type_type(cass_data_type_sub_data_type(datatype, 0)));
+    EXPECT_EQ(CASS_VALUE_TYPE_BIGINT,
+              cass_data_type_type(cass_data_type_sub_data_type(datatype, 1)));
+    const CassFunctionMeta* function_meta = cass_aggregate_meta_state_func(aggregate_meta);
+    cass_function_meta_name(function_meta, &data, &length);
+    EXPECT_EQ("avg_state", std::string(data, length));
+    function_meta = cass_aggregate_meta_final_func(aggregate_meta);
+    cass_function_meta_name(function_meta, &data, &length);
+    EXPECT_EQ("avg_final", std::string(data, length));
+    const CassValue* initcond = cass_aggregate_meta_init_cond(aggregate_meta);
+    EXPECT_EQ(CASS_VALUE_TYPE_VARCHAR, cass_value_type(initcond));
+    EXPECT_EQ(Text("(0, 0)"), Text(initcond));
+    ASSERT_TRUE(true);
+  }
+}
+
 /**
  * Ensure guardrails are enabled when performing a query against the DBaaS SNI single endpoint
  * docker image.

cpp-driver/src/address.cpp

@@ -176,43 +176,6 @@ String Address::to_string(bool with_port) const {
 
 namespace datastax { namespace internal { namespace core {
 
-bool determine_address_for_peer_host(const Address& connected_address, const Value* peer_value,
-                                     const Value* rpc_value, Address* output) {
-  Address peer_address;
-  if (!peer_value ||
-      !peer_value->decoder().as_inet(peer_value->size(), connected_address.port(), &peer_address)) {
-    LOG_WARN("Invalid address format for peer address");
-    return false;
-  }
-  if (rpc_value && !rpc_value->is_null()) {
-    if (!rpc_value->decoder().as_inet(rpc_value->size(), connected_address.port(), output)) {
-      LOG_WARN("Invalid address format for rpc address");
-      return false;
-    }
-    if (connected_address == *output || connected_address == peer_address) {
-      LOG_DEBUG("system.peers on %s contains a line with rpc_address for itself. "
-                "This is not normal, but is a known problem for some versions of DSE. "
-                "Ignoring this entry.",
-                connected_address.to_string(false).c_str());
-      return false;
-    }
-    if (Address("0.0.0.0", 0).equals(*output, false) || Address("::", 0).equals(*output, false)) {
-      LOG_WARN("Found host with 'bind any' for rpc_address; using listen_address (%s) to contact "
-               "instead. "
-               "If this is incorrect you should configure a specific interface for rpc_address on "
-               "the server.",
-               peer_address.to_string(false).c_str());
-      *output = peer_address;
-    }
-  } else {
-    LOG_WARN("No rpc_address for host %s in system.peers on %s. "
-             "Ignoring this entry.",
-             peer_address.to_string(false).c_str(), connected_address.to_string(false).c_str());
-    return false;
-  }
-  return true;
-}
-
 String determine_listen_address(const Address& address, const Row* row) {
   const Value* v = row->get_by_name("peer");
   if (v != NULL) {

cpp-driver/src/address.hpp

@@ -28,7 +28,6 @@
 namespace datastax { namespace internal { namespace core {
 
 class Row;
-class Value;
 
 class Address : public Allocated {
 public:
@@ -104,9 +103,6 @@ class Address : public Allocated {
   int port_;
 };
 
-bool determine_address_for_peer_host(const Address& connected_address, const Value* peer_value,
-                                     const Value* rpc_value, Address* output);
-
 String determine_listen_address(const Address& address, const Row* row);
 
 }}} // namespace datastax::internal::core

cpp-driver/src/address_factory.cpp

@@ -0,0 +1,80 @@
+/*
+  Copyright (c) DataStax, Inc.
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+#include "address_factory.hpp"
+
+#include "row.hpp"
+
+using namespace datastax::internal::core;
+
+bool DefaultAddressFactory::create(const Row* peers_row, const Host::Ptr& connected_host,
+                                   Address* output) {
+  Address connected_address = connected_host->address();
+  const Value* peer_value = peers_row->get_by_name("peer");
+  const Value* rpc_value = peers_row->get_by_name("rpc_address");
+
+  Address peer_address;
+  if (!peer_value ||
+      !peer_value->decoder().as_inet(peer_value->size(), connected_address.port(), &peer_address)) {
+    LOG_WARN("Invalid address format for peer address");
+    return false;
+  }
+  if (rpc_value && !rpc_value->is_null()) {
+    if (!rpc_value->decoder().as_inet(rpc_value->size(), connected_address.port(), output)) {
+      LOG_WARN("Invalid address format for rpc address");
+      return false;
+    }
+    if (connected_address == *output || connected_address == peer_address) {
+      LOG_DEBUG("system.peers on %s contains a line with rpc_address for itself. "
+                "This is not normal, but is a known problem for some versions of DSE. "
+                "Ignoring this entry.",
+                connected_address.to_string(false).c_str());
+      return false;
+    }
+    if (Address("0.0.0.0", 0).equals(*output, false) || Address("::", 0).equals(*output, false)) {
+      LOG_WARN("Found host with 'bind any' for rpc_address; using listen_address (%s) to contact "
+               "instead. If this is incorrect you should configure a specific interface for "
+               "rpc_address on the server.",
+               peer_address.to_string(false).c_str());
+      *output = peer_address;
+    }
+  } else {
+    LOG_WARN("No rpc_address for host %s in system.peers on %s. Ignoring this entry.",
+             peer_address.to_string(false).c_str(), connected_address.to_string(false).c_str());
+    return false;
+  }
+  return true;
+}
+
+bool SniAddressFactory::create(const Row* peers_row, const Host::Ptr& connected_host,
+                               Address* output) {
+  CassUuid host_id;
+  if (!peers_row->get_uuid_by_name("host_id", &host_id)) {
+    // Attempt to get an peer address for the error log.
+    Address peer_address;
+    const Value* peer_value = peers_row->get_by_name("peer");
+    if (!peer_value || !peer_value->decoder().as_inet(
+                           peer_value->size(), connected_host->address().port(), &peer_address)) {
+      LOG_WARN("Invalid address format for peer address");
+    }
+    LOG_ERROR("Invalid `host_id` for host. %s will be ignored.",
+              peer_address.is_valid() ? peer_address.to_string().c_str() : "<unknown>");
+    return false;
+  }
+  *output = Address(connected_host->address().hostname_or_address(),
+                    connected_host->address().port(), to_string(host_id));
+  return true;
+}

cpp-driver/src/address_factory.hpp

@@ -0,0 +1,63 @@
+/*
+  Copyright (c) DataStax, Inc.
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+#ifndef DATASTAX_INTERNAL_ADDRESS_FACTORY_HPP
+#define DATASTAX_INTERNAL_ADDRESS_FACTORY_HPP
+
+#include "config.hpp"
+#include "host.hpp"
+#include "ref_counted.hpp"
+
+namespace datastax { namespace internal { namespace core {
+
+class Row;
+
+/**
+ * An interface for constructing `Address` from `system.local`/`system.peers` row data.
+ */
+class AddressFactory : public RefCounted<AddressFactory> {
+public:
+  typedef SharedRefPtr<AddressFactory> Ptr;
+  virtual ~AddressFactory() {}
+  virtual bool create(const Row* peers_row, const Host::Ptr& connected_host, Address* output) = 0;
+};
+
+/**
+ * An address factory that creates `Address` using the `rpc_address` column.
+ */
+class DefaultAddressFactory : public AddressFactory {
+  virtual bool create(const Row* peers_row, const Host::Ptr& connected_host, Address* output);
+};
+
+/**
+ * An address factory that creates `Address` using the connected host's address and the `host_id`
+ * (for the SNI servername) column.
+ */
+class SniAddressFactory : public AddressFactory {
+  virtual bool create(const Row* peers_row, const Host::Ptr& connected_host, Address* output);
+};
+
+inline AddressFactory* create_address_factory_from_config(const Config& config) {
+  if (config.cloud_secure_connection_config().is_loaded()) {
+    return new SniAddressFactory();
+  } else {
+    return new DefaultAddressFactory();
+  }
+}
+
+}}} // namespace datastax::internal::core
+
+#endif

cpp-driver/src/control_connection.cpp

@@ -304,40 +304,6 @@ class NopControlConnectionListener : public ControlConnectionListener {
 
 static NopControlConnectionListener nop_listener__;
 
-bool DefaultAddressFactory::create(const Row* peers_row, const Host::Ptr& connected_host,
-                                   Address* address) {
-  return determine_address_for_peer_host(connected_host->address(), peers_row->get_by_name("peer"),
-                                         peers_row->get_by_name("rpc_address"), address);
-}
-
-bool SniAddressFactory::create(const Row* peers_row, const Host::Ptr& connected_host,
-                               Address* address) {
-  CassUuid host_id;
-  if (!peers_row->get_uuid_by_name("host_id", &host_id)) {
-    // Attempt to get an peer address for the error log.
-    Address peer_address;
-    const Value* peer_value = peers_row->get_by_name("peer");
-    if (!peer_value || !peer_value->decoder().as_inet(
-                           peer_value->size(), connected_host->address().port(), &peer_address)) {
-      LOG_WARN("Invalid address format for peer address");
-    }
-    LOG_ERROR("Invalid `host_id` for host. %s will be ignored.",
-              peer_address.is_valid() ? peer_address.to_string().c_str() : "<unknown>");
-    return false;
-  }
-  *address = Address(connected_host->address().hostname_or_address(),
-                     connected_host->address().port(), to_string(host_id));
-  return true;
-}
-
-static AddressFactory* create_address_factory_from_config(const Config& config) {
-  if (config.cloud_secure_connection_config().is_loaded()) {
-    return new SniAddressFactory();
-  } else {
-    return new DefaultAddressFactory();
-  }
-}
-
 ControlConnectionSettings::ControlConnectionSettings()
     : use_schema(CASS_DEFAULT_USE_SCHEMA)
     , use_token_aware_routing(CASS_DEFAULT_USE_TOKEN_AWARE_ROUTING)
@@ -433,9 +399,7 @@ void ControlConnection::handle_refresh_node(RefreshNodeCallback* callback) {
     row = rows.row();
     if (callback->is_all_peers) {
       Address address;
-      bool is_valid_address = determine_address_for_peer_host(
-          connection_->host()->rpc_address(), row->get_by_name("peer"),
-          row->get_by_name("rpc_address"), &address);
+      bool is_valid_address = settings_.address_factory->create(row, connection_->host(), &address);
       if (is_valid_address && callback->address == address) {
         found_host = true;
       }