added --ddl-only feature

Author: hexqnt

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "pgcopy"
-version = "0.1.2"
+version = "0.1.3"
 edition = "2024"
 description = "CLI tool to export selected PostgreSQL objects into a bundle and import them into another database."
 authors = ["hexqnt <kvover@gmail.org>"]

Cargo.toml

@@ -25,7 +25,7 @@
 
 ```bash
 pgcopy export --config <path/to/config.toml> --out <path/to/bundle> [--concurrency N] [--quiet] [--no-progress]
-pgcopy import --in <path/to/bundle> [--mode replace|append] [--concurrency N] [--quiet] [--no-progress]
+pgcopy import --in <path/to/bundle> [--mode replace|append] [--concurrency N] [--ddl-only] [--quiet] [--no-progress]
 pgcopy info --in <path/to/bundle> [--format text|json] [--objects] [--quiet]
 ```
 
@@ -35,6 +35,7 @@ pgcopy info --in <path/to/bundle> [--format text|json] [--objects] [--quiet]
   Операция выполняется атомарно на уровне объекта (`BEGIN/COMMIT`): при ошибке во время `replace`
   изменения по этому объекту откатываются (`ROLLBACK`).
 - `append`: если таблица есть, проверяет совместимость и дозаписывает данные.
+- `--ddl-only`: выполняет только DDL-часть импорта (создание/подготовка таблиц), без загрузки данных.
 
 Поддерживается также алиас `--concurency` (без второй `r`) для `--concurrency`.
 

README.md

@@ -85,8 +85,20 @@ pub async fn run(
     };
 
     progress.set_bundle_running(out_path);
