Add a mechanism for converting SQL schemas based on negotiated version

cli/src/clients/admin_client.rs

@@ -27,8 +27,8 @@ use crate::clients::AdminClientInterface;
 use super::errors::ApiError;
 
 /// Min/max supported admin API versions
-pub const MIN_ADMIN_API_VERSION: AdminApiVersion = AdminApiVersion::V1;
-pub const MAX_ADMIN_API_VERSION: AdminApiVersion = AdminApiVersion::V1;
+pub const MIN_ADMIN_API_VERSION: AdminApiVersion = AdminApiVersion::V2;
+pub const MAX_ADMIN_API_VERSION: AdminApiVersion = AdminApiVersion::V2;
 
 #[derive(Error, Debug)]
 #[error(transparent)]
@@ -186,10 +186,13 @@ impl AdminClient {
             .request(method, path)
             .timeout(self.request_timeout);
 
-        match self.bearer_token.as_deref() {
+        let request_builder = match self.bearer_token.as_deref() {
             Some(token) => request_builder.bearer_auth(token),
             None => request_builder,
-        }
+        };
+
+        let (api_version_header, api_version) = self.admin_api_version.into();
+        request_builder.header(api_version_header, api_version)
     }
 
     /// Prepare a request builder that encodes the body as JSON.

cli/src/clients/datafusion_helpers.rs

@@ -16,7 +16,7 @@ use std::str::FromStr;
 
 use anyhow::Result;
 use arrow::array::{Array, ArrayAccessor, AsArray, StringArray};
-use arrow::datatypes::{ArrowTemporalType, Date64Type};
+use arrow::datatypes::ArrowTemporalType;
 use arrow::record_batch::RecordBatch;
 use arrow_convert::{ArrowDeserialize, ArrowField};
 use bytes::Bytes;
@@ -111,10 +111,17 @@ fn value_as_u64_opt(batch: &RecordBatch, col: usize, row: usize) -> Option<u64>
 }
 
 fn value_as_dt_opt(batch: &RecordBatch, col: usize, row: usize) -> Option<chrono::DateTime<Local>> {
-    batch
-        .column(col)
-        .as_primitive::<arrow::datatypes::Date64Type>()
-        .value_as_local_datetime_opt(row)
+    let col = batch.column(col);
+    match col.data_type() {
+        arrow::datatypes::DataType::Timestamp(arrow::datatypes::TimeUnit::Millisecond, _) => col
+            .as_primitive::<arrow::datatypes::TimestampMillisecondType>()
+            .value_as_local_datetime_opt(row),
+        // older versions of restate used Date64 instead of TimestampMillisecond
+        arrow::datatypes::DataType::Date64 => col
+            .as_primitive::<arrow::datatypes::Date64Type>()
+            .value_as_local_datetime_opt(row),
+        _ => panic!("Column is not a timestamp"),
+    }
 }
 
 #[derive(ValueEnum, Copy, Clone, Eq, Hash, PartialEq, Debug, Default)]
@@ -515,11 +522,7 @@ pub async fn get_service_status(
                     .column(2)
                     .as_primitive::<arrow::datatypes::Int64Type>()
                     .value(i);
-                let oldest_at = batch
-                    .column(3)
-                    .as_primitive::<arrow::datatypes::Date64Type>()
-                    .value_as_local_datetime_opt(i)
-                    .unwrap();
+                let oldest_at = value_as_dt_opt(&batch, 3, i).unwrap();
 
                 let oldest_invocation = batch.column(4).as_string::<i32>().value_string(i);
 
@@ -744,23 +747,66 @@ impl From<RestateDateTime> for DateTime<Local> {
     }
 }
 
