Antalya 26.6: Add an Antalya-only protocol version advertised in the ServerHello

docs/en/antalya/protocol.md

@@ -0,0 +1,50 @@
+---
+description: 'How the Antalya fork versions its own wire-protocol changes independently of upstream ClickHouse'
+sidebar_label: 'Antalya Protocol Version'
+sidebar_position: 40
+slug: /antalya/protocol
+title: 'Antalya Protocol Version'
+doc_type: 'reference'
+---
+
+# Antalya protocol version {#antalya-protocol-version}
+
+Antalya versions its own wire-protocol changes with `DBMS_ANTALYA_PROTOCOL_VERSION`, a counter that
+upstream ClickHouse cannot reach, defined in `src/Core/AntalyaProtocol.h`. A server advertises it in
+the `ServerHello` name string, on every connection:
+
+```text
+server -> client   "ClickHouse (antalya:1)"
+```
+
+The client strips the suffix, caps the value with `min(own, server)` and keeps the result. `0` means
+the peer is not an Antalya build. Negotiation is per hop and not transitive: initiator to worker and
+worker to worker negotiate independently.
+
+Version 1 is the advertisement itself. Nothing is gated on it yet.
+
+## Adding an Antalya-only wire change {#adding-a-wire-change}
+
+- Bump `DBMS_ANTALYA_PROTOCOL_VERSION` by one and gate the change on the negotiated value.
+- Never bump `DBMS_TCP_PROTOCOL_VERSION`, and never take a slot in
+  `DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION` for a feature upstream does not have.
+- Keep the counter cumulative. A backport takes the whole contiguous range up to the value it needs,
+  or does not bump at all - the `min(own, server)` cap is only sound for a cumulative feature set.
+- Gate only what the *client* decides to do. The server never learns the client's version, because
+  only the server advertises.
+- Update this page, and update `docs/en/interfaces/specs/NativeProtocol.md` when the change alters
+  a packet layout described there.
+
+## Why a counter of our own {#why-a-counter-of-our-own}
+
+An upstream rebase can reuse the next value of an upstream protocol counter for a different feature.
+Keeping the Antalya counter separate prevents the same version from describing two wire layouts.
+
+## Why the marker rides in `ServerHello` {#why-the-marker-rides-in-serverhello}
+
+The client `Hello` cannot advertise the version because it is sent before the peer is known. Its
+`client_name` is also stored and validated against the Query packet, so changing it can raise
+`CLIENT_INFO_DOES_NOT_MATCH` on an upstream peer.
+
+The server advertises through `server_name`, which is display text. The marker stays inside that
+existing string because adding a field would make older peers read it as the next packet.

docs/en/interfaces/specs/NativeProtocol.md

@@ -440,7 +440,7 @@ Server → Client. The reply to ClientHello on successful authentication.
 
 | # | Field            | Type    | Role      | Condition              | Description |
 |---|------------------|---------|-----------|------------------------|-------------|
-| 1 | server_name      | String  | universal | always                 | Server identifier |
+| 1 | server_name      | String  | universal | always                 | Server identifier. An Altinity Antalya build appends `" (antalya:N)"`, where `N` is its Antalya protocol version; a client may ignore or strip the suffix. See [Antalya protocol version](/antalya/protocol). |
 | 2 | version_major    | VarUInt | universal | always                 | Server major version |
 | 3 | version_minor    | VarUInt | universal | always                 | Server minor version |
 | 4 | protocol_version | VarUInt | universal | always                 | Server's protocol version |

src/Client/Connection.cpp

@@ -26,7 +26,9 @@
 #include <Common/OpenSSLHelpers.h>
 #include <Common/formatReadable.h>
 #include <Common/randomSeed.h>
+#include <Core/AntalyaProtocol.h>
 #include <Core/Block.h>
+#include <Core/Protocol.h>
 #include <Core/ProtocolDefines.h>
 #include <Interpreters/ClientInfo.h>
 #include <Interpreters/OpenTelemetrySpanLog.h>
@@ -369,8 +371,11 @@ void Connection::connect(const ConnectionTimeouts & timeouts)
         if (proto_recv_chunked == "chunked")
             in->enableChunked();
 