-    let write_result =
-        bundle_io::write_bundle(scratch.path(), out_path, &manifest, password.as_deref());
+    let bundle_scratch_path = scratch.path().to_path_buf();
+    let bundle_out_path = out_path.to_path_buf();
+    let bundle_password = password;
+    let bundle_manifest = manifest;
+    let write_result = tokio::task::spawn_blocking(move || {
+        bundle_io::write_bundle(
+            &bundle_scratch_path,
+            &bundle_out_path,
+            &bundle_manifest,
+            bundle_password.as_deref(),
+        )
+    })
+    .await
+    .context("bundle writer task failed")?;
     match write_result {
         Ok(()) => {
             progress.finish_bundle_done(out_path);

src/export.rs

@@ -30,6 +30,7 @@ pub async fn run(
     bundle_path: &Path,
     mode: ImportMode,
     concurrency: usize,
+    ddl_only: bool,
     bundle_password: Option<&str>,
     target_config: tokio_postgres::Config,
     progress_enabled: bool,
@@ -40,15 +41,21 @@ pub async fn run(
 
     let access = bundle_io::resolve_access(bundle_path, bundle_password)?;
     let client = pg::connect(&target_config).await?;
-    let target_version_num = pg::server_version_num(&client).await?;
+    let target_version_num = if ddl_only {
+        0
+    } else {
+        pg::server_version_num(&client).await?
+    };
 
-    if concurrency == 1 {
+    // DDL-only всегда выполняем потоково: это избегает дорогой распаковки data/* на диск.
+    if ddl_only || concurrency == 1 {
         stream::import_objects_streaming(
             bundle_path,
             access.password.as_deref(),
             access.is_encrypted,
             &client,
             mode,
+            ddl_only,
             target_version_num,
             progress_enabled,
         )
@@ -57,12 +64,20 @@ pub async fn run(
     }
 
     let scratch = tempfile::tempdir().context("failed to create temporary directory for import")?;
-    bundle_io::unpack_bundle(
-        bundle_path,
-        scratch.path(),
-        access.password.as_deref(),
-        access.is_encrypted,
-    )?;
+    let unpack_bundle_path = bundle_path.to_path_buf();
+    let unpack_scratch_path = scratch.path().to_path_buf();
+    let unpack_password = access.password;
+    let unpack_is_encrypted = access.is_encrypted;
+    tokio::task::spawn_blocking(move || {
+        bundle_io::unpack_bundle(
+            &unpack_bundle_path,
+            &unpack_scratch_path,
+            unpack_password.as_deref(),
+            unpack_is_encrypted,
+        )
+    })
+    .await
+    .context("parallel import bundle unpack task failed")??;
 
     let manifest = bundle_io::read_manifest_from_dir(scratch.path())?;
     compat::validate_data_compatibility(&manifest, target_version_num)?;
@@ -72,6 +87,7 @@ pub async fn run(
         &manifest,
         mode,
         concurrency,
+        ddl_only,
         progress_enabled,
     )
     .await?;

src/importer.rs

@@ -84,3 +84,62 @@ pub(super) async fn validate_existing_table_compatibility(
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::validate_data_compatibility;
+    use crate::manifest::{Manifest, ManifestObject};
+    use crate::types::DataFormat;
+
+    fn manifest_for_compat(data_format: DataFormat, source_pg_version_num: i32) -> Manifest {
+        Manifest {
+            format_version: 1,
+            created_at: "2026-02-19T10:00:00Z".to_owned(),
+            source_fingerprint: Some("database=app user=app".to_owned()),
+            source_pg_version_num,
+            data_format,
+            consistent_snapshot: true,
+            objects: vec![ManifestObject {
+                kind: "table".to_owned(),
+                source_schema: "public".to_owned(),
+                source_name: "orders".to_owned(),
+                target_schema: "archive".to_owned(),
+                target_name: "orders".to_owned(),
+                source_select: "select * from public.orders".to_owned(),
+                normalized_select: "SELECT \"id\" FROM \"public\".\"orders\"".to_owned(),
+                ddl_path: "ddl/0001__public.orders.sql".to_owned(),
+                data_path: "data/0001__public.orders.copybin".to_owned(),
+                effective_columns: vec!["id".to_owned()],
+                effective_column_types: vec!["bigint".to_owned()],
+                column_projection: "*".to_owned(),
+                row_estimate: Some(10),
+            }],
+        }
+    }
+
+    #[test]
+    fn binary_compatibility_passes_for_same_major() {
+        let manifest = manifest_for_compat(DataFormat::Binary, 150002);
+        validate_data_compatibility(&manifest, 150099)
+            .expect("binary format should be compatible on same major version");
+    }
+
+    #[test]
+    fn binary_compatibility_fails_for_different_major() {
+        let manifest = manifest_for_compat(DataFormat::Binary, 140012);
+        let error = validate_data_compatibility(&manifest, 150001)
+            .expect_err("binary format must fail across major versions");
+        assert!(
+            error
+                .to_string()
+                .contains("binary COPY compatibility check failed")
+        );
+    }
+
+    #[test]
+    fn csv_compatibility_ignores_major_version_difference() {
+        let manifest = manifest_for_compat(DataFormat::Csv, 140012);
+        validate_data_compatibility(&manifest, 160003)
+            .expect("csv format should ignore major version difference");
+    }
+}

src/importer/compat.rs

@@ -75,8 +75,7 @@ pub async fn copy_data_in_reader<R: Read>(
 
     let mut buffer = vec![0_u8; 64 * 1024];
     loop {
-        let read = reader
-            .read(&mut buffer)
+        let read = tokio::task::block_in_place(|| reader.read(&mut buffer))
             .context("failed to read streaming data from bundle")?;
 
         if read == 0 {

src/importer/copy_stream.rs

@@ -49,3 +49,26 @@ where
         }
     }
 }
+
+/// Выполняет только DDL-подготовку целевого объекта без загрузки данных.
+pub(super) async fn prepare_object_ddl_only(
+    client: &tokio_postgres::Client,
+    object: &ManifestObject,
+    mode: ImportMode,
+    ddl_sql: &str,
+) -> Result<()> {
+    match mode {
+        ImportMode::Replace => {
+            replace_tx::run_replace_atomically(client, object, async {
+                target_table::prepare_target_table(client, object, ImportMode::Replace, ddl_sql)
+                    .await?;
+                Ok(())
+            })
+            .await
+        }
+        ImportMode::Append => {
+            target_table::prepare_target_table(client, object, ImportMode::Append, ddl_sql).await?;
+            Ok(())
+        }
+    }
+}

src/importer/load.rs

@@ -28,6 +28,7 @@ pub async fn import_objects_parallel(
     manifest: &Manifest,
     mode: ImportMode,
     concurrency: usize,
+    ddl_only: bool,
     progress_enabled: bool,
 ) -> Result<u64> {
     let data_format = manifest.data_format;
@@ -46,7 +47,15 @@ pub async fn import_objects_parallel(
         let scratch_dir = scratch_dir.to_path_buf();
         // Каждый worker использует отдельное соединение с target БД.
         workers.spawn(async move {
-            import_worker(&target_config, &scratch_dir, tasks, mode, data_format).await
+            import_worker(
+                &target_config,
+                &scratch_dir,
+                tasks,
+                mode,
+                data_format,
+                ddl_only,
+            )
+            .await
         });
     }
 
@@ -84,6 +93,7 @@ async fn import_worker(
     tasks: Vec<(usize, ManifestObject)>,
     mode: ImportMode,
     data_format: DataFormat,
+    ddl_only: bool,
 ) -> ImportWorkerOutcome {
     let Some((first_index, first_object)) = tasks.first().cloned() else {
         return ImportWorkerOutcome {
@@ -112,7 +122,7 @@ async fn import_worker(
     };
 
     for (index, object) in tasks {
-        let imported = import_object(&client, scratch_dir, &object, mode, data_format)
+        let imported = import_object(&client, scratch_dir, &object, mode, data_format, ddl_only)
             .await
             .with_context(|| {
                 format!(
@@ -149,8 +159,14 @@ async fn import_object(
     object: &ManifestObject,
     mode: ImportMode,
     data_format: DataFormat,
+    ddl_only: bool,
 ) -> Result<u64> {
     let ddl_sql = read_ddl_from_bundle(scratch_dir, &object.ddl_path).await?;
+    if ddl_only {
+        load::prepare_object_ddl_only(client, object, mode, &ddl_sql).await?;
+        return Ok(0);
+    }
+
     let data_path = scratch_dir.join(&object.data_path);
     let inserted_rows = load::load_object(client, object, mode, &ddl_sql, || async {
         copy_stream::copy_data_in_file(