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

docs/en/antalya/protocol.md

@@ -0,0 +1,91 @@
+---
+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.
+- Document it here. Also document it in `docs/en/interfaces/specs/NativeProtocol.md` if it changes
+  the layout of a packet that file describes field by field: a third-party client (`ch-go`,
+  `clickhouse-go`, `clickhouse-driver`) that does not know an added field cannot parse the stream
+  past it. Everything else stays here, so that the spec keeps tracking upstream through rebases.
+
+## Why a counter of our own {#why-a-counter-of-our-own}
+
+An Antalya-only change that takes a slot in an upstream counter has to be renumbered whenever
+upstream claims that slot for something else. The same number then means two different things in two
+shipped builds, and two nodes that negotiate it disagree about the bytes on the wire.
+
+The constant lives in its own header rather than in `src/Core/ProtocolDefines.h`, whose tail is
+where upstream adds its own constants, and therefore where every rebase conflicts.
+
+## Why the marker rides in `ServerHello` {#why-the-marker-rides-in-serverhello}
+
+**The client `Hello` is never marked.** The client writes that packet before it has read a byte from
+the peer, so it cannot gate a marker on what the peer is. Every field in it - `client_name`,
+`default_database`, `user`, `password` - is one an upstream server acts on. `client_name` is the
+worst: a server persists it, and `validate_tcp_client_information` compares it against the Query
+packet's `ClientInfo`, so a marker there fails a `remote()` query with `CLIENT_INFO_DOES_NOT_MATCH`
+on any peer that does not strip it.
+
+`ServerHello` has no such field. `server_name` is client-side display text and reaches no system
+table, which is why the server is the side that speaks.
+
+**The marker is a suffix, not extra bytes.** Neither Hello nor the Addendum has a length prefix or a
+terminator: a reader stops after the last field its negotiated revision knows about, so a peer that
+does not expect extra bytes reads them as the next packet.
+
+## What you see {#what-you-see}
+
+An Antalya client appends the negotiated version to its existing connection log line
+(`Connected to ... server version ...`). The server logs nothing, and degradation to `0` is silent.
+
+Nothing a server stores changes: `system.query_log` and `system.processes` keep the `client_name`
+the peer sent, so filters written as `client_name = 'ClickHouse server'` keep working.
+
+The suffix is visible in one place - an **upstream** client's `server_name`, which that client does
+not strip, so its banner reads `Connected to ClickHouse (antalya:1) server version ...`.
+
+## No opt-out {#no-opt-out}
+
+No setting suppresses the marker: a node either speaks this protocol or is not an Antalya build.
+
+`server_name` is already a build-time value - upstream sets it from the CMake project name, and any
+fork may set it to anything - so a native client cannot treat it as a fixed string. What a switch
+would protect is the displayed name on a peer that does not strip the suffix, which is cosmetic, and
+it would cost a state in which two Antalya nodes that both support a feature fail to negotiate it
+because one was configured not to say so. If a peer is ever found that breaks on the suffix, the fix
+is to stop appending it.
+
+## Implementation {#implementation}
+
+`src/Core/AntalyaProtocol.h` holds the version constant, the marker grammar and the `appendMarker` /
+`parseMarker` / `stripMarker` / `negotiate` helpers. `TCPHandler::sendHello` appends the marker;
+`Connection::receiveHello` strips it and stores the capped value.

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::negotiate(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/Client/tests/gtest_sanitize_untrusted_server_string.cpp

@@ -1,6 +1,7 @@
 #include <gtest/gtest.h>
 
 #include <Client/sanitizeUntrustedServerString.h>
+#include <Core/AntalyaProtocol.h>
 #include <Core/ProtocolDefines.h>
 
 using namespace DB;
@@ -155,3 +156,21 @@ TEST(SanitizeUntrustedServerString, MaxPasswordComplexityRulesCapIsTight)
     EXPECT_LE(DBMS_MAX_PASSWORD_COMPLEXITY_RULES, 4096u);
     EXPECT_GE(DBMS_MAX_PASSWORD_COMPLEXITY_RULES, 16u);
 }