-        LOG_TRACE(log_wrapper.get(), "Connected to {} server version {}.{}.{}.",
-            server_name, server_version_major, server_version_minor, server_version_patch);
+        LOG_TRACE(log_wrapper.get(), "Connected to {} server version {}.{}.{}{}.",
+            server_name, server_version_major, server_version_minor, server_version_patch,
+            (server_antalya_protocol_version > 0
+                ? ", Antalya protocol: " + std::to_string(server_antalya_protocol_version)
+                : ""));
 
         /// Now that the handshake is complete, use the regular timeouts
         socket->setReceiveTimeout(timeouts.receive_timeout);
@@ -622,6 +627,7 @@ void Connection::receiveHello()
     {
         readStringBinary(server_name, *in, DBMS_MAX_HELLO_STRING_SIZE);
         sanitizeUntrustedServerString(server_name);
+        server_antalya_protocol_version = AntalyaProtocol::stripMarker(server_name);
         readVarUInt(server_version_major, *in);
         readVarUInt(server_version_minor, *in);
         readVarUInt(server_revision, *in);

src/Client/Connection.h

@@ -238,6 +238,7 @@ class Connection : public IServerConnection
     UInt64 server_parallel_replicas_protocol_version = 0;
     UInt64 worker_cluster_function_protocol_version = 0;
     UInt64 server_query_plan_serialization_version = 0;
+    UInt64 server_antalya_protocol_version = 0;
     String server_timezone;
     String server_display_name;
     SettingsChanges settings_from_server;

src/Core/AntalyaProtocol.cpp

@@ -0,0 +1,61 @@
+#include <Core/AntalyaProtocol.h>
+
+#include <charconv>
+
+#include <algorithm>
+
+
+namespace DB
+{
+
+namespace AntalyaProtocol
+{
+
+constexpr std::string_view MARKER_PREFIX = " (antalya:";
+constexpr size_t MAX_MARKER_DIGITS = 9;
+constexpr size_t MAX_MARKER_SIZE = MARKER_PREFIX.size() + MAX_MARKER_DIGITS + 1;
+constexpr UInt64 MAX_MARKER_VERSION = 999999999;
+
+static_assert(
+    DBMS_ANTALYA_PROTOCOL_VERSION >= 1 && DBMS_ANTALYA_PROTOCOL_VERSION <= MAX_MARKER_VERSION,
+    "DBMS_ANTALYA_PROTOCOL_VERSION does not fit the marker grammar");
+
+String appendMarker(std::string_view name)
+{
+    String result;
+    result.reserve(name.size() + MAX_MARKER_SIZE);
+    result.append(name);
+    result.append(MARKER_PREFIX);
+    result.append(std::to_string(DBMS_ANTALYA_PROTOCOL_VERSION));
+    result.push_back(')');
+    return result;
+}
+
+UInt64 stripMarker(String & name)
+{
+    if (name.empty() || name.back() != ')')
+        return 0;
+
+    const size_t marker_pos = name.rfind(MARKER_PREFIX);
+    if (marker_pos == String::npos)
+        return 0;
+
+    const size_t first_digit = marker_pos + MARKER_PREFIX.size();
+    const size_t digits = name.size() - first_digit - 1;
+    if (digits == 0 || digits > MAX_MARKER_DIGITS || name[first_digit] == '0')
+        return 0;
+
+    UInt64 version;
+    const char * begin = name.data() + first_digit;
+    const char * end = name.data() + name.size() - 1;
+    const auto result = std::from_chars(begin, end, version);
+    if (result.ec != std::errc{} || result.ptr != end)
+        return 0;
+
+    name.resize(marker_pos);
+    return std::min<UInt64>(version, DBMS_ANTALYA_PROTOCOL_VERSION);
+}
+
+}
+
+}

src/Core/AntalyaProtocol.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include <base/types.h>
+
+#include <string_view>
+
+namespace DB
+{
+
+/// Bump for every Antalya-only wire protocol change. See `docs/en/antalya/protocol.md`.
+static constexpr auto DBMS_ANTALYA_PROTOCOL_VERSION = 1;
+
+namespace AntalyaProtocol
+{
+
+String appendMarker(std::string_view name);
+
+/// Removes a valid trailing marker and returns the negotiated version, or `0` if there is no marker.
+UInt64 stripMarker(String & name);
+
+}
+
+}

src/Core/tests/gtest_antalya_protocol.cpp

@@ -0,0 +1,36 @@
+#include <gtest/gtest.h>
+
+#include <Core/AntalyaProtocol.h>
+
+using namespace DB;
+using namespace DB::AntalyaProtocol;
+
+TEST(AntalyaProtocol, AppendMarkerSpellsTheWireForm)
+{
+    EXPECT_EQ(
+        appendMarker("ClickHouse server"),
+        "ClickHouse server (antalya:" + std::to_string(DBMS_ANTALYA_PROTOCOL_VERSION) + ")");
+}
+
+TEST(AntalyaProtocol, RejectsInvalidMarkers)
+{
+    const String rejected[] = {
+        "",
+        "ClickHouse server",
+        "ClickHouse server (antalya:1",
+        "ClickHouse server (antalya:0)",
+        "ClickHouse server (antalya:01)",
+        "ClickHouse server (antalya:1234567890)",
+        "ClickHouse server (antalya:1x)",
+    };
+
+    for (auto name : rejected)
+        EXPECT_EQ(stripMarker(name), 0u) << "should not have parsed: " << name;
+}
+
+TEST(AntalyaProtocol, StripsMarkerAndCapsVersion)
+{
+    String marked = "ClickHouse server (antalya:999999999)";
+    EXPECT_EQ(stripMarker(marked), static_cast<UInt64>(DBMS_ANTALYA_PROTOCOL_VERSION));
+    EXPECT_EQ(marked, "ClickHouse server");
+}

src/Server/TCPHandler.cpp

@@ -10,6 +10,7 @@
 #include <Compression/CompressedReadBuffer.h>
 #include <Compression/CompressedWriteBuffer.h>
 #include <Compression/CompressionFactory.h>
+#include <Core/AntalyaProtocol.h>
 #include <Core/ProtocolDefines.h>
 #include <Core/ServerSettings.h>
 #include <Core/Settings.h>
@@ -2136,7 +2137,7 @@ void TCPHandler::processUnexpectedHello()
 void TCPHandler::sendHello()
 {
     writeVarUInt(Protocol::Server::Hello, *out);
-    writeStringBinary(VERSION_NAME, *out);
+    writeStringBinary(AntalyaProtocol::appendMarker(VERSION_NAME), *out);
     writeVarUInt(VERSION_MAJOR, *out);
     writeVarUInt(VERSION_MINOR, *out);
     writeVarUInt(DBMS_TCP_PROTOCOL_VERSION, *out);

tests/integration/test_antalya_protocol/test.py

@@ -0,0 +1,50 @@
+import pytest
+
+from helpers.cluster import CLICKHOUSE_CI_MIN_TESTED_VERSION, ClickHouseCluster
+
+cluster = ClickHouseCluster(__file__)
+
+node1 = cluster.add_instance("node1")
+node2 = cluster.add_instance("node2")
+# An unmarked build predating Antalya protocol negotiation.
+node_old = cluster.add_instance(
+    "node_old",
+    image="altinity/clickhouse-server",
+    tag=CLICKHOUSE_CI_MIN_TESTED_VERSION,
+    with_installed_binary=True,
+)
+
+NEGOTIATED = "Antalya protocol: "
+
+
+@pytest.fixture(scope="module")
+def started_cluster():
+    try:
+        cluster.start()
+        yield cluster
+    finally:
+        cluster.shutdown()
+
+
+def count_in_log(node, substring):
+    return int(node.count_in_log(substring))
+
+
+def test_remote_function_negotiates(started_cluster):
+    initiator_before = count_in_log(node1, NEGOTIATED)
+    worker_before = count_in_log(node2, NEGOTIATED)
+
+    assert node1.query("SELECT count() FROM remote('node2', system.one)") == "1\n"
+
+    assert count_in_log(node1, NEGOTIATED) > initiator_before
+    assert count_in_log(node2, NEGOTIATED) == worker_before
+
+
+def test_new_initiator_against_an_unmarked_worker(started_cluster):
+    before = count_in_log(node1, NEGOTIATED)
+    assert node1.query("SELECT count() FROM remote('node_old', numbers(10))") == "10\n"
+    assert count_in_log(node1, NEGOTIATED) == before
+
+
+def test_unmarked_initiator_against_a_marked_server(started_cluster):
+    assert node_old.query("SELECT count() FROM remote('node1', numbers(10))") == "10\n"