ahri_tre_lake/

dataset_loading.rs

//! Dataset table loading and export helpers for DuckLake relations.

use std::fs::{self, File, OpenOptions};
use std::io::{self, BufReader, BufWriter, Write};
use std::os::unix::fs::OpenOptionsExt;
use std::path::{Path, PathBuf};

use ahri_tre_tabular::{
    AhriTreArrowMetadata, AnnotatedArrowTable, ArrowIpcCompression, ArrowIpcWriteOptions,
    ArrowTable, annotate_arrow_table, write_ipc_file_to_writer, write_parquet_file,
};
use ahri_tre_types::{NcName, StudyId};
use uuid::Uuid;

use crate::{DuckLakeAdapter, DuckLakeOpenedCatalog, LakeError, ScratchAttempt};

/// File formats accepted when loading a dataset table from disk.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DatasetFileFormat {
    ArrowIpc,
    Csv,
    Json,
    Parquet,
    Xlsx,
}

impl DatasetFileFormat {
    /// Returns the conventional extension for this dataset file format.
    pub fn extension(self) -> &'static str {
        match self {
            Self::ArrowIpc => "arrow",
            Self::Csv => "csv",
            Self::Json => "json",
            Self::Parquet => "parquet",
            Self::Xlsx => "xlsx",
        }
    }
}

/// Options passed to DuckDB file readers when importing a dataset table.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DatasetFileReadOptions {
    pub header: bool,
    pub delimiter: char,
    pub null_strings: Vec<String>,
    pub json_format: String,
    pub sheet: Option<String>,
}

impl Default for DatasetFileReadOptions {
    fn default() -> Self {
        Self {
            header: true,
            delimiter: ',',
            null_strings: Vec::new(),
            json_format: "auto".to_string(),
            sheet: None,
        }
    }
}

/// Request to load one versioned dataset table from a source file.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LoadDatasetTableFromFileRequest {
    pub study_id: StudyId,
    pub dataset_name: NcName,
    pub major: i32,
    pub minor: i32,
    pub patch: i32,
    pub source_path: PathBuf,
    pub format: DatasetFileFormat,
    pub options: DatasetFileReadOptions,
}

/// One trusted in-memory CSV selected for atomic dataset replacement.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DatasetCsvSeed<'a> {
    pub dataset_name: NcName,
    pub contents: &'a [u8],
}

/// Request to materialize one versioned dataset table from a SQL query.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LoadDatasetTableFromSqlRequest {
    pub study_id: StudyId,
    pub dataset_name: NcName,
    pub major: i32,
    pub minor: i32,
    pub patch: i32,
    pub source_sql: String,
}

/// Output format for exporting a dataset table from DuckLake.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DatasetExportFormat {
    Csv,
    Ndjson,
    Parquet,
    ArrowIpc,
}

/// Request to export one versioned dataset table to a local file.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExportDatasetTableRequest {
    pub study_id: StudyId,
    pub dataset_name: NcName,
    pub major: i32,
    pub minor: i32,
    pub patch: i32,
    pub limit: Option<usize>,
    pub destination_path: PathBuf,
    pub format: DatasetExportFormat,
    pub compress: bool,
    pub arrow_metadata: Option<AhriTreArrowMetadata>,
}

/// Request to read a bounded preview of one versioned dataset table.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PreviewDatasetTableRequest {
    pub study_id: StudyId,
    pub dataset_name: NcName,
    pub major: i32,
    pub minor: i32,
    pub patch: i32,
    pub limit: usize,
}

/// Summary of a dataset table export operation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExportedDatasetTable {
    pub relation: DatasetTableRelation,
    pub destination_path: PathBuf,
    pub format: DatasetExportFormat,
    pub compressed: bool,
    pub row_count: usize,
    pub column_count: usize,
    pub exported_size_bytes: u64,
    pub warnings: Vec<String>,
}

/// Bounded row preview for one dataset table.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PreviewedDatasetTable {
    pub relation: DatasetTableRelation,
    pub columns: Vec<ImportedColumn>,
    pub rows: Vec<Vec<Option<String>>>,
    pub displayed_row_count: usize,
    pub total_row_count: usize,
    pub truncated: bool,
}

/// Request to reshape a REDCap EAV export into a versioned dataset table.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LoadRedcapFormDatasetRequest {
    pub study_id: StudyId,
    pub dataset_name: NcName,
    pub major: i32,
    pub minor: i32,
    pub patch: i32,
    pub source_eav_path: PathBuf,
    pub fields: Vec<RedcapEavField>,
    pub synthetic_columns: Vec<RedcapSyntheticColumn>,
}

/// REDCap field projection metadata used when pivoting EAV data.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RedcapEavField {
    pub name: String,
    pub value_type_id: ahri_tre_types::ValueTypeId,
    pub value_format: Option<String>,
    pub choices: Vec<RedcapEavChoice>,
}

/// Mapping from a REDCap raw choice value to the TRE categorical integer value.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RedcapEavChoice {
    pub redcap_value: String,
    pub value: i32,
}

/// Derived REDCap output column projected from an existing source column.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RedcapSyntheticColumn {
    pub source_column: String,
    pub output_column: String,
    pub value_type_id: ahri_tre_types::ValueTypeId,
    pub choices: Vec<RedcapEavChoice>,
}

/// DuckDB schema and table name for a versioned dataset table.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DatasetTableRelation {
    pub schema_name: String,
    pub table_name: String,
}

impl DatasetTableRelation {
    /// Returns the unquoted `schema.table` relation name for display.
    pub fn qualified_name(&self) -> String {
        format!("{}.{}", self.schema_name, self.table_name)
    }

    fn to_sql(&self) -> String {
        format!(
            "{}.{}",
            quote_identifier(&self.schema_name),
            quote_identifier(&self.table_name)
        )
    }
}

/// Summary of a dataset table loaded into DuckLake.
#[derive(Debug)]
pub struct LoadedDatasetTable {
    pub relation: DatasetTableRelation,
    pub row_count: usize,
    pub column_count: usize,
    pub column_names: Vec<String>,
    pub columns: Vec<ImportedColumn>,
}

/// DuckDB-reported shape for one imported dataset column.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ImportedColumn {
    pub name: String,
    pub duckdb_type: String,
}

impl DuckLakeAdapter {
    /// Validates a dataset file by scanning it with the same DuckDB reader used
    /// by governed dataset imports, without creating a persistent relation.
    ///
    /// # Errors
    ///
    /// Returns an error if the source file is missing or DuckDB cannot parse
    /// and scan the complete file with the requested format and options.
    pub fn validate_dataset_file(
        &self,
        connection: &duckdb::Connection,
        source_path: &Path,
        format: DatasetFileFormat,
        options: &DatasetFileReadOptions,
    ) -> Result<(), LakeError> {
        if !source_path.is_file() {
            return Err(LakeError::InvalidFileName {
                path: source_path.display().to_string(),
            });
        }
        if format == DatasetFileFormat::ArrowIpc {
            return Err(LakeError::InvalidMetadataValue {
                key: "format",
                value: "Arrow IPC validation requires adapter-managed conversion".to_string(),
            });
        }

        let relation_sql = file_relation_sql(source_path, format, options)?;
        connection.query_row(
            &format!("SELECT count(*) FROM ({relation_sql}) AS validated_dataset_file"),
            [],
            |_| Ok(()),
        )?;
        Ok(())
    }

