fix(glue): real TableVersion delete/lookup, parallel GetPartitions, observable ResumeWorkflowRun + job field persistence (bug-hunt) (#2314)

Author: vieiralucas

crates/fakecloud-athena/src/service/mod.rs

@@ -1199,6 +1199,7 @@ mod tests {
             table_type: Some("EXTERNAL_TABLE".to_string()),
             parameters: std::collections::BTreeMap::new(),
             partitions: std::collections::BTreeMap::new(),
+            catalog_id: "123456789012".to_string(),
         }
     }
 

crates/fakecloud-cloudformation/src/resource_provisioner/glue.rs

@@ -97,6 +97,7 @@ impl ResourceProvisioner {
             fakecloud_glue::Table {
                 name: name.clone(),
                 database_name: db_name.clone(),
+                catalog_id: self.account_id.clone(),
                 description: input
                     .get("Description")
                     .and_then(|v| v.as_str())

crates/fakecloud-glue/src/catalog.rs

@@ -231,9 +231,12 @@ impl GlueService {
                     .collect()
             })
             .unwrap_or_default();
-        Ok(AwsResponse::ok_json(
-            json!({ "UserDefinedFunctions": list }),
-        ))
+        let (page, token) = crate::common::paginate_body(&body, list);
+        let mut resp = json!({ "UserDefinedFunctions": page });
+        if let Some(t) = token {
+            resp["NextToken"] = json!(t);
+        }
+        Ok(AwsResponse::ok_json(resp))
     }
 
     pub(crate) fn update_user_defined_function(

crates/fakecloud-glue/src/common.rs

@@ -93,6 +93,25 @@ pub(crate) fn resource_arn(account: &str, region: &str, kind: &str, name: &str)
     format!("arn:aws:glue:{region}:{account}:{kind}/{name}")
 }
 
