Skip to content

Commit 5e8cf6b

Browse files
committed
feat(ingest): --continuous flag on new-datasource (buckets)
Marks a bucket datasource for native continuous sync: the worker's scheduler re-runs it incrementally, appending only newly-arrived objects. Adds IngestRequest.continuous (serialized only when set) + the --continuous flag, wired into the flag-driven filesystem builder and the wizard-skip check. Server ignores it for non-filesystem families. Test: --continuous rides through to the request body; default off is omitted from the JSON.
1 parent 977aec2 commit 5e8cf6b

3 files changed

Lines changed: 35 additions & 0 deletions

File tree

‎skills/hotdata/SKILL.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -256,6 +256,8 @@ hotdata ingest new-datasource --service buckets --bucket-url s3://bucket/prefix
256256
# Files in S3/GCS/Azure buckets (csv, jsonl, parquet); --glob narrows the match.
257257
# Public buckets need no credentials; private ones take --config @creds.json
258258
# ({"aws_access_key_id": …, "aws_secret_access_key": …, "endpoint_url": …}).
259+
# --continuous keeps a bucket datasource synced: it's re-run incrementally on a
260+
# schedule, appending only newly-arrived objects (no re-read of the whole bucket).
259261

260262
hotdata ingest new-datasource --service iceberg --config @catalog.json --table ns.orders
261263
# Iceberg via a REST catalog. --table is REQUIRED (repeatable, namespace.table).

‎src/client/ingest.rs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -428,6 +428,10 @@ pub struct IngestRequest {
428428
pub tables: Vec<String>,
429429
#[serde(skip_serializing_if = "std::ops::Not::not")]
430430
pub validate_only: bool,
431+
/// filesystem only: keep this datasource continuously synced — the scheduler
432+
/// re-runs it incrementally (append only new objects). Ignored otherwise.
433+
#[serde(skip_serializing_if = "std::ops::Not::not")]
434+
pub continuous: bool,
431435
#[serde(skip_serializing_if = "Option::is_none")]
432436
pub database_id: Option<String>,
433437
}

‎src/commands/ingest.rs‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -693,6 +693,11 @@ pub struct CreateArgs {
693693
#[arg(long)]
694694
glob: Option<String>,
695695

696+
/// Keep this datasource continuously synced — refreshed incrementally on a
697+
/// schedule, appending only newly-arrived objects (buckets only)
698+
#[arg(long)]
699+
continuous: bool,
700+
696701
/// Catalog type, e.g. rest (iceberg)
697702
#[arg(long = "catalog-type")]
698703
catalog_type: Option<String>,
@@ -713,6 +718,7 @@ impl CreateArgs {
713718
|| self.bucket_url.is_some()
714719
|| self.format.is_some()
715720
|| self.glob.is_some()
721+
|| self.continuous
716722
|| self.catalog_type.is_some()
717723
|| self.database_id.is_some()
718724
}
@@ -779,6 +785,7 @@ fn build_create_request(
779785
bucket_url: Some(args.bucket_url.ok_or("buckets connectors need --bucket-url")?),
780786
file_glob: args.glob,
781787
file_format: args.format,
788+
continuous: args.continuous,
782789
..Default::default()
783790
},
784791
"iceberg" => IngestRequest {
@@ -1837,6 +1844,7 @@ mod tests {
18371844
bucket_url: None,
18381845
format: None,
18391846
glob: None,
1847+
continuous: false,
18401848
catalog_type: None,
18411849
database_id: None,
18421850
}
@@ -1869,6 +1877,27 @@ mod tests {
18691877
assert_eq!(req.database_id.as_deref(), Some("db_1"));
18701878
}
18711879

1880+
#[test]
1881+
fn create_request_filesystem_carries_continuous_flag() {
1882+
let e = entry("buckets", "filesystem");
1883+
let mut args = create_args();
1884+
args.bucket_url = Some("s3://b/prefix".into());
1885+
args.format = Some("jsonl".into());
1886+
args.continuous = true;
1887+
let req = build_create_request(&e, args, None).unwrap();
1888+
assert_eq!(req.family, "filesystem");
1889+
assert_eq!(req.bucket_url.as_deref(), Some("s3://b/prefix"));
1890+
assert!(req.continuous); // --continuous rides through to the request body
1891+
1892+
// Default is off, and it serializes only when true (skip_serializing_if).
1893+
let mut off = create_args();
1894+
off.bucket_url = Some("s3://b".into());
1895+
off.format = Some("jsonl".into());
1896+
let req_off = build_create_request(&e, off, None).unwrap();
1897+
assert!(!req_off.continuous);
1898+
assert!(!serde_json::to_string(&req_off).unwrap().contains("continuous"));
1899+
}
1900+
18721901
#[test]
18731902
fn create_request_rejects_invalid_names() {
18741903
let e = entry("postgres", "sql");

0 commit comments

Comments
 (0)