+pub static TIMEZONE_UTC: std::sync::LazyLock<std::sync::Arc<str>> =
+    std::sync::LazyLock::new(|| std::sync::Arc::from("+00:00"));
+
 impl arrow_convert::field::ArrowField for RestateDateTime {
     type Type = Self;
 
     #[inline]
     fn data_type() -> arrow::datatypes::DataType {
-        arrow::datatypes::DataType::Date64
+        arrow::datatypes::DataType::Timestamp(
+            arrow::datatypes::TimeUnit::Millisecond,
+            Some(TIMEZONE_UTC.clone()),
+        )
     }
 }
 
 impl arrow_convert::deserialize::ArrowDeserialize for RestateDateTime {
-    type ArrayType = arrow::array::Date64Array;
+    type ArrayType = TimestampMillisecondArray;
 
     #[inline]
     fn arrow_deserialize(v: Option<i64>) -> Option<Self> {
-        v.and_then(arrow::temporal_conversions::as_datetime::<Date64Type>)
-            .map(|naive| Local.from_utc_datetime(&naive))
-            .map(RestateDateTime)
+        let timestamp = arrow::temporal_conversions::as_datetime::<
+            arrow::datatypes::TimestampMillisecondType,
+        >(v?)?;
+        Some(RestateDateTime(Local.from_utc_datetime(&timestamp)))
+    }
+}
+
+// This newtype is necessary to implement ArrowArray, which is implemented for TimestampNanosecond but not TimestampMillisecond for some reason
+#[repr(transparent)]
+struct TimestampMillisecondArray(arrow::array::TimestampMillisecondArray);
+
+impl TimestampMillisecondArray {
+    fn from_ref(v: &arrow::array::TimestampMillisecondArray) -> &Self {
+        // SAFETY: transmuting a single-element newtype struct with repr(transparent) is safe
+        unsafe { std::mem::transmute(v) }
+    }
+}
+
+impl arrow_convert::deserialize::ArrowArray for TimestampMillisecondArray {
+    type BaseArrayType = arrow::array::TimestampMillisecondArray;
+    #[inline]
+    fn iter_from_array_ref(
+        b: &dyn Array,
+    ) -> <Self as arrow_convert::deserialize::ArrowArrayIterable>::Iter<'_> {
+        let b = b.as_any().downcast_ref::<Self::BaseArrayType>().unwrap();
+        <Self as arrow_convert::deserialize::ArrowArrayIterable>::iter(
+            TimestampMillisecondArray::from_ref(b),
+        )
+    }
+}
+
+impl arrow_convert::deserialize::ArrowArrayIterable for TimestampMillisecondArray {
+    type Item<'a> = Option<
+        <arrow::datatypes::TimestampMillisecondType as arrow::array::ArrowPrimitiveType>::Native,
+    >;
+
+    type Iter<'a> = arrow::array::PrimitiveIter<'a, arrow::datatypes::TimestampMillisecondType>;
+
+    fn iter(&self) -> Self::Iter<'_> {
+        IntoIterator::into_iter(&self.0)
     }
 }
 

crates/admin-rest-model/src/version.rs

@@ -23,13 +23,46 @@ use std::ops::RangeInclusive;
 pub enum AdminApiVersion {
     Unknown = 0,
     V1 = 1,
+    V2 = 2,
+}
+
+impl From<AdminApiVersion> for (http::HeaderName, http::HeaderValue) {
+    fn from(value: AdminApiVersion) -> Self {
+        (
+            AdminApiVersion::HEADER_NAME,
+            http::HeaderValue::from(value.as_repr()),
+        )
+    }
 }
 
 impl AdminApiVersion {
+    const HEADER_NAME: http::HeaderName =
+        http::HeaderName::from_static("x-restate-admin-api-version");
+
     pub fn as_repr(&self) -> u16 {
         *self as u16
     }
 
+    pub fn from_headers(headers: &http::HeaderMap) -> Self {
+        let is_cli = matches!(headers.get(http::header::USER_AGENT), Some(value) if value.as_bytes().starts_with(b"restate-cli"));
+
+        match headers.get(Self::HEADER_NAME) {
+            Some(value) => match value.to_str() {
+                Ok(value) => match value.parse::<u16>() {
+                    Ok(value) => match Self::try_from(value) {
+                        Ok(value) => value,
+                        Err(_) => Self::Unknown,
+                    },
+                    Err(_) => Self::Unknown,
+                },
+                Err(_) => Self::Unknown,
+            },
+            // CLI didn't used to send the version header, but if we know its the CLI, then we can treat that as V1
+            None if is_cli => Self::V1,
+            None => Self::Unknown,
+        }
+    }
+
     pub fn choose_max_supported_version(
         client_versions: RangeInclusive<AdminApiVersion>,
         server_versions: RangeInclusive<u16>,

crates/admin/src/rest_api/version.rs

@@ -14,7 +14,7 @@ use restate_admin_rest_model::version::{AdminApiVersion, VersionInformation};
 
 /// Min/max supported admin api versions by the server
 pub const MIN_ADMIN_API_VERSION: AdminApiVersion = AdminApiVersion::V1;
-pub const MAX_ADMIN_API_VERSION: AdminApiVersion = AdminApiVersion::V1;
+pub const MAX_ADMIN_API_VERSION: AdminApiVersion = AdminApiVersion::V2;
 
 /// Version information endpoint
 #[openapi(