+/// Paginate a Glue list op using the request's `MaxResults`/`NextToken`.
+///
+/// Uses the offset-token scheme in [`fakecloud_core::pagination::paginate`]:
+/// the token is the numeric offset of the next page. When `MaxResults` is
+/// absent the full remaining list (from any incoming offset) is returned in a
+/// single page with no continuation token — matching how AWS returns an
+/// unbounded page when the caller omits a page size.
+pub(crate) fn paginate_body(body: &Value, items: Vec<Value>) -> (Vec<Value>, Option<String>) {
+    let token = body.get("NextToken").and_then(|v| v.as_str());
+    let max = body
+        .get("MaxResults")
+        .and_then(|v| v.as_u64())
+        .map(|n| n as usize);
+    match max {
+        Some(m) => fakecloud_core::pagination::paginate(&items, token, m),
+        None => fakecloud_core::pagination::paginate(&items, token, usize::MAX),
+    }
+}
+
 /// A small UUID-ish identifier (32 hex chars). Glue uses these for run ids,
 /// transform ids, etc.
 pub(crate) fn new_id() -> String {

crates/fakecloud-glue/src/connections.rs

@@ -58,11 +58,27 @@ impl GlueService {
     pub(crate) fn get_connection(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
         let body = req.json_body();
         let name = req_str(&body, "Name")?;
+        let hide_password = body
+            .get("HidePassword")
+            .and_then(|v| v.as_bool())
+            .unwrap_or(false);
         let accounts = self.state.read();
-        let c = accounts
+        let mut c = accounts
             .get(&req.account_id)
             .and_then(|s| s.connections.get(name))
+            .cloned()
             .ok_or_else(|| entity_not_found(format!("Connection {name} not found")))?;
+        // With HidePassword=true AWS strips the PASSWORD from ConnectionProperties
+        // (and any encrypted password) rather than returning it verbatim.
+        if hide_password {
+            if let Some(props) = c
+                .get_mut("ConnectionProperties")
+                .and_then(|v| v.as_object_mut())
+            {
+                props.remove("PASSWORD");
+                props.remove("ENCRYPTED_PASSWORD");
+            }
+        }
         Ok(AwsResponse::ok_json(json!({ "Connection": c })))
     }
 

crates/fakecloud-glue/src/crawlers.rs

@@ -82,12 +82,18 @@ impl GlueService {
     }
 
     pub(crate) fn get_crawlers(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
+        let body = req.json_body();
         let accounts = self.state.read();
         let crawlers: Vec<Value> = accounts
             .get(&req.account_id)
             .map(|s| s.crawlers.values().cloned().collect())
             .unwrap_or_default();
-        Ok(AwsResponse::ok_json(json!({ "Crawlers": crawlers })))
+        let (page, token) = crate::common::paginate_body(&body, crawlers);
+        let mut resp = json!({ "Crawlers": page });
+        if let Some(t) = token {
+            resp["NextToken"] = json!(t);
+        }
+        Ok(AwsResponse::ok_json(resp))
     }
 
     pub(crate) fn batch_get_crawlers(

crates/fakecloud-glue/src/jobs.rs

@@ -68,6 +68,11 @@ pub(crate) fn job_to_json(j: &Job) -> Value {
         "SecurityConfiguration": j.security_configuration,
         "NotificationProperty": j.notification_property,
         "NonOverridableArguments": j.non_overridable_arguments,
+        "SourceControlDetails": j.source_control_details,
+        "CodeGenConfigurationNodes": j.code_gen_configuration_nodes,
+        "MaintenanceWindow": j.maintenance_window,
+        "LogUri": j.log_uri,
+        "AllocatedCapacity": j.allocated_capacity,
     })
 }
 
@@ -138,6 +143,11 @@ impl GlueService {
             security_configuration: body["SecurityConfiguration"].as_str().map(String::from),
             notification_property: opt_value(&body["NotificationProperty"]),
             non_overridable_arguments: parse_string_map(&body["NonOverridableArguments"]),
+            source_control_details: opt_value(&body["SourceControlDetails"]),
+            code_gen_configuration_nodes: opt_value(&body["CodeGenConfigurationNodes"]),
+            maintenance_window: body["MaintenanceWindow"].as_str().map(String::from),
+            log_uri: body["LogUri"].as_str().map(String::from),
+            allocated_capacity: body["AllocatedCapacity"].as_i64(),
         };
         state.jobs.insert(name.to_string(), job);
         Ok(AwsResponse::ok_json(json!({ "Name": name })))
@@ -155,21 +165,33 @@ impl GlueService {
     }
 
     pub(crate) fn get_jobs(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
+        let body = req.json_body();
         let accounts = self.state.read();
         let jobs: Vec<Value> = accounts
             .get(&req.account_id)
             .map(|s| s.jobs.values().map(job_to_json).collect())
             .unwrap_or_default();
-        Ok(AwsResponse::ok_json(json!({ "Jobs": jobs })))
+        let (page, token) = crate::common::paginate_body(&body, jobs);
+        let mut resp = json!({ "Jobs": page });
+        if let Some(t) = token {
+            resp["NextToken"] = json!(t);
+        }
+        Ok(AwsResponse::ok_json(resp))
     }
 
     pub(crate) fn list_jobs(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
+        let body = req.json_body();
         let accounts = self.state.read();
-        let names: Vec<String> = accounts
+        let names: Vec<Value> = accounts
             .get(&req.account_id)
-            .map(|s| s.jobs.keys().cloned().collect())
+            .map(|s| s.jobs.keys().map(|k| json!(k)).collect())
             .unwrap_or_default();
-        Ok(AwsResponse::ok_json(json!({ "JobNames": names })))
+        let (page, token) = crate::common::paginate_body(&body, names);
+        let mut resp = json!({ "JobNames": page });
+        if let Some(t) = token {
+            resp["NextToken"] = json!(t);
+        }
+        Ok(AwsResponse::ok_json(resp))
     }
 
     pub(crate) fn update_job(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
@@ -230,6 +252,21 @@ impl GlueService {
         if update["NonOverridableArguments"].is_object() {
             job.non_overridable_arguments = parse_string_map(&update["NonOverridableArguments"]);
         }
+        if !update["SourceControlDetails"].is_null() {
+            job.source_control_details = opt_value(&update["SourceControlDetails"]);
+        }
+        if !update["CodeGenConfigurationNodes"].is_null() {
+            job.code_gen_configuration_nodes = opt_value(&update["CodeGenConfigurationNodes"]);
+        }
+        if let Some(s) = update["MaintenanceWindow"].as_str() {
+            job.maintenance_window = Some(s.to_string());
+        }
+        if let Some(s) = update["LogUri"].as_str() {
+            job.log_uri = Some(s.to_string());
+        }
+        if let Some(n) = update["AllocatedCapacity"].as_i64() {
+            job.allocated_capacity = Some(n);
+        }
         job.last_modified_on = Utc::now();
         Ok(AwsResponse::ok_json(json!({ "JobName": name })))
     }

crates/fakecloud-glue/src/service.rs

@@ -920,6 +920,12 @@ pub(crate) fn table_json(t: &Table) -> Value {
         "CreateTime": t.create_time.timestamp() as f64,
         "UpdateTime": t.update_time.timestamp() as f64,
     });
+    // Emit CatalogId only when populated: AWS omits absent members rather than
+    // returning an empty string, and tables restored from a pre-`catalog_id`
+    // snapshot carry an empty default.
+    if !t.catalog_id.is_empty() {
+        o["CatalogId"] = json!(t.catalog_id);
+    }
     if let Some(ref d) = t.description {
         o["Description"] = json!(d);
     }
@@ -944,6 +950,15 @@ pub(crate) fn table_json(t: &Table) -> Value {
     o
 }
 
+/// The current (latest) archived VersionId for a table, defaulting to "1" for
+/// tables predating the version archive. Glue reports this on GetTable(s).
+fn current_table_version(st: &crate::state::GlueState, db: &str, table: &str) -> String {
+    st.table_version_ids(db, table)
+        .last()
+        .map(|n| n.to_string())
+        .unwrap_or_else(|| "1".to_string())
+}
+
 pub(crate) fn partition_json(p: &Partition) -> Value {
     let mut o = json!({
         "Values": p.values,
@@ -1038,13 +1053,19 @@ impl GlueService {
     }
 
     pub(crate) fn get_databases(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
+        let body = req.json_body();
         let accounts = self.state.read();
         let dbs: Vec<Value> = accounts
             .get(&req.account_id)
             .and_then(|s| s.dbs_in(&req.region))
             .map(|map| map.values().map(database_json).collect())
             .unwrap_or_default();
-        Ok(AwsResponse::ok_json(json!({"DatabaseList": dbs})))
+        let (page, token) = crate::common::paginate_body(&body, dbs);
+        let mut resp = json!({ "DatabaseList": page });
+        if let Some(t) = token {
+            resp["NextToken"] = json!(t);
+        }
+        Ok(AwsResponse::ok_json(resp))
     }
 
     pub(crate) fn update_database(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
@@ -1104,6 +1125,7 @@ impl GlueService {
         let table = Table {
             name: name.clone(),
             database_name: db_name.clone(),
+            catalog_id: req.account_id.clone(),
             description: input["Description"].as_str().map(|s| s.to_string()),
             owner: input["Owner"].as_str().map(|s| s.to_string()),
             create_time: now,
@@ -1145,7 +1167,9 @@ impl GlueService {
             .tables
             .get(name)
             .ok_or_else(|| entity_not_found(format!("Table {name} not found")))?;
-        Ok(AwsResponse::ok_json(json!({"Table": table_json(t)})))
+        let mut tj = table_json(t);
+        tj["VersionId"] = json!(current_table_version(state, db_name, name));
+        Ok(AwsResponse::ok_json(json!({"Table": tj})))
     }
 
     pub(crate) fn get_tables(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
@@ -1161,19 +1185,24 @@ impl GlueService {
             .filter(|e| !e.is_empty())
             .and_then(|e| regex::Regex::new(e).ok());
         let accounts = self.state.read();
-        let tables: Vec<Value> = accounts
-            .get(&req.account_id)
-            .and_then(|s| s.dbs_in(&req.region))
-            .and_then(|dbs| dbs.get(db_name))
-            .map(|db| {
-                db.tables
-                    .values()
-                    .filter(|t| name_filter.as_ref().is_none_or(|re| re.is_match(&t.name)))
-                    .map(table_json)
-                    .collect()
-            })
-            .unwrap_or_default();
-        Ok(AwsResponse::ok_json(json!({"TableList": tables})))
+        let mut tables: Vec<Value> = Vec::new();
+        if let Some(s) = accounts.get(&req.account_id) {
+            if let Some(db) = s.dbs_in(&req.region).and_then(|dbs| dbs.get(db_name)) {
+                for t in db.tables.values() {
+                    if name_filter.as_ref().is_none_or(|re| re.is_match(&t.name)) {
+                        let mut tj = table_json(t);
+                        tj["VersionId"] = json!(current_table_version(s, db_name, &t.name));
+                        tables.push(tj);
+                    }
+                }
+            }
+        }
+        let (page, token) = crate::common::paginate_body(&body, tables);
+        let mut resp = json!({ "TableList": page });
+        if let Some(t) = token {
+            resp["NextToken"] = json!(t);
+        }
+        Ok(AwsResponse::ok_json(resp))
     }
 
     pub(crate) fn update_table(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
@@ -1341,6 +1370,19 @@ impl GlueService {
             .as_str()
             .ok_or_else(|| missing("TableName"))?;
         let expression = body["Expression"].as_str().unwrap_or("");
+        // Parallel-scan support: when a `Segment{SegmentNumber, TotalSegments}`
+        // is supplied, each worker must see a disjoint slice of the matching
+        // partitions so the union across all segments equals the full set with
+        // no duplicates. Partition the deterministic (BTreeMap-ordered) match
+        // list by index modulo TotalSegments.
+        let segment = &body["Segment"];
+        let (segment_number, total_segments) = if segment.is_object() {
+            let total = segment["TotalSegments"].as_u64().unwrap_or(1).max(1);
+            let number = segment["SegmentNumber"].as_u64().unwrap_or(0);
+            (number, total)
+        } else {
+            (0, 1)
+        };
         let accounts = self.state.read();
         let parts: Vec<Value> = accounts
             .get(&req.account_id)
@@ -1358,11 +1400,20 @@ impl GlueService {
                             &p.values,
                         )
                     })
-                    .map(partition_json)
+                    .enumerate()
+                    .filter(|(i, _)| {
+                        total_segments == 1 || (*i as u64) % total_segments == segment_number
+                    })
+                    .map(|(_, p)| partition_json(p))
                     .collect()
             })
             .unwrap_or_default();
-        Ok(AwsResponse::ok_json(json!({"Partitions": parts})))
+        let (page, token) = crate::common::paginate_body(&body, parts);
+        let mut resp = json!({ "Partitions": page });
+        if let Some(t) = token {
+            resp["NextToken"] = json!(t);
+        }
+        Ok(AwsResponse::ok_json(resp))
     }
 
     pub(crate) fn update_partition(

crates/fakecloud-glue/src/state.rs

@@ -187,6 +187,19 @@ pub struct Job {
     pub notification_property: Option<serde_json::Value>,
     #[serde(default)]
     pub non_overridable_arguments: BTreeMap<String, String>,
+    /// Standard Job fields that were previously accepted then dropped
+    /// (bug-hunt 2026-07-16). Round-tripped verbatim. `CodeGenConfigurationNodes`
+    /// is a large nested shape stored opaquely rather than dropped.
+    #[serde(default)]
+    pub source_control_details: Option<serde_json::Value>,
+    #[serde(default)]
+    pub code_gen_configuration_nodes: Option<serde_json::Value>,
+    #[serde(default)]
+    pub maintenance_window: Option<String>,
+    #[serde(default)]
+    pub log_uri: Option<String>,
+    #[serde(default)]
+    pub allocated_capacity: Option<i64>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -283,6 +296,9 @@ pub struct Database {
 pub struct Table {
     pub name: String,
     pub database_name: String,
+    /// Owning Data Catalog id (the account id). Reported on read as `CatalogId`.
+    #[serde(default)]
+    pub catalog_id: String,
     pub description: Option<String>,
     pub owner: Option<String>,
     pub create_time: DateTime<Utc>,