    /// Atomically replaces the exact Study dataset set in an existing
    /// server-managed DuckLake catalog from trusted CSV inputs.
    ///
    /// # Errors
    ///
    /// Returns an error unless the catalog was opened as an existing
    /// server-managed catalog without an encryption override, or if private
    /// staging, dataset replacement, or exact-set verification fails.
    pub fn replace_server_managed_dataset_tables_from_csv(
        &self,
        opened: &DuckLakeOpenedCatalog,
        scratch: &ScratchAttempt,
        study_id: StudyId,
        seeds: &[DatasetCsvSeed<'_>],
        authority: &mut dyn ahri_tre_core::DatasetMaintenanceAuthority,
    ) -> Result<Vec<LoadedDatasetTable>, LakeError> {
        authority.ensure_exclusive()?;
        if opened.attach_plan.create_if_not_exists
            || opened.attach_plan.attach_description.contains("ENCRYPTED")
            || opened.attach_plan.configured_encryption_mode
                != ahri_tre_types::EncryptionMode::ServerManaged
            || opened.health.detected_encryption_mode
                != ahri_tre_types::EncryptionMode::ServerManaged
        {
            return Err(LakeError::InvalidMetadataValue {
                key: "dataset_seed_catalog",
                value: "existing server-managed catalog required".to_string(),
            });
        }
        self.replace_dataset_tables_from_csv(&opened.connection, scratch, study_id, seeds)
    }

    fn replace_dataset_tables_from_csv(
        &self,
        connection: &duckdb::Connection,
        scratch: &ScratchAttempt,
        study_id: StudyId,
        seeds: &[DatasetCsvSeed<'_>],
    ) -> Result<Vec<LoadedDatasetTable>, LakeError> {
        let retained = seeds
            .iter()
            .map(|seed| dataset_table_relation(study_id, seed.dataset_name.as_str(), 1, 0, 0))
            .collect::<Vec<_>>();
        if seeds.is_empty()
            || seeds.iter().any(|seed| seed.contents.is_empty())
            || retained
                .iter()
                .enumerate()
                .any(|(index, relation)| retained[..index].contains(relation))
        {
            return Err(LakeError::InvalidMetadataValue {
                key: "dataset_seed",
                value: "non-empty unique dataset inputs required".to_string(),
            });
        }

        connection.execute_batch("BEGIN TRANSACTION;")?;
        let replaced = (|| {
            // Exclusive maintenance replaces the contents of retained names too.
            // Ordinary writers still require an absent target before creation.
            self.retain_only_dataset_tables(connection, study_id, &[])?;
            let mut loaded = Vec::with_capacity(seeds.len());
            for seed in seeds {
                let relation =
                    dataset_table_relation(study_id, seed.dataset_name.as_str(), 1, 0, 0);
                let source_path = scratch.path().join(format!("{}.csv", relation.table_name));
                let mut file = OpenOptions::new()
                    .write(true)
                    .create_new(true)
                    .mode(0o600)
                    .open(&source_path)?;
                file.write_all(seed.contents)?;
                file.sync_all()?;
                loaded.push(self.load_dataset_table_from_file_with_creation_intent(
                    connection,
                    &LoadDatasetTableFromFileRequest {
                        study_id,
                        dataset_name: seed.dataset_name.clone(),
                        major: 1,
                        minor: 0,
                        patch: 0,
                        source_path,
                        format: DatasetFileFormat::Csv,
                        options: DatasetFileReadOptions::default(),
                    },
                    || Ok(()),
                )?);
            }

            let mut actual = study_dataset_tables(connection, &study_lake_schema(study_id))?;
            let mut expected = retained;
            actual.sort_by(|left, right| left.table_name.cmp(&right.table_name));
            expected.sort_by(|left, right| left.table_name.cmp(&right.table_name));
            if actual != expected {
                return Err(LakeError::InvalidMetadataValue {
                    key: "dataset_seed",
                    value: "exact dataset replacement verification failed".to_string(),
                });
            }
            Ok(loaded)
        })();

        match replaced {
            Ok(loaded) => {
                if let Err(error) = connection.execute_batch("COMMIT;") {
                    let _ = connection.execute_batch("ROLLBACK;");
                    return Err(error.into());
                }
                Ok(loaded)
            }
            Err(error) => {
                let _ = connection.execute_batch("ROLLBACK;");
                Err(error)
            }
        }
    }

    /// Removes versioned dataset tables in one Study namespace unless they are
    /// members of the supplied exact retained set.
    ///
    /// # Errors
    ///
    /// Returns an error if catalog inspection or a table drop fails.
    fn retain_only_dataset_tables(
        &self,
        connection: &duckdb::Connection,
        study_id: StudyId,
        retained: &[DatasetTableRelation],
    ) -> Result<Vec<DatasetTableRelation>, LakeError> {
        let schema_name = study_lake_schema(study_id);
        if retained
            .iter()
            .any(|relation| relation.schema_name != schema_name)
        {
            return Err(LakeError::InvalidMetadataValue {
                key: "retained_dataset_relation",
                value: "relation belongs to another Study namespace".to_string(),
            });
        }

        let existing = study_dataset_tables(connection, &schema_name)?;
        let mut removed = Vec::new();
        for relation in existing {
            if !retained.contains(&relation) {
                self.drop_dataset_table(connection, &relation)?;
                removed.push(relation);
            }
        }
        Ok(removed)
    }

    /// Loads a versioned dataset table from a CSV, JSON, Parquet, Arrow IPC, or XLSX file.
    ///
    /// # Errors
    ///
    /// Returns an error if the source file is missing, a version component is
    /// negative, DuckDB cannot read the file, or the target relation cannot be
    /// created.
    #[cfg(test)]
    fn load_dataset_table_from_file(
        &self,
        connection: &duckdb::Connection,
        request: &LoadDatasetTableFromFileRequest,
    ) -> Result<LoadedDatasetTable, LakeError> {
        self.load_dataset_table_from_file_with_creation_intent(connection, request, || Ok(()))
    }

    /// Loads only the exact output held by an active shared reservation.
    pub fn load_dataset_table_from_file_reserved(
        &self,
        connection: &duckdb::Connection,
        authority: &mut dyn ahri_tre_core::DatasetWriteAuthority,
        request: &LoadDatasetTableFromFileRequest,
    ) -> Result<LoadedDatasetTable, LakeError> {
        authorize_dataset_write(
            authority,
            request.study_id,
            &request.dataset_name,
            request.major,
            request.minor,
            request.patch,
        )?;
        self.load_dataset_table_from_file_with_creation_intent(connection, request, || {
            authority.record_creation_intent().map_err(Into::into)
        })
    }

    /// Observes one reserved load, including its existing rollback outcome.
    pub fn load_dataset_table_from_file_reserved_observed(
        &self,
        context: Option<ahri_tre_observability::CorrelationContext>,
        connection: &duckdb::Connection,
        authority: &mut dyn ahri_tre_core::DatasetWriteAuthority,
        request: &LoadDatasetTableFromFileRequest,
    ) -> Result<LoadedDatasetTable, LakeError> {
        let span = context.map(|context| context.span(ahri_tre_observability::Stage::LakeRead));
        let result = self.load_dataset_table_from_file_reserved(connection, authority, request);
        let evidence = result.as_ref().map(|loaded| {
            [
                ahri_tre_observability::Measurement::Rows(loaded.row_count as u64),
                ahri_tre_observability::Measurement::Columns(loaded.column_count as u64),
            ]
        });
        crate::error::finish_dataset_observation(
            span,
            &result,
            evidence
                .as_ref()
                .map(|counts| counts.as_slice())
                .unwrap_or_default(),
        );
        result
    }

    fn load_dataset_table_from_file_with_creation_intent(
        &self,
        connection: &duckdb::Connection,
        request: &LoadDatasetTableFromFileRequest,
        mut before_create: impl FnMut() -> Result<(), LakeError>,
    ) -> Result<LoadedDatasetTable, LakeError> {
        if !request.source_path.is_file() {
            return Err(LakeError::InvalidFileName {
                path: request.source_path.display().to_string(),
            });
        }
        validate_version_numbers(request.major, request.minor, request.patch)?;

        let relation = dataset_table_relation(
            request.study_id,
            request.dataset_name.as_str(),
            request.major,
            request.minor,
            request.patch,
        );
        let stage_table = dataset_stage_table_name();
        let stage_sql = quote_identifier(&stage_table);
        let converted_source_path = (request.format == DatasetFileFormat::ArrowIpc)
            .then(|| temporary_arrow_ipc_parquet_path(&request.source_path));
        let target_sql = relation.to_sql();
        let mut target_created = false;
        let result = (|| {
            let source_path = if let Some(path) = &converted_source_path {
                arrow_ipc_file_to_parquet(&request.source_path, path)?;
                path.as_path()
            } else {
                request.source_path.as_path()
            };
            let source_format = if request.format == DatasetFileFormat::ArrowIpc {
                DatasetFileFormat::Parquet
            } else {
                request.format
            };
            let relation_sql = file_relation_sql(source_path, source_format, &request.options)?;
            connection.execute_batch(&format!("DROP TABLE IF EXISTS {stage_sql};"))?;
            connection
                .execute_batch(&format!("CREATE TEMP TABLE {stage_sql} AS {relation_sql};"))?;
            connection.execute_batch(&format!(
                "CREATE SCHEMA IF NOT EXISTS {};",
                quote_identifier(&relation.schema_name)
            ))?;
            ensure_dataset_target_empty(connection, &relation)?;
            before_create()?;
            connection.execute_batch(&format!(
                "CREATE TABLE {target_sql} AS SELECT * FROM {stage_sql};"
            ))?;
            target_created = true;

            let row_count = table_row_count(connection, &target_sql)?;
            let columns = table_columns(connection, &target_sql)?;
            let column_names: Vec<String> =
                columns.iter().map(|column| column.name.clone()).collect();
            let column_count = column_names.len();
            Ok::<_, LakeError>(LoadedDatasetTable {
                relation,
                row_count,
                column_count,
                column_names,
                columns,
            })
        })();

        finish_dataset_load(
            connection,
            Some(&stage_sql),
            &target_sql,
            target_created,
            converted_source_path.as_deref(),
            result,
        )
    }

    /// Loads a versioned dataset table from a caller-supplied SQL `SELECT`.
    ///
    /// # Errors
    ///
    /// Returns an error if a version component is negative, the source SQL is
    /// empty or not a `SELECT`, or DuckDB cannot execute the staging or create
    /// statements.
    #[cfg(test)]
    fn load_dataset_table_from_sql(
        &self,
        connection: &duckdb::Connection,
        request: &LoadDatasetTableFromSqlRequest,
    ) -> Result<LoadedDatasetTable, LakeError> {
        self.load_dataset_table_from_sql_with_creation_intent(connection, request, || Ok(()))
    }

    /// The output file is produced by the restricted executor, never supplied
    /// as a client path or SQL document. Existing reservation/cleanup semantics
    /// remain the sole authority for publishing its Dataset table.
    pub fn load_restricted_query_output_reserved(
        &self,
        connection: &duckdb::Connection,
        authority: &mut dyn ahri_tre_core::DatasetWriteAuthority,
        target: &LoadDatasetTableFromSqlRequest,
        output: crate::PreparedQueryOutput,
    ) -> Result<LoadedDatasetTable, LakeError> {
        self.load_dataset_table_from_file_reserved(
            connection,
            authority,
            &LoadDatasetTableFromFileRequest {
                study_id: target.study_id,
                dataset_name: target.dataset_name.clone(),
                major: target.major,
                minor: target.minor,
                patch: target.patch,
                source_path: output.path.clone(),
                format: DatasetFileFormat::Parquet,
                options: DatasetFileReadOptions::default(),
            },
        )
    }

    /// Loads only the exact output held by an active shared reservation.
    pub fn load_dataset_table_from_sql_reserved(
        &self,
        connection: &duckdb::Connection,
        authority: &mut dyn ahri_tre_core::DatasetWriteAuthority,
        request: &LoadDatasetTableFromSqlRequest,
    ) -> Result<LoadedDatasetTable, LakeError> {
        authorize_dataset_write(
            authority,
            request.study_id,
            &request.dataset_name,
            request.major,
            request.minor,
            request.patch,
        )?;
        self.load_dataset_table_from_sql_with_creation_intent(connection, request, || {
            authority.record_creation_intent().map_err(Into::into)
        })
    }

    fn load_dataset_table_from_sql_with_creation_intent(
        &self,
        connection: &duckdb::Connection,
        request: &LoadDatasetTableFromSqlRequest,
        mut before_create: impl FnMut() -> Result<(), LakeError>,
    ) -> Result<LoadedDatasetTable, LakeError> {
        validate_version_numbers(request.major, request.minor, request.patch)?;
        let source_sql = trimmed_select_sql(&request.source_sql)?;

        let relation = dataset_table_relation(
            request.study_id,
            request.dataset_name.as_str(),
            request.major,
            request.minor,
            request.patch,
        );
        let stage_table = dataset_stage_table_name();
        let stage_sql = quote_identifier(&stage_table);
        let target_sql = relation.to_sql();

        let mut target_created = false;
        let result = (|| {
            connection.execute_batch(&format!("DROP TABLE IF EXISTS {stage_sql};"))?;
            connection.execute_batch(&format!(
                "CREATE TEMP TABLE {stage_sql} AS SELECT * FROM ({source_sql}) src;"
            ))?;
            connection.execute_batch(&format!(
                "CREATE SCHEMA IF NOT EXISTS {};",
                quote_identifier(&relation.schema_name)
            ))?;
            ensure_dataset_target_empty(connection, &relation)?;
            before_create()?;
            connection.execute_batch(&format!(
                "CREATE TABLE {target_sql} AS SELECT * FROM {stage_sql};"
            ))?;
            target_created = true;

            let row_count = table_row_count(connection, &target_sql)?;
            let columns = table_columns(connection, &target_sql)?;
            let column_names: Vec<String> =
                columns.iter().map(|column| column.name.clone()).collect();
            let column_count = column_names.len();
            Ok::<_, LakeError>(LoadedDatasetTable {
                relation,
                row_count,
                column_count,
                column_names,
                columns,
            })
        })();

        finish_dataset_load(
            connection,
            Some(&stage_sql),
            &target_sql,
            target_created,
            None,
            result,
        )
    }

    /// Pivots a REDCap EAV CSV export into a versioned dataset table.
    ///
    /// # Errors
    ///
    /// Returns an error if the source file is missing, a version component is
    /// negative, the REDCap CSV cannot be inspected, or DuckDB cannot create the
    /// target relation.
    #[cfg(test)]
    fn load_redcap_form_dataset(
        &self,
        connection: &duckdb::Connection,
        request: &LoadRedcapFormDatasetRequest,
    ) -> Result<LoadedDatasetTable, LakeError> {
        self.load_redcap_form_dataset_with_creation_intent(connection, request, || Ok(()))
    }

    /// Loads only the exact output held by an active shared reservation.
    pub fn load_redcap_form_dataset_reserved(
        &self,
        connection: &duckdb::Connection,
        authority: &mut dyn ahri_tre_core::DatasetWriteAuthority,
        request: &LoadRedcapFormDatasetRequest,
    ) -> Result<LoadedDatasetTable, LakeError> {
        authorize_dataset_write(
            authority,
            request.study_id,
            &request.dataset_name,
            request.major,
            request.minor,
            request.patch,
        )?;
        self.load_redcap_form_dataset_with_creation_intent(connection, request, || {
            authority.record_creation_intent().map_err(Into::into)
        })
    }

    fn load_redcap_form_dataset_with_creation_intent(
        &self,
        connection: &duckdb::Connection,
        request: &LoadRedcapFormDatasetRequest,
        mut before_create: impl FnMut() -> Result<(), LakeError>,
    ) -> Result<LoadedDatasetTable, LakeError> {
        if !request.source_eav_path.is_file() {
            return Err(LakeError::InvalidFileName {
                path: request.source_eav_path.display().to_string(),
            });
        }
        validate_version_numbers(request.major, request.minor, request.patch)?;

        let relation = dataset_table_relation(
            request.study_id,
            request.dataset_name.as_str(),
            request.major,
            request.minor,
            request.patch,
        );
        let target_sql = relation.to_sql();
        let eav_sql = sql_literal(&request.source_eav_path.display().to_string());
        let field_list = sql_string_list(request.fields.iter().map(|field| field.name.as_str()));
        ensure_redcap_eav_columns(
            connection,
            &request.source_eav_path,
            &["field_name", "value"],
        )?;
        let has_event = csv_has_column(connection, &request.source_eav_path, "redcap_event_name")?;
        let has_repeat_instrument = csv_has_column(
            connection,
            &request.source_eav_path,
            "redcap_repeat_instrument",
        )?;
        let has_repeat_instance = csv_has_column(
            connection,
            &request.source_eav_path,
            "redcap_repeat_instance",
        )?;

        let mut key_columns = vec!["record".to_string()];
        if has_event {
            key_columns.push("redcap_event_name".to_string());
        }
        if has_repeat_instrument {
            key_columns.push("redcap_repeat_instrument".to_string());
        }
        if has_repeat_instance {
            key_columns.push("redcap_repeat_instance".to_string());
        }
        let key_sql = key_columns
            .iter()
            .map(|column| quote_identifier(column))
            .collect::<Vec<_>>()
            .join(", ");

        let projections = redcap_projection_sql(&request.fields, &request.synthetic_columns);
        let mut target_created = false;
        let result = (|| {
            connection.execute_batch(&format!(
                "CREATE SCHEMA IF NOT EXISTS {};",
                quote_identifier(&relation.schema_name)
            ))?;
            ensure_dataset_target_empty(connection, &relation)?;
            before_create()?;
            connection.execute_batch(&format!(
                r#"
                CREATE TABLE {target_sql} AS
                WITH src AS (
                    SELECT * FROM {}
                    WHERE field_name IN ({field_list})
                ),
                grouped AS (
                    SELECT {key_sql}, field_name, string_agg(CAST(value AS VARCHAR), ', ') AS value
                    FROM src
                    GROUP BY {key_sql}, field_name
                ),
                pivoted AS (
                    PIVOT grouped
                    ON field_name
                    USING any_value(value)
                    GROUP BY {key_sql}
                )
                SELECT {projections}
                FROM pivoted
                ORDER BY {key_sql};
                "#,
                redcap_eav_scan_sql(&eav_sql)
            ))?;
            target_created = true;

            let row_count = table_row_count(connection, &target_sql)?;
            let columns = table_columns(connection, &target_sql)?;
            let column_names: Vec<String> =
                columns.iter().map(|column| column.name.clone()).collect();
            let column_count = column_names.len();
            Ok::<_, LakeError>(LoadedDatasetTable {
                relation,
                row_count,
                column_count,
                column_names,
                columns,
            })
        })();

        finish_dataset_load(connection, None, &target_sql, target_created, None, result)
    }

    /// Compensates only a still-owned, non-admitted output of a known attempt.
    pub fn cleanup_reserved_dataset_output(
        &self,
        connection: &duckdb::Connection,
        authority: &mut dyn ahri_tre_core::DatasetOutputAuthority,
    ) -> Result<(), LakeError> {
        authority.ensure_current()?;
        let version = authority.output_version()?;
        let relation = dataset_table_relation(
            version.output.study_id,
            &version.output.lake_name,
            version.major,
            version.minor,
            version.patch,
        );
        self.drop_dataset_table(connection, &relation)
    }

    /// Drops a dataset table if it exists.
    ///
    /// # Errors
    ///
    /// Returns an error if DuckDB rejects the `DROP TABLE` command.
    pub fn drop_dataset_table(
        &self,
        connection: &duckdb::Connection,
        relation: &DatasetTableRelation,
    ) -> Result<(), LakeError> {
        connection.execute_batch(&format!("DROP TABLE IF EXISTS {};", relation.to_sql()))?;
        Ok(())
    }

    /// Exports a versioned dataset table to CSV, NDJSON, Parquet, or Arrow IPC.
    ///
    /// # Errors
    ///
    /// Returns an error if the relation cannot be read, the destination
    /// directory cannot be created, DuckDB export fails, or the Arrow/Parquet
    /// writer fails.
    pub fn export_dataset_table(
        &self,
        connection: &duckdb::Connection,
        request: &ExportDatasetTableRequest,
    ) -> Result<ExportedDatasetTable, LakeError> {
        validate_version_numbers(request.major, request.minor, request.patch)?;
        let relation = dataset_table_relation(
            request.study_id,
            request.dataset_name.as_str(),
            request.major,
            request.minor,
            request.patch,
        );
        let source_sql = limited_relation_sql(&relation.to_sql(), request.limit);
        let row_count = table_row_count(connection, &source_sql)?;
        let column_names = table_column_names(connection, &source_sql)?;
        let column_count = column_names.len();

        if let Some(parent) = request.destination_path.parent() {
            fs::create_dir_all(parent)?;
        }

        let mut warnings = Vec::new();
        match request.format {
            DatasetExportFormat::Csv => {
                export_relation_with_copy(
                    connection,
                    &source_sql,
                    &request.destination_path,
                    "FORMAT CSV, HEADER TRUE",
                    request.compress,
                )?;
            }
            DatasetExportFormat::Ndjson => {
                export_relation_with_copy(
                    connection,
                    &source_sql,
                    &request.destination_path,
                    "FORMAT JSON, ARRAY FALSE",
                    request.compress,
                )?;
            }
            DatasetExportFormat::Parquet => {
                let exported = export_relation_to_arrow_table(
                    connection,
                    &source_sql,
                    request.arrow_metadata.as_ref(),
                )?;
                warnings.extend(exported.warnings);
                write_parquet_file(&request.destination_path, &exported.table)?;
            }
            DatasetExportFormat::ArrowIpc => {
                let exported = export_relation_to_arrow_table(
                    connection,
                    &source_sql,
                    request.arrow_metadata.as_ref(),
                )?;
                warnings.extend(exported.warnings);
                write_ipc_file_to_writer(
                    BufWriter::new(File::create(&request.destination_path)?),
                    &exported.table,
                    ArrowIpcWriteOptions {
                        compression: Some(ArrowIpcCompression::Zstd),
                    },
                )?;
            }
        }

        Ok(ExportedDatasetTable {
            relation,
            destination_path: request.destination_path.clone(),
            format: request.format,
            compressed: request.compress
                || matches!(
                    request.format,
                    DatasetExportFormat::Parquet | DatasetExportFormat::ArrowIpc
                ),
            row_count,
            column_count,
            exported_size_bytes: fs::metadata(&request.destination_path)?.len(),
            warnings,
        })
    }

    /// Reads a bounded, stringified row preview for a versioned dataset table.
    ///
    /// # Errors
    ///
    /// Returns an error if the relation cannot be inspected or queried.
    pub fn preview_dataset_table(
        &self,
        connection: &duckdb::Connection,
        request: &PreviewDatasetTableRequest,
    ) -> Result<PreviewedDatasetTable, LakeError> {
        validate_version_numbers(request.major, request.minor, request.patch)?;
        let relation = dataset_table_relation(
            request.study_id,
            request.dataset_name.as_str(),
            request.major,
            request.minor,
            request.patch,
        );
        let relation_sql = relation.to_sql();
        let total_row_count = table_row_count(connection, &relation_sql)?;
        let columns = table_columns(connection, &relation_sql)?;
        let rows = preview_rows(connection, &relation_sql, &columns, request.limit)?;
        let displayed_row_count = rows.len();
        Ok(PreviewedDatasetTable {
            relation,
            columns,
            rows,
            displayed_row_count,
            total_row_count,
            truncated: displayed_row_count < total_row_count,
        })
    }
}

fn study_dataset_tables(
    connection: &duckdb::Connection,
    schema_name: &str,
) -> Result<Vec<DatasetTableRelation>, LakeError> {
    let mut statement = connection.prepare(
        "SELECT table_name
         FROM information_schema.tables
         WHERE table_catalog = current_database()
           AND table_schema = ?
           AND table_type = 'BASE TABLE'
         ORDER BY table_name",
    )?;
    let rows = statement.query_map([schema_name], |row| row.get::<_, String>(0))?;
    let mut relations = Vec::new();
    for row in rows {
        relations.push(DatasetTableRelation {
            schema_name: schema_name.to_string(),
            table_name: row?,
        });
    }
    Ok(relations)
}

fn redcap_projection_sql(
    fields: &[RedcapEavField],
    synthetic_columns: &[RedcapSyntheticColumn],
) -> String {
    let mut projections = vec!["TRY_CAST(record AS INTEGER) AS record".to_string()];
    for synthetic in synthetic_columns {
        projections.push(redcap_synthetic_projection_sql(synthetic));
    }
    for field in fields {
        projections.push(redcap_field_projection_sql(field));
    }
    projections.join(", ")
}

fn redcap_synthetic_projection_sql(column: &RedcapSyntheticColumn) -> String {
    let source = quote_identifier(&column.source_column);
    let target = quote_identifier(&column.output_column);
    match column.value_type_id.0 {
        1 => format!("TRY_CAST({source} AS INTEGER) AS {target}"),
        7 => redcap_category_case_sql(&source, &target, &column.choices),
        _ => format!("CAST({source} AS VARCHAR) AS {target}"),
    }
}

fn redcap_field_projection_sql(field: &RedcapEavField) -> String {
    let column = quote_identifier(&field.name);
    match field.value_type_id.0 {
        1 => format!("TRY_CAST({column} AS INTEGER) AS {column}"),
        2 => format!("TRY_CAST({column} AS DOUBLE) AS {column}"),
        4 => redcap_temporal_projection_sql(field, &column, "DATE"),
        5 => redcap_temporal_projection_sql(field, &column, "TIMESTAMP"),
        6 => redcap_temporal_projection_sql(field, &column, "TIME"),
        7 => redcap_category_case_sql(&column, &column, &field.choices),
        8 => format!("CAST({column} AS VARCHAR) AS {column}"),
        _ => format!("CAST({column} AS VARCHAR) AS {column}"),
    }
}

fn redcap_category_case_sql(source: &str, target: &str, choices: &[RedcapEavChoice]) -> String {
    let clauses = choices
        .iter()
        .map(|choice| {
            format!(
                "WHEN CAST({source} AS VARCHAR) = {} THEN {}",
                sql_literal(&choice.redcap_value),
                choice.value
            )
        })
        .collect::<Vec<_>>()
        .join(" ");
    format!("CASE WHEN {source} IS NULL THEN NULL {clauses} ELSE NULL END AS {target}")
}

fn redcap_temporal_projection_sql(
    field: &RedcapEavField,
    column: &str,
    target_type: &str,
) -> String {
    if let Some(format) = field
        .value_format
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
    {
        let parsed = format!("TRY_STRPTIME({column}, {})", sql_literal(format));
        if target_type == "TIMESTAMP" {
            format!("{parsed} AS {column}")
        } else {
            format!("CAST({parsed} AS {target_type}) AS {column}")
        }
    } else {
        format!("TRY_CAST({column} AS {target_type}) AS {column}")
    }
}

fn csv_has_column(
    connection: &duckdb::Connection,
    path: &Path,
    column_name: &str,
) -> Result<bool, LakeError> {
    let columns = redcap_eav_columns(connection, path)?;
    Ok(columns.iter().any(|value| value == column_name))
}

fn ensure_redcap_eav_columns(
    connection: &duckdb::Connection,
    path: &Path,
    required: &[&str],
) -> Result<(), LakeError> {
    let columns = redcap_eav_columns(connection, path)?;
    let missing = required
        .iter()
        .filter(|column| !columns.iter().any(|value| value == **column))
        .copied()
        .collect::<Vec<_>>();
    if !missing.is_empty() {
        return Err(LakeError::InvalidDataFileMetadata {
            field: "redcap_eav_columns",
            message: format!(
                "expected REDCap EAV CSV columns {}, got {}",
                missing.join(", "),
                if columns.is_empty() {
                    "<none>".to_string()
                } else {
                    columns.join(", ")
                }
            ),
        });
    }
    Ok(())
}

fn redcap_eav_columns(
    connection: &duckdb::Connection,
    path: &Path,
) -> Result<Vec<String>, LakeError> {
    let path = sql_literal(&path.display().to_string());
    let mut statement = connection.prepare(&format!(
        "DESCRIBE SELECT * FROM {};",
        redcap_eav_scan_sql(&path)
    ))?;
    let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
    let mut columns = Vec::new();
    for row in rows {
        columns.push(row?);
    }
    Ok(columns)
}

fn redcap_eav_scan_sql(path: &str) -> String {
    format!("read_csv_auto({path}, header=true, delim=',', all_varchar=true)")
}

fn sql_string_list<'a>(values: impl Iterator<Item = &'a str>) -> String {
    values.map(sql_literal).collect::<Vec<_>>().join(", ")
}

/// Canonical reservation key shared by new-name and existing-Asset selectors.
/// Version allocation occurs after this logical Dataset key is reserved.
pub fn dataset_output_key(
    study_id: StudyId,
    dataset_name: &ahri_tre_types::NcName,
) -> ahri_tre_core::DatasetOutputKey {
    ahri_tre_core::DatasetOutputKey {
        study_id,
        lake_name: strict_ncname(dataset_name.as_str()).to_ascii_lowercase(),
    }
}

/// Builds the DuckDB relation name used for a versioned dataset table.
pub fn dataset_table_relation(
    study_id: StudyId,
    dataset_name: &str,
    major: i32,
    minor: i32,
    patch: i32,
) -> DatasetTableRelation {
    DatasetTableRelation {
        schema_name: study_lake_schema(study_id),
        table_name: format!(
            "{}{}",
            strict_ncname(dataset_name),
            version_ncname(major, minor, patch)
        ),
    }
}

fn study_lake_schema(study_id: StudyId) -> String {
    format!(
        "study_{}",
        study_id
            .0
            .to_string()
            .to_ascii_lowercase()
            .replace('-', "_")
    )
}

/// Cleanup includes partial conversion files and staging tables on every exit.
/// Only a target successfully created by this call can be compensated.
fn authorize_dataset_write(
    authority: &mut dyn ahri_tre_core::DatasetWriteAuthority,
    study_id: StudyId,
    dataset_name: &ahri_tre_types::NcName,
    major: i32,
    minor: i32,
    patch: i32,
) -> Result<(), LakeError> {
    let owned = authority.output_version()?;
    if owned.output != dataset_output_key(study_id, dataset_name)
        || (owned.major, owned.minor, owned.patch) != (major, minor, patch)
    {
        return Err(ahri_tre_core::CoreError::Conflict(
            "Dataset reservation does not own this output".into(),
        )
        .into());
    }
    authority.ensure_current().map_err(Into::into)
}

fn ensure_dataset_target_empty(
    connection: &duckdb::Connection,
    relation: &DatasetTableRelation,
) -> Result<(), LakeError> {
    let occupied: bool = connection.query_row("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE lower(table_schema) = lower(?) AND lower(table_name) = lower(?))",
        [&relation.schema_name, &relation.table_name], |row| row.get(0))?;
    if occupied {
        return Err(LakeError::DatasetOutputOccupied);
    }
    Ok(())
}

fn finish_dataset_load(
    connection: &duckdb::Connection,
    stage_sql: Option<&str>,
    target_sql: &str,
    target_created: bool,
    converted_source: Option<&Path>,
    result: Result<LoadedDatasetTable, LakeError>,
) -> Result<LoadedDatasetTable, LakeError> {
    let mut cleanup_failed = false;
    if let Some(stage_sql) = stage_sql {
        cleanup_failed |= connection
            .execute_batch(&format!("DROP TABLE IF EXISTS {stage_sql};"))
            .is_err();
    }
    if let Some(path) = converted_source {
        cleanup_failed |=
            fs::remove_file(path).is_err_and(|error| error.kind() != std::io::ErrorKind::NotFound);
    }
    if target_created && (result.is_err() || cleanup_failed) {
        cleanup_failed |= connection
            .execute_batch(&format!("DROP TABLE IF EXISTS {target_sql};"))
            .is_err();
    }
    if cleanup_failed {
        Err(LakeError::PhysicalRollback {
            operation: "Dataset loading did not complete".into(),
            rollback: "Dataset staging or output cleanup could not be confirmed".into(),
        })
    } else {
        result
    }
}

fn dataset_stage_table_name() -> String {
    format!("__tre_stage_{}", Uuid::new_v4().simple())
}

fn validate_version_numbers(major: i32, minor: i32, patch: i32) -> Result<(), LakeError> {
    if major < 0 || minor < 0 || patch < 0 {
        return Err(LakeError::InvalidMetadataValue {
            key: "dataset_version",
            value: format!("{major}.{minor}.{patch}"),
        });
    }
    Ok(())
}

fn version_ncname(major: i32, minor: i32, patch: i32) -> String {
    strict_ncname(&format!("{major}.{minor}.{patch}"))
}

fn strict_ncname(value: &str) -> String {
    let trimmed = value.trim();
    if trimmed.is_empty() {
        return "_x".to_string();
    }

    let mut output = String::new();
    for (index, character) in trimmed.chars().enumerate() {
        let valid = character.is_ascii_alphanumeric() || character == '_';
        if index == 0 && !(character.is_ascii_alphabetic() || character == '_') {
            output.push('_');
        }
        output.push(if valid { character } else { '_' });
    }

    let mut collapsed = String::new();
    let mut previous_was_replacement = false;
    for character in output.chars() {
        if character == '_' {
            if !previous_was_replacement {
                collapsed.push(character);
            }
            previous_was_replacement = true;
        } else {
            collapsed.push(character);
            previous_was_replacement = false;
        }
    }
    if collapsed.to_ascii_lowercase().starts_with("xml") {
        collapsed.insert(0, '_');
    }
    collapsed
}

pub(crate) fn file_relation_sql(
    source_path: &Path,
    format: DatasetFileFormat,
    options: &DatasetFileReadOptions,
) -> Result<String, LakeError> {
    let canonical_path =
        fs::canonicalize(source_path).unwrap_or_else(|_| source_path.to_path_buf());
    let path_sql = sql_literal(&canonical_path.display().to_string());
    let relation = match format {
        DatasetFileFormat::ArrowIpc => {
            return Err(LakeError::InvalidMetadataValue {
                key: "format",
                value: "Arrow IPC must be converted inside the lake adapter before DuckDB import"
                    .to_string(),
            });
        }
        DatasetFileFormat::Csv => {
            let mut args = vec![
                path_sql,
                format!("header={}", sql_bool(options.header)),
                format!("delim={}", sql_literal(&options.delimiter.to_string())),
            ];
            if !options.null_strings.is_empty() {
                args.push(format!(
                    "nullstr={}",
                    string_list_sql(&options.null_strings)
                ));
            }
            format!("SELECT * FROM read_csv_auto({})", args.join(", "))
        }
        DatasetFileFormat::Json => format!(
            "SELECT * FROM read_json_auto({}, format={})",
            path_sql,
            sql_literal(&options.json_format)
        ),
        DatasetFileFormat::Parquet => format!("SELECT * FROM read_parquet({path_sql})"),
        DatasetFileFormat::Xlsx => {
            let mut args = vec![path_sql];
            if let Some(sheet) = options
                .sheet
                .as_deref()
                .filter(|value| !value.trim().is_empty())
            {
                args.push(format!("sheet={}", sql_literal(sheet)));
            }
            format!("SELECT * FROM read_xlsx({})", args.join(", "))
        }
    };
    Ok(relation)
}

fn arrow_ipc_file_to_parquet(source_path: &Path, parquet_path: &Path) -> Result<(), LakeError> {
    let result = (|| {
        let reader = arrow_ipc::reader::StreamReader::try_new(
            ahri_tre_tabular::open_ipc_messages(source_path)?,
            None,
        )?;
        let mut writer = parquet::arrow::ArrowWriter::try_new(
            File::create(parquet_path)?,
            reader.schema(),
            None,
        )
        .map_err(ahri_tre_tabular::TabularError::from)?;
        for batch in reader {
            let batch = batch?;
            if batch.get_array_memory_size() > ahri_tre_tabular::streaming::MAX_BATCH_BYTES {
                return Err(
                    std::io::Error::other("Arrow batch exceeds upload memory ceiling").into(),
                );
            }
            writer
                .write(&batch)
                .map_err(ahri_tre_tabular::TabularError::from)?;
            writer
                .flush()
                .map_err(ahri_tre_tabular::TabularError::from)?;
        }
        writer
            .close()
            .map_err(ahri_tre_tabular::TabularError::from)?;
        Ok::<_, LakeError>(())
    })();
    if result.is_err() {
        let _ = fs::remove_file(parquet_path);
    }
    result
}

fn temporary_arrow_ipc_parquet_path(source_path: &Path) -> PathBuf {
    source_path.with_extension(format!("arrow-ipc.{}.parquet", Uuid::new_v4().simple()))
}

fn trimmed_select_sql(sql: &str) -> Result<String, LakeError> {
    let trimmed = sql.trim().trim_end_matches(';').trim();
    if trimmed.is_empty()
        || trimmed.contains(';')
        || !trimmed.to_ascii_lowercase().starts_with("select ")
    {
        return Err(LakeError::InvalidMetadataValue {
            key: "source_sql",
            value: "expected a single read-only SELECT statement".to_string(),
        });
    }
    Ok(trimmed.to_string())
}

fn table_row_count(connection: &duckdb::Connection, table_sql: &str) -> Result<usize, LakeError> {
    let count: i64 =
        connection.query_row(&format!("SELECT count(*) FROM {table_sql}"), [], |row| {
            row.get(0)
        })?;
    usize::try_from(count).map_err(|_| LakeError::InvalidMetadataValue {
        key: "row_count",
        value: count.to_string(),
    })
}

fn limited_relation_sql(relation_sql: &str, limit: Option<usize>) -> String {
    match limit {
        Some(limit) => format!("(SELECT * FROM {relation_sql} LIMIT {limit}) AS limited_dataset"),
        None => relation_sql.to_string(),
    }
}

fn table_column_names(
    connection: &duckdb::Connection,
    table_sql: &str,
) -> Result<Vec<String>, LakeError> {
    Ok(table_columns(connection, table_sql)?
        .into_iter()
        .map(|column| column.name)
        .collect())
}

fn table_columns(
    connection: &duckdb::Connection,
    table_sql: &str,
) -> Result<Vec<ImportedColumn>, LakeError> {
    let mut statement = connection.prepare(&format!("DESCRIBE SELECT * FROM {table_sql}"))?;
    let columns = statement.query_map([], |row| {
        Ok(ImportedColumn {
            name: row.get::<_, String>(0)?,
            duckdb_type: row.get::<_, String>(1)?,
        })
    })?;
    let mut imported = Vec::new();
    for column in columns {
        imported.push(column?);
    }
    Ok(imported)
}

fn preview_rows(
    connection: &duckdb::Connection,
    relation_sql: &str,
    columns: &[ImportedColumn],
    limit: usize,
) -> Result<Vec<Vec<Option<String>>>, LakeError> {
    if columns.is_empty() || limit == 0 {
        return Ok(Vec::new());
    }
    let projection = columns
        .iter()
        .map(|column| {
            let name = quote_identifier(&column.name);
            format!("CAST({name} AS VARCHAR)")
        })
        .collect::<Vec<_>>()
        .join(", ");
    let mut statement = connection.prepare(&format!(
        "SELECT {projection} FROM {relation_sql} LIMIT {limit}"
    ))?;
    let rows = statement.query_map([], |row| {
        let mut values = Vec::with_capacity(columns.len());
        for index in 0..columns.len() {
            values.push(row.get::<_, Option<String>>(index)?);
        }
        Ok(values)
    })?;
    let mut preview = Vec::new();
    for row in rows {
        preview.push(row?);
    }
    Ok(preview)
}

fn export_relation_with_copy(
    connection: &duckdb::Connection,
    source_sql: &str,
    destination_path: &Path,
    copy_options: &str,
    compress: bool,
) -> Result<(), LakeError> {
    if compress {
        let temporary_path = temporary_export_path(destination_path);
        let result = (|| {
            copy_relation_to_path(connection, source_sql, &temporary_path, copy_options)?;
            compress_zstd_file(&temporary_path, destination_path)
        })();
        let _ = fs::remove_file(&temporary_path);
        result
    } else {
        copy_relation_to_path(connection, source_sql, destination_path, copy_options)
    }
}

fn copy_relation_to_path(
    connection: &duckdb::Connection,
    source_sql: &str,
    destination_path: &Path,
    copy_options: &str,
) -> Result<(), LakeError> {
    let destination_sql = sql_literal(&destination_path.display().to_string());
    connection.execute_batch(&format!(
        "COPY (SELECT * FROM {source_sql}) TO {destination_sql} ({copy_options});"
    ))?;
    Ok(())
}

fn export_relation_to_arrow_table(
    connection: &duckdb::Connection,
    source_sql: &str,
    metadata: Option<&AhriTreArrowMetadata>,
) -> Result<AnnotatedArrowTable, LakeError> {
    let mut statement = connection.prepare(&format!("SELECT * FROM {source_sql}"))?;
    let mut arrow = statement.query_arrow([])?;
    let schema = arrow.get_schema();
    let table = ArrowTable::try_new(schema, arrow.by_ref().collect())?;
    if let Some(metadata) = metadata {
        Ok(annotate_arrow_table(&table, metadata)?)
    } else {
        Ok(AnnotatedArrowTable {
            table,
            warnings: Vec::new(),
        })
    }
}

fn compress_zstd_file(source_path: &Path, destination_path: &Path) -> Result<(), LakeError> {
    let source = BufReader::new(File::open(source_path)?);
    let mut encoder =
        zstd::stream::write::Encoder::new(BufWriter::new(File::create(destination_path)?), 3)?;
    let mut reader = source;
    io::copy(&mut reader, &mut encoder)?;
    encoder.finish()?.flush()?;
    Ok(())
}

fn temporary_export_path(destination_path: &Path) -> PathBuf {
    let extension = destination_path
        .extension()
        .and_then(|value| value.to_str())
        .filter(|value| !value.trim().is_empty())
        .unwrap_or("tmp");
    destination_path.with_extension(format!("{}.{}.tmp", extension, Uuid::new_v4().simple()))
}

fn quote_identifier(value: &str) -> String {
    format!("\"{}\"", value.replace('"', "\"\""))
}

fn sql_literal(value: &str) -> String {
    format!("'{}'", value.replace('\'', "''"))
}

fn string_list_sql(values: &[String]) -> String {
    format!(
        "[{}]",
        values
            .iter()
            .map(|value| sql_literal(value))
            .collect::<Vec<_>>()
            .join(", ")
    )
}

fn sql_bool(value: bool) -> &'static str {
    if value { "true" } else { "false" }
}