+
+TEST(SanitizeUntrustedServerString, PreservesAnAntalyaProtocolMarker)
+{
+    /// The client parses the marker only after sanitizing, so it must pass through byte for byte.
+    String name = AntalyaProtocol::appendMarker("ClickHouse");
+    const String before = name;
+    sanitizeUntrustedServerString(name);
+    EXPECT_EQ(name, before);
+    EXPECT_EQ(AntalyaProtocol::parseMarker(name), static_cast<UInt64>(DBMS_ANTALYA_PROTOCOL_VERSION));
+}
+
+TEST(SanitizeUntrustedServerString, CannotForgeAnAntalyaProtocolMarker)
+{
+    /// Control bytes are replaced, never deleted, so a hostile name cannot collapse into a marker.
+    String hostile = "ClickHouse\x01 (antalya:\x02" "1)";
+    sanitizeUntrustedServerString(hostile);
+    EXPECT_EQ(AntalyaProtocol::parseMarker(hostile), 0u);
+}

src/Core/AntalyaProtocol.cpp

@@ -0,0 +1,81 @@
+#include <Core/AntalyaProtocol.h>
+
+#include <Common/StringUtils.h>
+#include <Common/intExp10.h>
+
+#include <algorithm>
+
+
+namespace DB
+{
+
+namespace AntalyaProtocol
+{
+
+constexpr UInt64 MAX_VERSION = intExp10(static_cast<int>(MAX_MARKER_DIGITS)) - 1;
+
+static_assert(
+    DBMS_ANTALYA_PROTOCOL_VERSION >= 1 && DBMS_ANTALYA_PROTOCOL_VERSION <= MAX_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(MARKER_TERMINATOR);
+    return result;
+}
+
+UInt64 parseMarker(std::string_view name) noexcept
+{
+    if (name.empty() || name.back() != MARKER_TERMINATOR)
+        return 0;
+
+    const size_t close = name.size() - 1;
+
+    size_t first_digit = close;
+    while (first_digit > 0 && isNumericASCII(name[first_digit - 1]))
+    {
+        --first_digit;
+        if (close - first_digit > MAX_MARKER_DIGITS)
+            return 0;
+    }
+
+    const size_t digits = close - first_digit;
+    /// Exactly one spelling per version: no empty digit run, no leading zero.
+    if (digits == 0 || name[first_digit] == '0')
+        return 0;
+
+    if (first_digit < MARKER_PREFIX.size())
+        return 0;
+    if (name.substr(first_digit - MARKER_PREFIX.size(), MARKER_PREFIX.size()) != MARKER_PREFIX)
+        return 0;
+
+    UInt64 version = 0;
+    for (size_t i = first_digit; i < close; ++i)
+        version = version * 10 + static_cast<UInt64>(name[i] - '0');
+    return version;
+}
+
+UInt64 stripMarker(String & name)
+{
+    const UInt64 version = parseMarker(name);
+    if (version != 0)
+    {
+        /// Only the canonical spelling parses, so the digits on the wire are the ones `to_string` gives.
+        name.resize(name.size() - MARKER_PREFIX.size() - std::to_string(version).size() - 1);
+    }
+    return version;
+}
+
+UInt64 negotiate(UInt64 peer_version) noexcept
+{
+    return std::min<UInt64>(peer_version, DBMS_ANTALYA_PROTOCOL_VERSION);
+}
+
+}
+
+}

src/Core/AntalyaProtocol.h

@@ -0,0 +1,46 @@
+#pragma once
+
+#include <base/types.h>
+
+#include <string_view>
+
+
+namespace DB
+{
+
+/// Counter for Antalya-only wire features, independent of every upstream counter in
+/// `Core/ProtocolDefines.h`, so that a rebase can never renumber an Antalya feature. Bump it by one
+/// per feature. The client caps the server's value with `min(own, server)`, so the feature set must
+/// stay cumulative: a backport takes the whole contiguous range up to the value it needs, or does
+/// not bump at all. See `docs/en/antalya/protocol.md`.
+static constexpr auto DBMS_ANTALYA_PROTOCOL_VERSION = 1;
+
+namespace AntalyaProtocol
+{
+
+/// The version is advertised in the `ServerHello` name string - "ClickHouse (antalya:M)" - and
+/// nowhere else: the client `Hello` is never marked, because `client_name` reaches the peer's
+/// `system.query_log` and `validate_tcp_client_information`. See `docs/en/antalya/protocol.md`.
+
+/// " (antalya:" <digits> ")", where <digits> is [1-9][0-9]{0,8}.
+inline constexpr std::string_view MARKER_PREFIX = " (antalya:";
+inline constexpr char MARKER_TERMINATOR = ')';
+inline constexpr size_t MAX_MARKER_DIGITS = 9;
+inline constexpr size_t MAX_MARKER_SIZE = MARKER_PREFIX.size() + MAX_MARKER_DIGITS + 1;
+
+String appendMarker(std::string_view name);
+
+/// Returns 0 unless `name` ends with exactly one canonical marker. Scans at most `MAX_MARKER_SIZE`
+/// bytes back from the end, because the client runs it on a `server_name` it has not authenticated.
+UInt64 parseMarker(std::string_view name) noexcept;
+
+/// Removes a trailing canonical marker from `name` and returns the version it spelled; returns 0 and
+/// leaves `name` alone when there is none.
+UInt64 stripMarker(String & name);
+
+/// The version to speak with a server that advertised `peer_version`; 0 means it is not Antalya.
+UInt64 negotiate(UInt64 peer_version) noexcept;
+
+}
+
+}

