fix: Use unique ETL target prefixes (#53)

* fix: Use unique ETL target prefixes

* fix

Author: peasee

.github/workflows/run_spicebench.yml

@@ -27,10 +27,10 @@ on:
         required: false
         default: 'data-gen/tpch_sf1'
         type: string
-      etl_target_prefix:
-        description: 'S3 key prefix for ETL target (rehydrated) data'
+      etl_target_base_prefix:
+        description: 'Base S3 key prefix for ETL target (rehydrated) data. A random suffix is appended per run.'
         required: false
-        default: 'rehydrated/tpch_sf1'
+        default: 'rehydrated'
         type: string
       etl_region:
         description: 'AWS region for the ETL S3 bucket'
@@ -197,7 +197,7 @@ jobs:
           SYSTEM_ADAPTER: ${{ github.event.inputs.system_adapter || 'spidapter' }}
           ETL_BUCKET: ${{ github.event.inputs.etl_bucket }}
           ETL_SOURCE_PREFIX: ${{ github.event.inputs.etl_source_prefix }}
-          ETL_TARGET_PREFIX: ${{ github.event.inputs.etl_target_prefix }}
+          ETL_TARGET_BASE_PREFIX: ${{ github.event.inputs.etl_target_base_prefix }}
           ETL_REGION: ${{ github.event.inputs.etl_region || 'us-east-1' }}
           ETL_ENDPOINT: ${{ github.event.inputs.etl_endpoint }}
           ETL_NUM_STEPS: ${{ github.event.inputs.etl_num_steps || '25' }}
@@ -209,8 +209,8 @@ jobs:
           if [ -n "${ETL_SOURCE_PREFIX}" ]; then
             ETL_ARGS="${ETL_ARGS} --etl-source-prefix ${ETL_SOURCE_PREFIX}"
           fi
-          if [ -n "${ETL_TARGET_PREFIX}" ]; then
-            ETL_ARGS="${ETL_ARGS} --etl-target-prefix ${ETL_TARGET_PREFIX}"
+          if [ -n "${ETL_TARGET_BASE_PREFIX}" ]; then
+            ETL_ARGS="${ETL_ARGS} --etl-target-base-prefix ${ETL_TARGET_BASE_PREFIX}"
           fi
           if [ -n "${ETL_REGION}" ]; then
             ETL_ARGS="${ETL_ARGS} --etl-region ${ETL_REGION}"

Cargo.lock

@@ -25,3 +25,4 @@ tokio.workspace = true
 tokio-util.workspace = true
 tracing.workspace = true
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
+uuid = { workspace = true, features = ["v4"] }

crates/etl/Cargo.toml

@@ -46,9 +46,10 @@ struct Cli {
     #[arg(long, default_value = "")]
     source_prefix: String,
 
-    /// S3 key prefix for target (rehydrated) data
+    /// Base S3 key prefix for target (rehydrated) data.
+    /// A random suffix is appended automatically to create a unique destination per run.
     #[arg(long, default_value = "")]
-    target_prefix: String,
+    target_base_prefix: String,
 
     /// AWS region
     #[arg(long)]
@@ -88,9 +89,15 @@ impl Cli {
     }
 
     fn target_config(&self) -> TargetConfig {
+        let run_suffix = uuid::Uuid::new_v4().to_string();
+        let prefix = if self.target_base_prefix.is_empty() {
+            run_suffix
+        } else {
+            format!("{}/{run_suffix}", self.target_base_prefix)
+        };
         TargetConfig {
             bucket: self.bucket.clone(),
-            prefix: self.target_prefix.clone(),
+            prefix,
             region: self.region.clone(),
             endpoint: self.endpoint.clone(),
         }
@@ -117,7 +124,7 @@ async fn main() -> anyhow::Result<()> {
         dataset = %cli.dataset,
         bucket = %cli.bucket,
         source_prefix = %cli.source_prefix,
-        target_prefix = %cli.target_prefix,
+        target_base_prefix = %cli.target_base_prefix,
         scale_factor = cli.scale_factor,
         num_steps = cli.num_steps,
         "Starting ETL pipeline"

crates/etl/src/main.rs

@@ -85,9 +85,10 @@ pub struct CommonArgs {
     #[arg(long, default_value = "")]
     pub(crate) etl_source_prefix: String,
 
-    /// S3 key prefix for the ETL target (rehydrated) data
+    /// Base S3 key prefix for the ETL target (rehydrated) data.
+    /// A random suffix is appended automatically to create a unique destination per run.
     #[arg(long, default_value = "")]
-    pub(crate) etl_target_prefix: String,
+    pub(crate) etl_target_base_prefix: String,
 
     /// AWS region for the ETL S3 bucket
     #[arg(long)]

src/args/mod.rs

@@ -82,9 +82,17 @@ async fn main() -> anyhow::Result<()> {
         endpoint: cli.common.etl_endpoint.clone(),
     };
 
+    let run_suffix = Uuid::new_v4().to_string();
+    let target_prefix = if cli.common.etl_target_base_prefix.is_empty() {
+        run_suffix.clone()
+    } else {
+        format!("{}/{run_suffix}", cli.common.etl_target_base_prefix)
+    };
+    tracing::info!(target_prefix = %target_prefix, "Generated unique ETL target prefix");
+
     let target_config = TargetConfig {
         bucket: cli.common.etl_bucket.clone(),
-        prefix: cli.common.etl_target_prefix.clone(),
+        prefix: target_prefix,
         region: cli.common.etl_region.clone(),
         endpoint: cli.common.etl_endpoint.clone(),
     };