#[cfg(test)]
mod tests {
    use std::fs;
    use std::os::unix::fs::PermissionsExt;
    use std::path::{Path, PathBuf};

    use ahri_tre_types::{NcName, StudyId};
    use uuid::Uuid;

    use super::*;
    use crate::{ScratchAttemptId, TrustedScratch};

    #[test]
    fn rejected_sql_load_preserves_existing_target_and_removes_staging() {
        let connection = duckdb::Connection::open_in_memory().unwrap();
        let lake = DuckLakeAdapter::new("/unused");
        let mut request = LoadDatasetTableFromSqlRequest {
            study_id: StudyId(Uuid::new_v4()),
            dataset_name: NcName::parse("preserved").unwrap(),
            major: 1,
            minor: 0,
            patch: 0,
            source_sql: "SELECT 42 AS answer".into(),
        };
        let loaded = lake
            .load_dataset_table_from_sql(&connection, &request)
            .unwrap();
        request.source_sql = "SELECT * FROM missing_source".into();
        assert!(
            lake.load_dataset_table_from_sql(&connection, &request)
                .is_err()
        );
        let answer: i64 = connection
            .query_row(
                &format!("SELECT answer FROM {}", loaded.relation.to_sql()),
                [],
                |row| row.get(0),
            )
            .expect("rejected staging must preserve existing target");
        assert_eq!(answer, 42);
        request.source_sql = "SELECT 99 AS answer".into();
        assert!(
            lake.load_dataset_table_from_sql(&connection, &request)
                .is_err(),
            "an occupied version must not be overwritten"
        );
        let stages: i64 = connection
            .query_row(
                "SELECT count(*) FROM duckdb_tables() WHERE table_name LIKE '__tre_stage_%'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(stages, 0);
    }

    #[test]
    fn loads_csv_file_into_study_scoped_dataset_table() {
        let root = temp_lake_root();
        let source_path = root.join("ingest.csv");
        fs::write(&source_path, b"category,country\nA,USA\nB,South Africa\n")
            .expect("source fixture should write");
        let adapter = DuckLakeAdapter::new(path_string(&root));
        let connection = duckdb::Connection::open_in_memory().expect("DuckDB should open");
        let study_id = StudyId(Uuid::new_v4());

        let loaded = adapter
            .load_dataset_table_from_file(
                &connection,
                &LoadDatasetTableFromFileRequest {
                    study_id,
                    dataset_name: NcName::parse("ingested_dataset")
                        .expect("dataset name should parse"),
                    major: 1,
                    minor: 0,
                    patch: 0,
                    source_path,
                    format: DatasetFileFormat::Csv,
                    options: DatasetFileReadOptions::default(),
                },
            )
            .expect("dataset table should load");

        assert_eq!(
            loaded.relation.schema_name,
            format!(
                "study_{}",
                study_id
                    .0
                    .to_string()
                    .to_ascii_lowercase()
                    .replace('-', "_")
            )
        );
        assert_eq!(loaded.relation.table_name, "ingested_dataset_1_0_0");
        assert_eq!(loaded.row_count, 2);
        assert_eq!(loaded.column_count, 2);
        assert_eq!(
            loaded.columns,
            vec![
                ImportedColumn {
                    name: "category".to_string(),
                    duckdb_type: "VARCHAR".to_string(),
                },
                ImportedColumn {
                    name: "country".to_string(),
                    duckdb_type: "VARCHAR".to_string(),
                },
            ]
        );
        let count: i64 = connection
            .query_row(
                &format!("SELECT count(*) FROM {}", loaded.relation.to_sql()),
                [],
                |row| row.get(0),
            )
            .expect("loaded dataset should query");
        assert_eq!(count, 2);

        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn validates_csv_with_the_same_duckdb_reader_without_persisting_a_relation() {
        let root = temp_lake_root();
        let valid_path = root.join("valid.csv");
        let invalid_path = root.join("invalid.csv");
        fs::write(&valid_path, b"id,notes\n1,\"first line\nsecond line\"\n")
            .expect("valid source fixture should write");
        fs::write(&invalid_path, b"id,value\n1\n").expect("invalid source fixture should write");
        let adapter = DuckLakeAdapter::new(path_string(&root));
        let connection = duckdb::Connection::open_in_memory().expect("DuckDB should open");

        adapter
            .validate_dataset_file(
                &connection,
                &valid_path,
                DatasetFileFormat::Csv,
                &DatasetFileReadOptions::default(),
            )
            .expect("DuckDB should accept the valid CSV");
        assert!(
            adapter
                .validate_dataset_file(
                    &connection,
                    &invalid_path,
                    DatasetFileFormat::Csv,
                    &DatasetFileReadOptions::default(),
                )
                .is_err()
        );

        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn retaining_an_exact_dataset_set_removes_stale_study_tables() {
        let adapter = DuckLakeAdapter::new("/lake");
        let connection = duckdb::Connection::open_in_memory().expect("DuckDB should open");
        let study_id = StudyId(Uuid::new_v4());
        let retained = dataset_table_relation(study_id, "retained", 1, 0, 0);
        let stale = dataset_table_relation(study_id, "stale", 1, 0, 0);
        connection
            .execute_batch(&format!(
                "CREATE SCHEMA {}; CREATE TABLE {} (id INTEGER); CREATE TABLE {} (id INTEGER);",
                quote_identifier(&retained.schema_name),
                retained.to_sql(),
                stale.to_sql()
            ))
            .expect("study fixtures should be created");

        let removed = adapter
            .retain_only_dataset_tables(&connection, study_id, std::slice::from_ref(&retained))
            .expect("stale study tables should be removed");

        assert_eq!(removed, vec![stale]);
        connection
            .query_row(
                &format!("SELECT count(*) FROM {}", retained.to_sql()),
                [],
                |row| row.get::<_, i64>(0),
            )
            .expect("retained table should remain queryable");
    }

    #[test]
    fn exact_csv_replacement_removes_contamination_and_reloads_expected_tables() {
        let root = temp_lake_root();
        let scratch_root = root.join("scratch");
        fs::create_dir(&scratch_root).expect("scratch root should be created");
        fs::set_permissions(&scratch_root, fs::Permissions::from_mode(0o700))
            .expect("scratch root should be private");
        let scratch = TrustedScratch::open(&scratch_root, [Path::new("/lake")])
            .expect("trusted scratch should open")
            .create_attempt(
                ScratchAttemptId::new("00000000000000000000000000000015")
                    .expect("attempt identifier should be valid"),
            )
            .expect("scratch attempt should be created");
        let adapter = DuckLakeAdapter::new("/lake");
        let connection = duckdb::Connection::open_in_memory().expect("DuckDB should open");
        let study_id = StudyId(Uuid::new_v4());
        let stale = dataset_table_relation(study_id, "stale", 1, 0, 0);
        connection
            .execute_batch(&format!(
                "CREATE SCHEMA {}; CREATE TABLE {} (id INTEGER);",
                quote_identifier(&stale.schema_name),
                stale.to_sql()
            ))
            .expect("contaminating table should be created");
        let prior = dataset_table_relation(study_id, "alpha", 1, 0, 0);
        connection
            .execute_batch(&format!(
                "CREATE TABLE {} (old_value INTEGER); INSERT INTO {} VALUES (99);",
                prior.to_sql(),
                prior.to_sql()
            ))
            .unwrap();
        let seeds = [
            DatasetCsvSeed {
                dataset_name: NcName::parse("alpha").unwrap(),
                contents: b"id,label\n1,first\n",
            },
            DatasetCsvSeed {
                dataset_name: NcName::parse("beta").unwrap(),
                contents: b"id,label\n2,second\n",
            },
        ];

        let loaded = adapter
            .replace_dataset_tables_from_csv(&connection, &scratch, study_id, &seeds)
            .expect("exact seed replacement should succeed");

        assert_eq!(loaded.len(), 2);
        assert_eq!(
            connection
                .query_row(
                    &format!("SELECT label FROM {}", prior.to_sql()),
                    [],
                    |row| row.get::<_, String>(0)
                )
                .unwrap(),
            "first"
        );
        assert_eq!(
            study_dataset_tables(&connection, &study_lake_schema(study_id)).unwrap(),
            vec![
                dataset_table_relation(study_id, "alpha", 1, 0, 0),
                dataset_table_relation(study_id, "beta", 1, 0, 0),
            ]
        );
        drop(scratch);
        let _ = fs::remove_dir_all(root);
    }

    #[test]
    fn exact_csv_replacement_rolls_back_after_a_mid_seed_failure() {
        let root = temp_lake_root();
        let scratch_root = root.join("scratch");
        fs::create_dir(&scratch_root).expect("scratch root should be created");
        fs::set_permissions(&scratch_root, fs::Permissions::from_mode(0o700))
            .expect("scratch root should be private");
        let scratch = TrustedScratch::open(&scratch_root, [Path::new("/lake")])
            .expect("trusted scratch should open")
            .create_attempt(
                ScratchAttemptId::new("00000000000000000000000000000016")
                    .expect("attempt identifier should be valid"),
            )
            .expect("scratch attempt should be created");
        let adapter = DuckLakeAdapter::new("/lake");
        let connection = duckdb::Connection::open_in_memory().expect("DuckDB should open");
        let study_id = StudyId(Uuid::new_v4());
        let stale = dataset_table_relation(study_id, "stale", 1, 0, 0);
        connection
            .execute_batch(&format!(
                "CREATE SCHEMA {}; CREATE TABLE {} (id INTEGER); INSERT INTO {} VALUES (7);",
                quote_identifier(&stale.schema_name),
                stale.to_sql(),
                stale.to_sql()
            ))
            .expect("pre-seed state should be created");
        fs::write(scratch.path().join("beta_1_0_0.csv"), b"occupied")
            .expect("later seed path should be occupied");
        let seeds = [
            DatasetCsvSeed {
                dataset_name: NcName::parse("alpha").unwrap(),
                contents: b"id\n1\n",
            },
            DatasetCsvSeed {
                dataset_name: NcName::parse("beta").unwrap(),
                contents: b"id\n2\n",
            },
        ];

        adapter
            .replace_dataset_tables_from_csv(&connection, &scratch, study_id, &seeds)
            .expect_err("occupied second seed path should fail after the first replacement");

        assert_eq!(
            study_dataset_tables(&connection, &study_lake_schema(study_id)).unwrap(),
            vec![stale.clone()]
        );
        assert_eq!(
            connection
                .query_row(&format!("SELECT id FROM {}", stale.to_sql()), [], |row| {
                    row.get::<_, i64>(0)
                })
                .unwrap(),
            7
        );
        drop(scratch);
        let _ = fs::remove_dir_all(root);
    }

    #[test]
    fn loads_arrow_ipc_file_into_study_scoped_dataset_table() {
        let root = temp_lake_root();
        let source_path = root.join("ingest.arrow");
        let batch = ahri_tre_tabular::columns_to_record_batch(vec![
            ahri_tre_tabular::TabularColumn::Utf8 {
                name: "status".to_string(),
                values: vec![Some("A".to_string()), Some("D".to_string())],
            },
            ahri_tre_tabular::TabularColumn::Int64 {
                name: "age".to_string(),
                values: vec![Some(34), Some(41)],
            },
        ])
        .expect("Arrow IPC fixture batch should build");
        let table = ArrowTable::from_batches(vec![batch]).expect("Arrow IPC table should build");
        let bytes =
            ahri_tre_tabular::write_ipc_file(&table).expect("Arrow IPC fixture should serialize");
        fs::write(&source_path, bytes).expect("Arrow IPC fixture should write");
        let adapter = DuckLakeAdapter::new(path_string(&root));
        let connection = duckdb::Connection::open_in_memory().expect("DuckDB should open");
        let study_id = StudyId(Uuid::new_v4());

        let loaded = adapter
            .load_dataset_table_from_file(
                &connection,
                &LoadDatasetTableFromFileRequest {
                    study_id,
                    dataset_name: NcName::parse("ingested_arrow_dataset")
                        .expect("dataset name should parse"),
                    major: 1,
                    minor: 0,
                    patch: 0,
                    source_path,
                    format: DatasetFileFormat::ArrowIpc,
                    options: DatasetFileReadOptions {
                        null_strings: vec!["".to_string()],
                        ..DatasetFileReadOptions::default()
                    },
                },
            )
            .expect("Arrow IPC dataset table should load");

        assert_eq!(loaded.row_count, 2);
        assert_eq!(loaded.column_count, 2);
        assert_eq!(
            loaded.columns,
            vec![
                ImportedColumn {
                    name: "status".to_string(),
                    duckdb_type: "VARCHAR".to_string(),
                },
                ImportedColumn {
                    name: "age".to_string(),
                    duckdb_type: "BIGINT".to_string(),
                },
            ]
        );
        let count: i64 = connection
            .query_row(
                &format!("SELECT count(*) FROM {}", loaded.relation.to_sql()),
                [],
                |row| row.get(0),
            )
            .expect("loaded Arrow IPC dataset should query");
        assert_eq!(count, 2);
        assert!(
            fs::read_dir(&root)
                .expect("lake root should list")
                .all(|entry| entry
                    .expect("lake entry should read")
                    .path()
                    .extension()
                    .and_then(|extension| extension.to_str())
                    .is_none_or(|extension| extension != "parquet"))
        );

        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn exports_dataset_table_to_supported_local_formats() {
        let root = temp_lake_root();
        let source_path = root.join("ingest.csv");
        fs::write(&source_path, b"category,country\nA,USA\nB,South Africa\n")
            .expect("source fixture should write");
        let adapter = DuckLakeAdapter::new(path_string(&root));
        let connection = duckdb::Connection::open_in_memory().expect("DuckDB should open");
        let study_id = StudyId(Uuid::new_v4());
        let dataset_name = NcName::parse("exportable_dataset").unwrap();
        adapter
            .load_dataset_table_from_file(
                &connection,
                &LoadDatasetTableFromFileRequest {
                    study_id,
                    dataset_name: dataset_name.clone(),
                    major: 1,
                    minor: 0,
                    patch: 0,
                    source_path,
                    format: DatasetFileFormat::Csv,
                    options: DatasetFileReadOptions::default(),
                },
            )
            .expect("dataset table should load");

        let parquet = adapter
            .export_dataset_table(
                &connection,
                &ExportDatasetTableRequest {
                    study_id,
                    dataset_name: dataset_name.clone(),
                    major: 1,
                    minor: 0,
                    patch: 0,
                    limit: None,
                    destination_path: root.join("export.parquet"),
                    format: DatasetExportFormat::Parquet,
                    compress: false,
                    arrow_metadata: None,
                },
            )
            .expect("parquet export should succeed");
        assert_eq!(parquet.row_count, 2);
        assert_eq!(parquet.column_count, 2);
        assert!(parquet.compressed);
        assert!(parquet.exported_size_bytes > 0);

        let ndjson = adapter
            .export_dataset_table(
                &connection,
                &ExportDatasetTableRequest {
                    study_id,
                    dataset_name: dataset_name.clone(),
                    major: 1,
                    minor: 0,
                    patch: 0,
                    limit: None,
                    destination_path: root.join("export.ndjson.zst"),
                    format: DatasetExportFormat::Ndjson,
                    compress: true,
                    arrow_metadata: None,
                },
            )
            .expect("compressed ndjson export should succeed");
        assert_eq!(ndjson.row_count, 2);
        assert!(ndjson.compressed);
        let compressed = fs::read(root.join("export.ndjson.zst")).unwrap();
        let decoded =
            zstd::stream::decode_all(compressed.as_slice()).expect("ndjson should decompress");
        let text = String::from_utf8(decoded).expect("ndjson should be UTF-8");
        assert!(text.contains("\"category\""));
        assert!(text.lines().count() >= 2);

        let arrow = adapter
            .export_dataset_table(
                &connection,
                &ExportDatasetTableRequest {
                    study_id,
                    dataset_name,
                    major: 1,
                    minor: 0,
                    patch: 0,
                    limit: None,
                    destination_path: root.join("export.arrow"),
                    format: DatasetExportFormat::ArrowIpc,
                    compress: false,
                    arrow_metadata: None,
                },
            )
            .expect("compressed Arrow IPC export should succeed");
        assert!(arrow.compressed);
        assert!(
            fs::read(root.join("export.arrow"))
                .expect("Arrow IPC file should read")
                .starts_with(b"ARROW1")
        );

        let limited_csv = adapter
            .export_dataset_table(
                &connection,
                &ExportDatasetTableRequest {
                    study_id,
                    dataset_name: NcName::parse("exportable_dataset").unwrap(),
                    major: 1,
                    minor: 0,
                    patch: 0,
                    limit: Some(1),
                    destination_path: root.join("limited.csv"),
                    format: DatasetExportFormat::Csv,
                    compress: false,
                    arrow_metadata: None,
                },
            )
            .expect("limited CSV export should succeed");
        assert_eq!(limited_csv.row_count, 1);
        assert_eq!(
            fs::read_to_string(root.join("limited.csv")).expect("limited CSV should read"),
            "category,country\nA,USA\n"
        );

        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn previews_dataset_table_with_limit_and_nulls_without_exporting() {
        let root = temp_lake_root();
        let source_path = root.join("preview.csv");
        fs::write(
            &source_path,
            b"id,label,note\n1,alpha,short\n2,,a very long note\n3,gamma,\n",
        )
        .expect("source fixture should write");
        let adapter = DuckLakeAdapter::new(path_string(&root));
        let connection = duckdb::Connection::open_in_memory().expect("DuckDB should open");
        let study_id = StudyId(Uuid::new_v4());
        let dataset_name = NcName::parse("preview_dataset").unwrap();
        adapter
            .load_dataset_table_from_file(
                &connection,
                &LoadDatasetTableFromFileRequest {
                    study_id,
                    dataset_name: dataset_name.clone(),
                    major: 1,
                    minor: 0,
                    patch: 0,
                    source_path,
                    format: DatasetFileFormat::Csv,
                    options: DatasetFileReadOptions::default(),
                },
            )
            .expect("dataset table should load");

        let preview = adapter
            .preview_dataset_table(
                &connection,
                &PreviewDatasetTableRequest {
                    study_id,
                    dataset_name,
                    major: 1,
                    minor: 0,
                    patch: 0,
                    limit: 2,
                },
            )
            .expect("dataset preview should read");

        assert_eq!(
            preview
                .columns
                .iter()
                .map(|column| column.name.as_str())
                .collect::<Vec<_>>(),
            vec!["id", "label", "note"]
        );
        assert_eq!(preview.displayed_row_count, 2);
        assert_eq!(preview.total_row_count, 3);
        assert!(preview.truncated);
        assert_eq!(
            preview.rows,
            vec![
                vec![
                    Some("1".to_string()),
                    Some("alpha".to_string()),
                    Some("short".to_string())
                ],
                vec![
                    Some("2".to_string()),
                    None,
                    Some("a very long note".to_string())
                ],
            ]
        );
        assert!(!root.join("preview_dataset.csv").exists());

        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn loads_redcap_eav_with_event_and_repeat_metadata() {
        let root = temp_lake_root();
        let source_path = root.join("redcap_eav.csv");
        fs::write(
            &source_path,
            b"record,redcap_event_name,redcap_repeat_instrument,redcap_repeat_instance,field_name,value\n\
1,baseline,visit,1,age,40\n\
1,baseline,visit,1,status,1\n\
1,followup,visit,2,age,41\n\
1,followup,visit,2,status,2\n",
        )
        .expect("source fixture should write");
        let adapter = DuckLakeAdapter::new(path_string(&root));
        let connection = duckdb::Connection::open_in_memory().expect("DuckDB should open");
        let study_id = StudyId(Uuid::new_v4());

        let loaded = adapter
            .load_redcap_form_dataset(
                &connection,
                &LoadRedcapFormDatasetRequest {
                    study_id,
                    dataset_name: NcName::parse("redcap_1_visit").unwrap(),
                    major: 1,
                    minor: 0,
                    patch: 0,
                    source_eav_path: source_path,
                    fields: vec![
                        RedcapEavField {
                            name: "age".to_string(),
                            value_type_id: ahri_tre_types::ValueTypeId(1),
                            value_format: None,
                            choices: Vec::new(),
                        },
                        RedcapEavField {
                            name: "status".to_string(),
                            value_type_id: ahri_tre_types::ValueTypeId(7),
                            value_format: None,
                            choices: vec![
                                RedcapEavChoice {
                                    redcap_value: "1".to_string(),
                                    value: 1,
                                },
                                RedcapEavChoice {
                                    redcap_value: "2".to_string(),
                                    value: 2,
                                },
                            ],
                        },
                    ],
                    synthetic_columns: vec![
                        RedcapSyntheticColumn {
                            source_column: "redcap_event_name".to_string(),
                            output_column: "redcap_1_visit_redcap_event".to_string(),
                            value_type_id: ahri_tre_types::ValueTypeId(7),
                            choices: vec![
                                RedcapEavChoice {
                                    redcap_value: "baseline".to_string(),
                                    value: 1,
                                },
                                RedcapEavChoice {
                                    redcap_value: "followup".to_string(),
                                    value: 2,
                                },
                            ],
                        },
                        RedcapSyntheticColumn {
                            source_column: "redcap_repeat_instance".to_string(),
                            output_column: "redcap_1_visit_redcap_repeat_instance".to_string(),
                            value_type_id: ahri_tre_types::ValueTypeId(1),
                            choices: Vec::new(),
                        },
                    ],
                },
            )
            .expect("REDCap EAV dataset should load");

        assert_eq!(loaded.row_count, 2);
        assert!(
            loaded
                .column_names
                .contains(&"redcap_1_visit_redcap_event".to_string())
        );
        let sum_age: i64 = connection
            .query_row(
                &format!("SELECT sum(age) FROM {}", loaded.relation.to_sql()),
                [],
                |row| row.get(0),
            )
            .expect("loaded REDCap dataset should query");
        assert_eq!(sum_age, 81);
        let repeat_type: String = connection
            .query_row(
                &format!(
                    "SELECT typeof(redcap_1_visit_redcap_repeat_instance) FROM {} LIMIT 1",
                    loaded.relation.to_sql()
                ),
                [],
                |row| row.get(0),
            )
            .expect("repeat instance type should be readable");
        assert_eq!(repeat_type, "INTEGER");

        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn redcap_eav_loader_reports_missing_required_eav_columns() {
        let root = temp_lake_root();
        let source_path = root.join("redcap_eav_error.csv");
        fs::write(&source_path, b"{\"error\":\"export too large\"}\n")
            .expect("source fixture should write");
        let adapter = DuckLakeAdapter::new(path_string(&root));
        let connection = duckdb::Connection::open_in_memory().expect("DuckDB should open");
        let study_id = StudyId(Uuid::new_v4());

        let error = adapter
            .load_redcap_form_dataset(
                &connection,
                &LoadRedcapFormDatasetRequest {
                    study_id,
                    dataset_name: NcName::parse("redcap_1_visit").unwrap(),
                    major: 1,
                    minor: 0,
                    patch: 0,
                    source_eav_path: source_path,
                    fields: vec![RedcapEavField {
                        name: "age".to_string(),
                        value_type_id: ahri_tre_types::ValueTypeId(1),
                        value_format: None,
                        choices: Vec::new(),
                    }],
                    synthetic_columns: Vec::new(),
                },
            )
            .expect_err("malformed REDCap EAV export should fail before binding field_name");

        assert!(
            error
                .to_string()
                .contains("expected REDCap EAV CSV columns field_name, value")
        );

        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn loads_redcap_eav_with_event_column_after_value() {
        let root = temp_lake_root();
        let source_path = root.join("redcap_eav_event_after_value.csv");
        fs::write(
            &source_path,
            b"record,field_name,value,redcap_event_name\n\
2,ls_clinic_attended,1,initial_visit_arm_1\n\
2,ls_services_requested,3,initial_visit_arm_1\n\
2,attendance_and_services_complete,2,initial_visit_arm_1\n\
2,ls_clinic_attended,1,follow_up_visit__1_arm_1\n\
2,ls_services_requested,1,follow_up_visit__1_arm_1\n\
2,ls_services_requested,3,follow_up_visit__1_arm_1\n",
        )
        .expect("source fixture should write");
        let adapter = DuckLakeAdapter::new(path_string(&root));
        let connection = duckdb::Connection::open_in_memory().expect("DuckDB should open");
        let study_id = StudyId(Uuid::new_v4());

        let loaded = adapter
            .load_redcap_form_dataset(
                &connection,
                &LoadRedcapFormDatasetRequest {
                    study_id,
                    dataset_name: NcName::parse("redcap_797_attendance_and_services").unwrap(),
                    major: 1,
                    minor: 0,
                    patch: 0,
                    source_eav_path: source_path,
                    fields: vec![
                        RedcapEavField {
                            name: "ls_clinic_attended".to_string(),
                            value_type_id: ahri_tre_types::ValueTypeId(7),
                            value_format: None,
                            choices: vec![RedcapEavChoice {
                                redcap_value: "1".to_string(),
                                value: 1,
                            }],
                        },
                        RedcapEavField {
                            name: "ls_services_requested".to_string(),
                            value_type_id: ahri_tre_types::ValueTypeId(8),
                            value_format: None,
                            choices: Vec::new(),
                        },
                        RedcapEavField {
                            name: "attendance_and_services_complete".to_string(),
                            value_type_id: ahri_tre_types::ValueTypeId(7),
                            value_format: None,
                            choices: vec![RedcapEavChoice {
                                redcap_value: "2".to_string(),
                                value: 2,
                            }],
                        },
                    ],
                    synthetic_columns: vec![RedcapSyntheticColumn {
                        source_column: "redcap_event_name".to_string(),
                        output_column: "redcap_797_attendance_and_services_redcap_event"
                            .to_string(),
                        value_type_id: ahri_tre_types::ValueTypeId(7),
                        choices: vec![
                            RedcapEavChoice {
                                redcap_value: "initial_visit_arm_1".to_string(),
                                value: 1,
                            },
                            RedcapEavChoice {
                                redcap_value: "follow_up_visit__1_arm_1".to_string(),
                                value: 2,
                            },
                        ],
                    }],
                },
            )
            .expect("REDCap EAV dataset should load when event appears after value");

        assert_eq!(loaded.row_count, 2);
        assert!(
            loaded
                .column_names
                .contains(&"redcap_797_attendance_and_services_redcap_event".to_string())
        );
        let services: String = connection
            .query_row(
                &format!(
                    "SELECT ls_services_requested FROM {} WHERE redcap_797_attendance_and_services_redcap_event = 2",
                    loaded.relation.to_sql()
                ),
                [],
                |row| row.get(0),
            )
            .expect("loaded REDCap follow-up row should query");
        assert_eq!(services, "1, 3");

        let _ = fs::remove_dir_all(&root);
    }

    fn temp_lake_root() -> PathBuf {
        let root = std::env::temp_dir().join(format!(
            "ahri_tre_lake_dataset_{}_{}",
            std::process::id(),
            Uuid::new_v4().simple()
        ));
        fs::create_dir_all(&root).expect("temp lake root should be created");
        root
    }

    fn path_string(path: &Path) -> String {
        path.display().to_string()
    }
}