fixing search indexing2

common/proto/search.proto

@@ -13,6 +13,32 @@ service SearchHealth {
   rpc GetHealth(komp_ac.common.Empty) returns (SearchHealthResponse);
 }
 
+service SearchAdmin {
+  rpc ReindexTableChunk(ReindexTableChunkRequest) returns (ReindexResponse);
+  rpc ReindexTable(ReindexTableRequest) returns (ReindexResponse);
+  rpc ReindexProfile(ReindexProfileRequest) returns (ReindexResponse);
+}
+
+message ReindexTableChunkRequest {
+  string profile_name = 1;
+  string table_name = 2;
+  repeated int64 row_ids = 3;
+}
+
+message ReindexTableRequest {
+  string profile_name = 1;
+  string table_name = 2;
+}
+
+message ReindexProfileRequest {
+  string profile_name = 1;
+}
+
+message ReindexResponse {
+  uint64 row_count = 1;
+  uint64 affected_tables = 2;
+}
+
 message SearchIndexError {
   int64 job_id = 1;
   string profile_name = 2;

common/src/proto/komp_ac.search.rs

@@ -1,6 +1,38 @@
 // This file is @generated by prost-build.
 #[derive(serde::Serialize, serde::Deserialize)]
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct ReindexTableChunkRequest {
+    #[prost(string, tag = "1")]
+    pub profile_name: ::prost::alloc::string::String,
+    #[prost(string, tag = "2")]
+    pub table_name: ::prost::alloc::string::String,
+    #[prost(int64, repeated, tag = "3")]
+    pub row_ids: ::prost::alloc::vec::Vec<i64>,
+}
+#[derive(serde::Serialize, serde::Deserialize)]
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct ReindexTableRequest {
+    #[prost(string, tag = "1")]
+    pub profile_name: ::prost::alloc::string::String,
+    #[prost(string, tag = "2")]
+    pub table_name: ::prost::alloc::string::String,
+}
+#[derive(serde::Serialize, serde::Deserialize)]
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct ReindexProfileRequest {
+    #[prost(string, tag = "1")]
+    pub profile_name: ::prost::alloc::string::String,
+}
+#[derive(serde::Serialize, serde::Deserialize)]
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct ReindexResponse {
+    #[prost(uint64, tag = "1")]
+    pub row_count: u64,
+    #[prost(uint64, tag = "2")]
+    pub affected_tables: u64,
+}
+#[derive(serde::Serialize, serde::Deserialize)]
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct SearchIndexError {
     #[prost(int64, tag = "1")]
     pub job_id: i64,
@@ -871,3 +903,448 @@ pub mod search_health_server {
         const NAME: &'static str = SERVICE_NAME;
     }
 }
+/// Generated client implementations.
+pub mod search_admin_client {
+    #![allow(
+        unused_variables,
+        dead_code,
+        missing_docs,
+        clippy::wildcard_imports,
+        clippy::let_unit_value,
+    )]
+    use tonic::codegen::*;
+    use tonic::codegen::http::Uri;
+    #[derive(Debug, Clone)]
+    pub struct SearchAdminClient<T> {
+        inner: tonic::client::Grpc<T>,
+    }
+    impl SearchAdminClient<tonic::transport::Channel> {
+        /// Attempt to create a new client by connecting to a given endpoint.
+        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
+        where
+            D: TryInto<tonic::transport::Endpoint>,
+            D::Error: Into<StdError>,
+        {
+            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
+            Ok(Self::new(conn))
+        }
+    }
+    impl<T> SearchAdminClient<T>
+    where
+        T: tonic::client::GrpcService<tonic::body::Body>,
+        T::Error: Into<StdError>,
+        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
+        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
+    {
+        pub fn new(inner: T) -> Self {
+            let inner = tonic::client::Grpc::new(inner);
+            Self { inner }
+        }
+        pub fn with_origin(inner: T, origin: Uri) -> Self {
+            let inner = tonic::client::Grpc::with_origin(inner, origin);
+            Self { inner }
+        }
+        pub fn with_interceptor<F>(
+            inner: T,
+            interceptor: F,
+        ) -> SearchAdminClient<InterceptedService<T, F>>
+        where
+            F: tonic::service::Interceptor,
+            T::ResponseBody: Default,
+            T: tonic::codegen::Service<
+                http::Request<tonic::body::Body>,
+                Response = http::Response<
+                    <T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
+                >,
+            >,
+            <T as tonic::codegen::Service<
+                http::Request<tonic::body::Body>,
+            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
+        {
+            SearchAdminClient::new(InterceptedService::new(inner, interceptor))
+        }
+        /// Compress requests with the given encoding.
+        ///
+        /// This requires the server to support it otherwise it might respond with an
+        /// error.
+        #[must_use]
+        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.inner = self.inner.send_compressed(encoding);
+            self
+        }
+        /// Enable decompressing responses.
+        #[must_use]
+        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.inner = self.inner.accept_compressed(encoding);
+            self
+        }
+        /// Limits the maximum size of a decoded message.
+        ///
+        /// Default: `4MB`
+        #[must_use]
+        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
+            self.inner = self.inner.max_decoding_message_size(limit);
+            self
+        }
+        /// Limits the maximum size of an encoded message.
+        ///
+        /// Default: `usize::MAX`
+        #[must_use]
+        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
+            self.inner = self.inner.max_encoding_message_size(limit);
+            self
+        }
+        pub async fn reindex_table_chunk(
+            &mut self,
+            request: impl tonic::IntoRequest<super::ReindexTableChunkRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::ReindexResponse>,
+            tonic::Status,
+        > {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| {
+                    tonic::Status::unknown(
+                        format!("Service was not ready: {}", e.into()),
+                    )
+                })?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static(
+                "/komp_ac.search.SearchAdmin/ReindexTableChunk",
+            );
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(
+                    GrpcMethod::new("komp_ac.search.SearchAdmin", "ReindexTableChunk"),
+                );
+            self.inner.unary(req, path, codec).await
+        }
+        pub async fn reindex_table(
+            &mut self,
+            request: impl tonic::IntoRequest<super::ReindexTableRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::ReindexResponse>,
+            tonic::Status,
+        > {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| {
+                    tonic::Status::unknown(
+                        format!("Service was not ready: {}", e.into()),
+                    )
+                })?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static(
+                "/komp_ac.search.SearchAdmin/ReindexTable",
+            );
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("komp_ac.search.SearchAdmin", "ReindexTable"));
+            self.inner.unary(req, path, codec).await
+        }
+        pub async fn reindex_profile(
+            &mut self,
+            request: impl tonic::IntoRequest<super::ReindexProfileRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::ReindexResponse>,
+            tonic::Status,
+        > {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| {
+                    tonic::Status::unknown(
+                        format!("Service was not ready: {}", e.into()),
+                    )
+                })?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static(
+                "/komp_ac.search.SearchAdmin/ReindexProfile",
+            );
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("komp_ac.search.SearchAdmin", "ReindexProfile"));
+            self.inner.unary(req, path, codec).await
+        }
+    }
+}
+/// Generated server implementations.
+pub mod search_admin_server {
+    #![allow(
+        unused_variables,
+        dead_code,
+        missing_docs,
+        clippy::wildcard_imports,
+        clippy::let_unit_value,
+    )]
+    use tonic::codegen::*;
+    /// Generated trait containing gRPC methods that should be implemented for use with SearchAdminServer.
+    #[async_trait]
+    pub trait SearchAdmin: std::marker::Send + std::marker::Sync + 'static {
+        async fn reindex_table_chunk(
+            &self,
+            request: tonic::Request<super::ReindexTableChunkRequest>,
+        ) -> std::result::Result<tonic::Response<super::ReindexResponse>, tonic::Status>;
+        async fn reindex_table(
+            &self,
+            request: tonic::Request<super::ReindexTableRequest>,
+        ) -> std::result::Result<tonic::Response<super::ReindexResponse>, tonic::Status>;
+        async fn reindex_profile(
+            &self,
+            request: tonic::Request<super::ReindexProfileRequest>,
+        ) -> std::result::Result<tonic::Response<super::ReindexResponse>, tonic::Status>;
+    }
+    #[derive(Debug)]
+    pub struct SearchAdminServer<T> {
+        inner: Arc<T>,
+        accept_compression_encodings: EnabledCompressionEncodings,
+        send_compression_encodings: EnabledCompressionEncodings,
+        max_decoding_message_size: Option<usize>,
+        max_encoding_message_size: Option<usize>,
+    }
+    impl<T> SearchAdminServer<T> {
+        pub fn new(inner: T) -> Self {
+            Self::from_arc(Arc::new(inner))
+        }
+        pub fn from_arc(inner: Arc<T>) -> Self {
+            Self {
+                inner,
+                accept_compression_encodings: Default::default(),
+                send_compression_encodings: Default::default(),
+                max_decoding_message_size: None,
+                max_encoding_message_size: None,
+            }
+        }
+        pub fn with_interceptor<F>(
+            inner: T,
+            interceptor: F,
+        ) -> InterceptedService<Self, F>
+        where
+            F: tonic::service::Interceptor,
+        {
+            InterceptedService::new(Self::new(inner), interceptor)
+        }
+        /// Enable decompressing requests with the given encoding.
+        #[must_use]
+        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.accept_compression_encodings.enable(encoding);
+            self
+        }
+        /// Compress responses with the given encoding, if the client supports it.
+        #[must_use]
+        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.send_compression_encodings.enable(encoding);
+            self
+        }
+        /// Limits the maximum size of a decoded message.
+        ///
+        /// Default: `4MB`
+        #[must_use]
+        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
+            self.max_decoding_message_size = Some(limit);
+            self
+        }
+        /// Limits the maximum size of an encoded message.
+        ///
+        /// Default: `usize::MAX`
+        #[must_use]
+        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
+            self.max_encoding_message_size = Some(limit);
+            self
+        }
+    }
+    impl<T, B> tonic::codegen::Service<http::Request<B>> for SearchAdminServer<T>
+    where
+        T: SearchAdmin,
+        B: Body + std::marker::Send + 'static,
+        B::Error: Into<StdError> + std::marker::Send + 'static,
+    {
+        type Response = http::Response<tonic::body::Body>;
+        type Error = std::convert::Infallible;
+        type Future = BoxFuture<Self::Response, Self::Error>;
+        fn poll_ready(
+            &mut self,
+            _cx: &mut Context<'_>,
+        ) -> Poll<std::result::Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+        fn call(&mut self, req: http::Request<B>) -> Self::Future {
+            match req.uri().path() {
+                "/komp_ac.search.SearchAdmin/ReindexTableChunk" => {
+                    #[allow(non_camel_case_types)]
+                    struct ReindexTableChunkSvc<T: SearchAdmin>(pub Arc<T>);
+                    impl<
+                        T: SearchAdmin,
+                    > tonic::server::UnaryService<super::ReindexTableChunkRequest>
+                    for ReindexTableChunkSvc<T> {
+                        type Response = super::ReindexResponse;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::ReindexTableChunkRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as SearchAdmin>::reindex_table_chunk(&inner, request)
+                                    .await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = ReindexTableChunkSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/komp_ac.search.SearchAdmin/ReindexTable" => {
+                    #[allow(non_camel_case_types)]
+                    struct ReindexTableSvc<T: SearchAdmin>(pub Arc<T>);
+                    impl<
+                        T: SearchAdmin,
+                    > tonic::server::UnaryService<super::ReindexTableRequest>
+                    for ReindexTableSvc<T> {
+                        type Response = super::ReindexResponse;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::ReindexTableRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as SearchAdmin>::reindex_table(&inner, request).await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = ReindexTableSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/komp_ac.search.SearchAdmin/ReindexProfile" => {
+                    #[allow(non_camel_case_types)]
+                    struct ReindexProfileSvc<T: SearchAdmin>(pub Arc<T>);
+                    impl<
+                        T: SearchAdmin,
+                    > tonic::server::UnaryService<super::ReindexProfileRequest>
+                    for ReindexProfileSvc<T> {
+                        type Response = super::ReindexResponse;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::ReindexProfileRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as SearchAdmin>::reindex_profile(&inner, request).await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = ReindexProfileSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                _ => {
+                    Box::pin(async move {
+                        let mut response = http::Response::new(
+                            tonic::body::Body::default(),
+                        );
+                        let headers = response.headers_mut();
+                        headers
+                            .insert(
+                                tonic::Status::GRPC_STATUS,
+                                (tonic::Code::Unimplemented as i32).into(),
+                            );
+                        headers
+                            .insert(
+                                http::header::CONTENT_TYPE,
+                                tonic::metadata::GRPC_CONTENT_TYPE,
+                            );
+                        Ok(response)
+                    })
+                }
+            }
+        }
+    }
+    impl<T> Clone for SearchAdminServer<T> {
+        fn clone(&self) -> Self {
+            let inner = self.inner.clone();
+            Self {
+                inner,
+                accept_compression_encodings: self.accept_compression_encodings,
+                send_compression_encodings: self.send_compression_encodings,
+                max_decoding_message_size: self.max_decoding_message_size,
+                max_encoding_message_size: self.max_encoding_message_size,
+            }
+        }
+    }
+    /// Generated gRPC service name
+    pub const SERVICE_NAME: &str = "komp_ac.search.SearchAdmin";
+    impl<T> tonic::server::NamedService for SearchAdminServer<T> {
+        const NAME: &'static str = SERVICE_NAME;
+    }
+}

search/src/lib.rs

@@ -160,6 +160,14 @@ impl SearcherService {
         }
     }
 
+    pub fn invalidate_profile(&self, profile_name: &str) -> Result<(), Status> {
+        self.profiles
+            .lock()
+            .map_err(|_| Status::internal("Profile index cache lock poisoned"))?
+            .remove(profile_name);
+        Ok(())
+    }
+
     async fn run_rpc(
         &self,
         request: Request<SearchRequest>,

web/locales/cs/main.ftl

@@ -659,6 +659,7 @@ import-th-status = Stav
 import-source-none = Přeskočit tento sloupec souboru
 import-source-column = Sloupec { $position }
 import-required-short = povinný
+import-verification-short = ověřit FROM
 import-status-mapped = Namapováno
 import-status-duplicate = Stejný název je použit dvakrát
 import-status-unmapped = Nenamapováno

web/locales/en/main.ftl

@@ -648,6 +648,7 @@ import-th-status = Status
 import-source-none = Skip this file column
 import-source-column = Column { $position }
 import-required-short = required
+import-verification-short = verify FROM
 import-status-mapped = Mapped
 import-status-duplicate = Same name appears twice
 import-status-unmapped = Not mapped

web/locales/sk/main.ftl

@@ -657,6 +657,7 @@ import-th-status = Stav
 import-source-none = Preskočiť tento stĺpec súboru
 import-source-column = Stĺpec { $position }
 import-required-short = povinný
+import-verification-short = overiť FROM
 import-status-mapped = Namapované
 import-status-duplicate = Rovnaký názov je použitý dvakrát
 import-status-unmapped = Nenamapované

web/src/pages/import_export/import/destination.rs

@@ -1,10 +1,14 @@
-//! The columns an import may write into, and how the form names one.
+//! The columns an import may write into or verify, and how the form names one.
 //!
 //! The review is source-first: every file position gets one row and chooses a
 //! destination by stable identity. Duplicate destination choices are prevented
 //! in the browser and refused again by the Rust mapping validation.
 
-use crate::definitions::table_structure::{TableColumn, TableStructureResponse};
+use std::collections::HashSet;
+
+use crate::definitions::table_structure::{
+    ImportFieldDescriptor, TableColumn, TableStructureResponse,
+};
 
 use super::super::common::schema::{is_importable_system_column, is_system_column};
 
@@ -50,7 +54,7 @@ impl DestinationKey {
     }
 }
 
-/// One column an import may write into.
+/// One column an import may write into or use to verify a linked projection.
 #[derive(Clone, Debug)]
 pub(crate) struct DestinationColumn {
     pub key: DestinationKey,
@@ -61,27 +65,38 @@ pub(crate) struct DestinationColumn {
     /// column may have a default, and `information_schema` does not say which
     /// do, so refusing here would block imports the server would accept.
     pub required: bool,
+    /// A linked projection is not written by the import. When mapped, its CSV
+    /// value is carried to the server as an assertion against the value copied
+    /// through its LINK column.
+    pub verification: bool,
 }
 
-/// The columns of `schema` an import may write into, in the order the table
-/// declares them.
+/// The columns of `schema` an import may write into or verify, in the order the
+/// table declares them.
 ///
 /// The exclusions are the server's own flags rather than a guess:
 ///
 /// * `is_primary_key` — `id` comes from a sequence.
-/// * `read_only` — set for a quantity-ledger column and for a link projection,
-///   and an insert naming either is refused. Accounting columns are marked
-///   `generated` but *not* read-only, so they stay: they are a user's to fill
-///   in.
+/// * `read_only` — ordinary read-only columns remain excluded. A linked
+///   projection is the exception: it is offered as an optional verification
+///   destination, although the import still never writes it. Accounting
+///   columns are marked `generated` but *not* read-only, so they stay: they are
+///   a user's to fill in.
 /// * system columns, except the ones an insert actually takes. `deleted` is
 ///   offered, because writing it is how a file that recorded deleted rows loads
 ///   back as deleted rows; `row_revision` and `created_at` are not, because the
 ///   server assigns them and answers `Invalid column` to anything else.
-pub(crate) fn destination_columns(schema: &TableStructureResponse) -> Vec<DestinationColumn> {
+pub(crate) fn destination_columns(
+    schema: &TableStructureResponse,
+    projection_column_ids: &HashSet<i64>,
+) -> Vec<DestinationColumn> {
     schema
         .columns
         .iter()
-        .filter(|column| !column.is_primary_key && !column.read_only)
+        .filter(|column| {
+            !column.is_primary_key
+                && (!column.read_only || projection_column_ids.contains(&column.column_id))
+        })
         .filter(|column| {
             !is_system_column(&column.name) || is_importable_system_column(&column.name)
         })
@@ -92,10 +107,35 @@ pub(crate) fn destination_columns(schema: &TableStructureResponse) -> Vec<Destin
             // `deleted` is NOT NULL with a default on every managed table —
             // reporting it would warn about every import ever prepared.
             required: !column.is_nullable && !is_system_column(&column.name),
+            verification: projection_column_ids.contains(&column.column_id),
         })
         .collect()
 }
 
+/// Stable column identities of read-only outputs whose recorded source begins
+/// with an actual LINK column's complete alias. Looking at punctuation alone
+/// is insufficient because aliases themselves may contain dots.
+pub(crate) fn projection_verification_column_ids(
+    fields: &[ImportFieldDescriptor],
+) -> HashSet<i64> {
+    let link_prefixes = fields
+        .iter()
+        .filter(|field| field.link.is_some())
+        .map(|field| format!("{}.", field.name))
+        .collect::<Vec<_>>();
+    fields
+        .iter()
+        .filter(|field| {
+            !field.writable
+                && field.generated
+                && link_prefixes
+                    .iter()
+                    .any(|prefix| field.generated_from.starts_with(prefix))
+        })
+        .map(|field| field.column_id)
+        .collect()
+}
+
 /// The column a posted key names, in the table as it stands right now.
 ///
 /// `None` means the table no longer has it — dropped, or made read-only, while
@@ -113,6 +153,10 @@ pub(crate) fn resolve<'a>(
 mod tests {
     use super::*;
 
+    fn destinations(schema: &TableStructureResponse) -> Vec<DestinationColumn> {
+        destination_columns(schema, &HashSet::from([92]))
+    }
+
     fn column(name: &str, column_id: i64) -> TableColumn {
         TableColumn {
             name: name.to_string(),
@@ -149,21 +193,62 @@ mod tests {
                     generated_from: "accounting".to_string(),
                     ..column("debit", 81)
                 },
+                // A linked projection is visible for verification, but remains
+                // distinguished from columns the import writes.
+                TableColumn {
+                    generated: true,
+                    read_only: true,
+                    generated_from: "customer.name".to_string(),
+                    ..column("customer_name", 92)
+                },
                 column("created_at", 0),
             ],
         }
     }
 
+    #[test]
+    fn only_outputs_reached_through_real_links_are_verifications() {
+        let fields = vec![
+            ImportFieldDescriptor {
+                column_id: 10,
+                name: "customer.link".to_string(),
+                link: Some(Default::default()),
+                ..Default::default()
+            },
+            ImportFieldDescriptor {
+                column_id: 92,
+                name: "customer_name".to_string(),
+                generated: true,
+                generated_from: "customer.link.name".to_string(),
+                ..Default::default()
+            },
+            ImportFieldDescriptor {
+                column_id: 93,
+                name: "phone_country".to_string(),
+                generated: true,
+                generated_from: "phone.with.dot".to_string(),
+                ..Default::default()
+            },
+        ];
+
+        assert_eq!(
+            projection_verification_column_ids(&fields),
+            HashSet::from([92])
+        );
+    }
+
     #[test]
     fn only_the_columns_an_insert_takes_are_offered() {
-        let columns = destination_columns(&schema());
+        let columns = destinations(&schema());
         assert_eq!(
             columns
                 .iter()
                 .map(|column| column.name.as_str())
                 .collect::<Vec<_>>(),
-            vec!["deleted", "number", "debit"]
+            vec!["deleted", "number", "debit", "customer_name"]
         );
+        assert!(!columns[2].verification);
+        assert!(columns[3].verification);
     }
 
     /// A user column is identified by the id the server gave it, so a rename
@@ -171,7 +256,7 @@ mod tests {
     /// id and no rename, so its name is identity enough.
     #[test]
     fn a_user_column_is_named_by_its_stable_id_and_a_system_one_by_its_name() {
-        let columns = destination_columns(&schema());
+        let columns = destinations(&schema());
         assert_eq!(columns[0].key.encode(), "system:deleted");
         assert_eq!(columns[1].key.encode(), "id:42");
 
@@ -187,22 +272,22 @@ mod tests {
     /// column rather than the name.
     #[test]
     fn a_renamed_column_is_still_the_same_destination() {
-        let posted = destination_columns(&schema())[1].key.encode();
+        let posted = destinations(&schema())[1].key.encode();
 
         let mut renamed = schema();
         renamed.columns[3].name = "invoice_number".to_string();
-        let columns = destination_columns(&renamed);
+        let columns = destinations(&renamed);
         let resolved =
             resolve(&columns, &posted).expect("column 42 is column 42 whatever it is called");
         assert_eq!(resolved.name, "invoice_number");
     }
 
-    /// A destination that is gone, or that the table has since made read-only,
-    /// resolves to nothing — which the caller has to refuse rather than guess
-    /// past.
+    /// A destination that is gone, or that the table has since become an
+    /// ordinary read-only column, resolves to nothing — which the caller has
+    /// to refuse rather than guess past.
     #[test]
     fn a_destination_the_table_no_longer_offers_resolves_to_nothing() {
-        let columns = destination_columns(&schema());
+        let columns = destinations(&schema());
         assert!(resolve(&columns, "id:57").is_none());
         assert!(resolve(&columns, "id:999").is_none());
         assert!(resolve(&columns, "system:row_revision").is_none());
@@ -213,7 +298,7 @@ mod tests {
     /// Only the user's own `NOT NULL` columns are reported as required.
     #[test]
     fn required_is_the_users_own_not_null_columns() {
-        let columns = destination_columns(&schema());
+        let columns = destinations(&schema());
         assert_eq!(
             columns
                 .iter()

web/src/pages/import_export/import/logic.rs

@@ -27,10 +27,13 @@ use super::{
         loader::LoadError,
         schema::{column_types, csv_value},
     },
-    destination::{DestinationColumn, destination_columns, resolve},
+    destination::{
+        DestinationColumn, destination_columns, projection_verification_column_ids, resolve,
+    },
     loader::load_page,
     prepare::{
-        Prepared, Source, canonical_csv, normalize_dates, prepare, read_mapping, read_source,
+        Assignment, Prepared, Source, canonical_csv, normalize_dates, prepare, read_mapping,
+        read_source,
     },
     progress::Outcome,
     state::{ImportForm, MappingRow, MappingStep, PreviewStep, SourceOption, Step},
@@ -58,11 +61,16 @@ struct Destination {
 }
 
 impl Destination {
-    /// The destination names, for the parts of the preparation that only need
-    /// to know which columns exist.
-    fn names(&self) -> Vec<String> {
+    /// Writable columns are always prepared so an unmapped form field keeps
+    /// its existing NULL/default behavior. Projection verifications only
+    /// travel when the user mapped a CSV value to them.
+    fn prepared_names(&self, assignments: &[Assignment]) -> Vec<String> {
         self.columns
             .iter()
+            .filter(|column| {
+                !column.verification
+                    || assignments.iter().any(|assignment| assignment.column == column.name)
+            })
             .map(|column| column.name.clone())
             .collect()
     }
@@ -132,6 +140,7 @@ pub(crate) async fn prepare_step(
                 key,
                 name: column.name.clone(),
                 required: column.required,
+                verification: column.verification,
                 chosen: chosen_index
                     .map(|index| (index + 1).to_string())
                     .unwrap_or_default(),
@@ -142,7 +151,10 @@ pub(crate) async fn prepare_step(
         })
         .collect::<Vec<_>>();
     let mapped = rows.iter().filter(|row| !row.chosen.is_empty()).count();
-    let attention = rows.len().saturating_sub(mapped);
+    let attention = rows
+        .iter()
+        .filter(|row| row.chosen.is_empty() && !row.verification)
+        .count();
 
     let step = Step::Mapping(MappingStep {
         table_name: destination.table_name.clone(),
@@ -196,9 +208,9 @@ pub(crate) async fn preview_step(
 
 /// POST /admin/import — the prepared rows, converted and inserted.
 ///
-/// From here on nothing about mapping exists any more: what is sent is the
-/// canonical import, and the server decides types, validations, scripts, links
-/// and permissions exactly as it does for any other insert.
+/// From here on the prepared column names carry the mapping. The server applies
+/// ordinary insert rules and treats any mapped linked projection as an
+/// assertion that must match the value reached through its link.
 pub(crate) async fn import_csv(
     State(state): State<AppState>,
     headers: HeaderMap,
@@ -557,7 +569,8 @@ async fn destination(
         .remove(&table_name)
         .ok_or_else(|| unavailable(headers, tr!(locale, "import-err-missing-structure")))?;
 
-    let columns = destination_columns(&structure);
+    let projection_column_ids = projection_verification_column_ids(&descriptor.fields);
+    let columns = destination_columns(&structure, &projection_column_ids);
     if columns.is_empty() {
         return Err(reject(
             headers,
@@ -610,7 +623,8 @@ async fn prepared(
 
     let assignments =
         read_mapping(locale, &chosen, &source).map_err(|message| reject(headers, message))?;
-    let mut prepared = prepare(&assignments, &source, &destination.names());
+    let prepared_names = destination.prepared_names(&assignments);
+    let mut prepared = prepare(&assignments, &source, &prepared_names);
     normalize_dates(locale, &mut prepared, &destination.types, form.date_format)
         .map_err(|message| reject(headers, message))?;
     Ok((destination, source, prepared))

web/src/pages/import_export/import/prepare.rs

@@ -181,10 +181,11 @@ impl Prepared {
 /// Applies the mapping: the values that were asked for, under the names they
 /// were asked for.
 ///
-/// `writable` is every destination column the table offers. Normal form fields
-/// are always included: an unmapped one becomes NULL, exactly as when the user
-/// leaves that field empty in the client form. An unmapped system column is
-/// omitted so its server-owned default still applies (`deleted = false`).
+/// `writable` contains every writable destination plus any mapped projection
+/// verification. Normal form fields are always included: an unmapped one
+/// becomes NULL, exactly as when the user leaves that field empty in the
+/// client form. An unmapped system column is omitted so its server-owned
+/// default still applies (`deleted = false`).
 pub(crate) fn prepare(
     assignments: &[Assignment],
     source: &Source,

web/src/pages/import_export/import/state.rs

@@ -188,6 +188,8 @@ pub(crate) struct MappingRow {
     pub key: String,
     pub name: String,
     pub required: bool,
+    /// This destination asserts a linked FROM value instead of writing it.
+    pub verification: bool,
     /// One-based source position, empty when intentionally not mapped.
     pub chosen: String,
     pub example: String,

web/src/pages/import_export/import/ui.rs

@@ -235,6 +235,7 @@ mod tests {
                     key: "id:42".to_string(),
                     name: "a".to_string(),
                     required: true,
+                    verification: false,
                     example: "value-a".to_string(),
                     chosen: "1".to_string(),
                 },
@@ -242,6 +243,7 @@ mod tests {
                     key: "id:57".to_string(),
                     name: "b".to_string(),
                     required: false,
+                    verification: false,
                     example: String::new(),
                     chosen: String::new(),
                 },
@@ -299,7 +301,11 @@ mod tests {
     /// to them or deliberately left unused.
     #[test]
     fn the_mapping_step_reviews_every_source_column_once() {
-        let html = render_step(&page(mapping()));
+        let mut step = mapping();
+        if let Step::Mapping(mapping) = &mut step {
+            mapping.rows[1].verification = true;
+        }
+        let html = render_step(&page(step));
 
         assert_eq!(html.matches(r#"name="source_position""#).count(), 2);
         assert_eq!(html.matches(r#"name="destination""#).count(), 2);
@@ -319,6 +325,7 @@ mod tests {
             html.contains(">Not mapped — leave empty</option>"),
             "{html}"
         );
+        assert!(html.contains("verify FROM"), "{html}");
     }
 
     /// The two sides are two lists, not one zipped table: a source chip carries

web/templates/pages/import_export/import/step.html

@@ -118,6 +118,7 @@
               <div class="destination-name">
                 <code>{{ row.name }}</code>
                 {% if row.required %}<span class="tag">{{ nav.tr("import-required-short") }}</span>{% endif %}
+                {% if row.verification %}<span class="tag">{{ nav.tr("import-verification-short") }}</span>{% endif %}
               </div>
               <span class="mapping-arrow" aria-hidden="true">←</span>
               <label>