table revision number

This commit is contained in:
Filipriec
2026-08-29 08:15:49 +02:00
parent 96f8d7bf39
commit fead8685b6
8 changed files with 33 additions and 4 deletions

2
client

Submodule client updated: bcd298b3f1...1d3e75b150

View File

@@ -143,6 +143,8 @@ message GetTableImportDescriptorResponse {
string storage_profile_name = 2; string storage_profile_name = 2;
string table_name = 3; string table_name = 3;
int64 table_id = 4; int64 table_id = 4;
// Version of this import contract. Send it as expected_table_revision on
// every staged chunk so schema/rule changes refuse stale prepared data.
int64 table_revision = 5; int64 table_revision = 5;
repeated ImportFieldDescriptor fields = 6; repeated ImportFieldDescriptor fields = 6;
} }

View File

@@ -199,6 +199,9 @@ message StageTableDataImportRequest {
string import_id = 1; string import_id = 1;
string table_name = 2; string table_name = 2;
repeated TableDataImportRow rows = 3; repeated TableDataImportRow rows = 3;
// Import-contract revision returned by GetTableImportDescriptor. A zero
// value is accepted for older clients and snapshots the current revision.
int64 expected_table_revision = 4;
} }
message TableDataImportRow { message TableDataImportRow {

Binary file not shown.

View File

@@ -135,6 +135,8 @@ pub struct GetTableImportDescriptorResponse {
pub table_name: ::prost::alloc::string::String, pub table_name: ::prost::alloc::string::String,
#[prost(int64, tag = "4")] #[prost(int64, tag = "4")]
pub table_id: i64, pub table_id: i64,
/// Version of this import contract. Send it as expected_table_revision on
/// every staged chunk so schema/rule changes refuse stale prepared data.
#[prost(int64, tag = "5")] #[prost(int64, tag = "5")]
pub table_revision: i64, pub table_revision: i64,
#[prost(message, repeated, tag = "6")] #[prost(message, repeated, tag = "6")]

View File

@@ -95,6 +95,10 @@ pub struct StageTableDataImportRequest {
pub table_name: ::prost::alloc::string::String, pub table_name: ::prost::alloc::string::String,
#[prost(message, repeated, tag = "3")] #[prost(message, repeated, tag = "3")]
pub rows: ::prost::alloc::vec::Vec<TableDataImportRow>, pub rows: ::prost::alloc::vec::Vec<TableDataImportRow>,
/// Import-contract revision returned by GetTableImportDescriptor. A zero
/// value is accepted for older clients and snapshots the current revision.
#[prost(int64, tag = "4")]
pub expected_table_revision: i64,
} }
#[derive(Clone, PartialEq, ::prost::Message)] #[derive(Clone, PartialEq, ::prost::Message)]
pub struct TableDataImportRow { pub struct TableDataImportRow {

2
server

Submodule server updated: e4392276bc...3b4f7ee71e

View File

@@ -12,7 +12,7 @@ use axum_extra::extract::Form;
use crate::{ use crate::{
AppState, AppState,
definitions::{ definitions::{
table_structure::GetTableStructureRequest, table_structure::{GetTableImportDescriptorRequest, GetTableStructureRequest},
tables_data::{ tables_data::{
AbortTableDataImportRequest, BeginTableDataImportRequest, CommitTableDataImportRequest, AbortTableDataImportRequest, BeginTableDataImportRequest, CommitTableDataImportRequest,
StageTableDataImportRequest, TableDataImportRow, StageTableDataImportRequest, TableDataImportRow,
@@ -52,6 +52,7 @@ const CHUNK_ROWS: usize = 1_000;
/// a refusal on the next click instead of a value written somewhere else. /// a refusal on the next click instead of a value written somewhere else.
struct Destination { struct Destination {
table_name: String, table_name: String,
table_revision: i64,
columns: Vec<DestinationColumn>, columns: Vec<DestinationColumn>,
types: HashMap<String, String>, types: HashMap<String, String>,
} }
@@ -258,6 +259,7 @@ pub(crate) async fn import_csv(
id: id.clone(), id: id.clone(),
profile_name, profile_name,
table_name: destination.table_name.clone(), table_name: destination.table_name.clone(),
table_revision: destination.table_revision,
rows: converted, rows: converted,
})); }));
@@ -363,6 +365,7 @@ struct Running {
id: String, id: String,
profile_name: String, profile_name: String,
table_name: String, table_name: String,
table_revision: i64,
rows: Vec<TableDataImportRow>, rows: Vec<TableDataImportRow>,
} }
@@ -390,6 +393,7 @@ async fn run_import(job: Running) {
import_id: import_id.clone(), import_id: import_id.clone(),
table_name: job.table_name.clone(), table_name: job.table_name.clone(),
rows: chunk.to_vec(), rows: chunk.to_vec(),
expected_table_revision: job.table_revision,
}; };
let Ok(request) = authenticated_request(&job.headers, request) else { let Ok(request) = authenticated_request(&job.headers, request) else {
abort_import(&job, &import_id).await; abort_import(&job, &import_id).await;
@@ -522,11 +526,24 @@ async fn destination(
)); ));
} }
let descriptor_request = GetTableImportDescriptorRequest {
profile_name: profile_name.clone(),
table_name: table_name.clone(),
};
let mut structures = state.structures.clone();
let descriptor = structures
.get_table_import_descriptor(
authenticated_request(headers, descriptor_request)
.map_err(|_| Redirect::to("/login").into_response())?,
)
.await
.map_err(|error| grpc_error(headers, &error))?
.into_inner();
let request = GetTableStructureRequest { let request = GetTableStructureRequest {
profile_name, profile_name,
table_names: vec![table_name.clone()], table_names: vec![table_name.clone()],
}; };
let mut structures = state.structures.clone();
let structure = structures let structure = structures
.get_table_structure( .get_table_structure(
authenticated_request(headers, request) authenticated_request(headers, request)
@@ -552,6 +569,7 @@ async fn destination(
} }
Ok(Destination { Ok(Destination {
table_name, table_name,
table_revision: descriptor.table_revision,
types: column_types(&structure), types: column_types(&structure),
columns, columns,
}) })