src/Core/Protocol.h

@@ -68,6 +68,9 @@ const char JWT_AUTHENTICAION_MARKER[] = " JWT AUTHENTICATION ";
 
 };
 
+/// `client_name` for server-to-server connections, the counterpart of `DEFAULT_CLIENT_NAME`.
+static constexpr std::string_view SERVER_CLIENT_NAME = "server";
+
 namespace Protocol
 {
     /// Packet types that server transmits.

src/Core/tests/gtest_antalya_protocol.cpp

@@ -0,0 +1,76 @@
+#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, AppendThenParseRoundTrips)
+{
+    EXPECT_EQ(parseMarker(appendMarker("ClickHouse server")), static_cast<UInt64>(DBMS_ANTALYA_PROTOCOL_VERSION));
+}
+
+TEST(AntalyaProtocol, ParseRejectsNonCanonicalMarkers)
+{
+    const std::string_view rejected[] = {
+        "",
+        "ClickHouse server",
+        "ClickHouse client",
+        "(antalya:1)",
+        "ClickHouse server(antalya:1)",
+        "ClickHouse server (antalya:1",
+        "ClickHouse server (antalya:)",
+        "ClickHouse server (antalya:0)",
+        "ClickHouse server (antalya:01)",
+        "ClickHouse server (antalya:1234567890)",
+        "ClickHouse server (antalya:1) v2",
+        "ClickHouse server (ANTALYA:1)",
+        "ClickHouse server (antalya:1x)",
+        "ClickHouse server (antalya:1 )",
+        "antalya:1)",
+        ")",
+    };
+
+    for (const auto & name : rejected)
+        EXPECT_EQ(parseMarker(name), 0u) << "should not have parsed: " << name;
+}
+
+TEST(AntalyaProtocol, ParseTakesTheTrailingMarkerWhenRepeated)
+{
+    EXPECT_EQ(parseMarker("ClickHouse server (antalya:99) (antalya:1)"), 1u);
+}
+
+TEST(AntalyaProtocol, StripMarkerRemovesOnlyTheMarker)
+{
+    String marked = "ClickHouse server (antalya:42)";
+    EXPECT_EQ(stripMarker(marked), 42u);
+    EXPECT_EQ(marked, "ClickHouse server");
+
+    String widest = "ClickHouse (antalya:999999999)";
+    EXPECT_EQ(stripMarker(widest), 999999999u);
+    EXPECT_EQ(widest, "ClickHouse");
+
+    String plain = "ClickHouse server";
+    EXPECT_EQ(stripMarker(plain), 0u);
+    EXPECT_EQ(plain, "ClickHouse server");
+
+    String malformed = "ClickHouse server (antalya:01)";
+    EXPECT_EQ(stripMarker(malformed), 0u);
+    EXPECT_EQ(malformed, "ClickHouse server (antalya:01)");
+}
+
+TEST(AntalyaProtocol, NegotiateCapsToOurVersion)
+{
+    EXPECT_EQ(negotiate(0), 0u);
+    EXPECT_EQ(negotiate(DBMS_ANTALYA_PROTOCOL_VERSION), static_cast<UInt64>(DBMS_ANTALYA_PROTOCOL_VERSION));
+    EXPECT_EQ(negotiate(DBMS_ANTALYA_PROTOCOL_VERSION + 1), static_cast<UInt64>(DBMS_ANTALYA_PROTOCOL_VERSION));
+    EXPECT_EQ(negotiate(999999999), static_cast<UInt64>(DBMS_ANTALYA_PROTOCOL_VERSION));
+}