@@ -12,7 +12,7 @@ use chrono::{NaiveDateTime, TimeDelta}; |
12 | 12 | use crossbeam::channel::Receiver; |
13 | 13 | use feldera_adapterlib::test::{Updates, ZSet, wait}; |
14 | 14 | use feldera_types::{ |
15 | | - adapter_stats::TransactionStatus, |
| 15 | + adapter_stats::{ConnectorHealthStatus, TransactionStatus}, |
16 | 16 | program_schema::{ColumnType, Field, Relation}, |
17 | 17 | }; |
18 | 18 | use serde_json::{Value, json}; |
@@ -930,54 +930,78 @@ fn rest_follow_config_with(table: &str, mode: &str, key: &str, value: Value) -> |
930 | 930 | config |
931 | 931 | } |
932 | 932 |
|
933 | | -/// Create (`op = "create"`) or append to (`op = "append"`) the REST-catalog test |
934 | | -/// table with `chunk`, producing a new snapshot. Returns the new snapshot's id so |
935 | | -/// a test can pin it as a follow start point. Shells out to the pyiceberg helper, |
936 | | -/// since iceberg-rust cannot write tables. |
937 | | -fn follow_table_op(op: &str, table: &str, chunk: &[Value]) -> i64 { |
938 | | - let ndjson = data_to_ndjson(chunk); |
| 933 | +/// Run the pyiceberg helper with `args` and return its stdout, since |
| 934 | +/// iceberg-rust can only append and these tests also overwrite rows. |
| 935 | +// TODO: write from Rust once iceberg-rust can overwrite and delete rows. |
| 936 | +fn follow_table_script(args: &[String]) -> String { |
939 | 937 | let script = "../iceberg/src/test/follow_table.py"; |
940 | 938 | let python = env_or("FELDERA_ICEBERG_PYTHON", "python3"); |
941 | 939 | let output = std::process::Command::new(python) |
942 | 940 | .arg(script) |
943 | | - .arg(format!("--op={op}")) |
944 | | - .arg(format!("--table={table}")) |
945 | | - .arg(format!("--json-file={}", ndjson.path().display())) |
| 941 | + .args(args) |
946 | 942 | .output() |
947 | 943 | .unwrap_or_else(|e| panic!("failed to run '{script}': {e}")); |
948 | 944 | if !output.status.success() { |
949 | 945 | panic!( |
950 | | - "'{script} --op={op}' failed (status {}):\nstdout: {}\nstderr: {}", |
| 946 | + "'{script} {args:?}' failed (status {}):\nstdout: {}\nstderr: {}", |
951 | 947 | output.status, |
952 | 948 | String::from_utf8_lossy(&output.stdout), |
953 | 949 | String::from_utf8_lossy(&output.stderr), |
954 | 950 | ); |
955 | 951 | } |
| 952 | + String::from_utf8_lossy(&output.stdout).into_owned() |
| 953 | +} |
| 954 | + |
| 955 | +/// Create (`op = "create"`) or append to (`op = "append"`) the REST-catalog test |
| 956 | +/// table with `chunk`, producing a new snapshot. Returns the new snapshot's id so |
| 957 | +/// a test can pin it as a follow start point. |
| 958 | +fn follow_table_op(op: &str, table: &str, chunk: &[Value]) -> i64 { |
| 959 | + let ndjson = data_to_ndjson(chunk); |
| 960 | + let stdout = follow_table_script(&[ |
| 961 | + format!("--op={op}"), |
| 962 | + format!("--table={table}"), |
| 963 | + format!("--json-file={}", ndjson.path().display()), |
| 964 | + ]); |
956 | 965 | // The writer prints the new snapshot id on its last stdout line. |
957 | | - let stdout = String::from_utf8_lossy(&output.stdout); |
958 | 966 | let last = stdout.trim().lines().last().unwrap_or_default(); |
959 | 967 | last.parse::<i64>() |
960 | | - .unwrap_or_else(|_| panic!("'{script} --op={op}' printed unexpected output: {last:?}")) |
| 968 | + .unwrap_or_else(|_| panic!("'--op={op}' printed unexpected output: {last:?}")) |
| 969 | +} |
| 970 | + |
| 971 | +/// Append `chunk` to `table` with one data file per row, in one snapshot. |
| 972 | +/// Returns the data files in manifest order. |
| 973 | +fn follow_table_append_files(table: &str, chunk: &[Value]) -> Vec<String> { |
| 974 | + let ndjson = data_to_ndjson(chunk); |
| 975 | + let stdout = follow_table_script(&[ |
| 976 | + "--op=append-files".to_string(), |
| 977 | + format!("--table={table}"), |
| 978 | + format!("--json-file={}", ndjson.path().display()), |
| 979 | + ]); |
| 980 | + stdout.lines().map(str::to_string).collect() |
| 981 | +} |
| 982 | + |
| 983 | +/// Break (`op = "break-file"`) or restore (`op = "restore-file"`) the data |
| 984 | +/// file `path` of `table`. |
| 985 | +fn follow_table_file_op(op: &str, table: &str, path: &str) { |
| 986 | + follow_table_script(&[ |
| 987 | + format!("--op={op}"), |
| 988 | + format!("--table={table}"), |
| 989 | + format!("--path={path}"), |
| 990 | + ]); |
961 | 991 | } |
962 | 992 |
|
963 | 993 | /// Create the REST-catalog test table `table` without any snapshot. |
964 | 994 | fn follow_table_create_empty(table: &str) { |
965 | | - let script = "../iceberg/src/test/follow_table.py"; |
966 | | - let python = env_or("FELDERA_ICEBERG_PYTHON", "python3"); |
967 | | - let output = std::process::Command::new(python) |
968 | | - .arg(script) |
969 | | - .arg("--op=create-empty") |
970 | | - .arg(format!("--table={table}")) |
971 | | - .output() |
972 | | - .unwrap_or_else(|e| panic!("failed to run '{script}': {e}")); |
973 | | - if !output.status.success() { |
974 | | - panic!( |
975 | | - "'{script} --op=create-empty' failed (status {}):\nstdout: {}\nstderr: {}", |
976 | | - output.status, |
977 | | - String::from_utf8_lossy(&output.stdout), |
978 | | - String::from_utf8_lossy(&output.stderr), |
979 | | - ); |
980 | | - } |
| 995 | + follow_table_script(&["--op=create-empty".to_string(), format!("--table={table}")]); |
| 996 | +} |
| 997 | + |
| 998 | +/// Whether the input connector of the test pipeline reports that it is unhealthy. |
| 999 | +fn input_is_unhealthy(pipeline: &Controller) -> bool { |
| 1000 | + pipeline |
| 1001 | + .input_endpoint_status("test_input1") |
| 1002 | + .ok() |
| 1003 | + .and_then(|s| s.health) |
| 1004 | + .is_some_and(|h| matches!(h.status, ConnectorHealthStatus::Unhealthy)) |
981 | 1005 | } |
982 | 1006 |
|
983 | 1007 | /// Read a single custom metric of the Iceberg connector, or `0.0` if absent. |
@@ -1479,6 +1503,42 @@ pub fn rest_follow_copy_on_write_delete(engine: &dyn PipelineTestEngine) { |
1479 | 1503 | ); |
1480 | 1504 | } |
1481 | 1505 |
|
| 1506 | +/// A retry after a read failed part way does not queue the read's rows again. |
| 1507 | +/// One parser reads the files in order, so the broken last file fails last. |
| 1508 | +#[engine_test(cfg(feature = "iceberg-tests-follow"), services(S3))] |
| 1509 | +pub fn rest_follow_retry_does_not_repeat_rows(engine: &dyn PipelineTestEngine) { |
| 1510 | + const TIMEOUT_MS: u64 = 120_000; |
| 1511 | + let all = data(10); |
| 1512 | + let (before, appended) = all.split_at(2); |
| 1513 | + let ns = env_or("FELDERA_ICEBERG_NAMESPACE", "follow_ns"); |
| 1514 | + let table = &format!("{ns}.retry_no_repeat"); |
| 1515 | + |
| 1516 | + // Break the last file before the connector can see the snapshot. |
| 1517 | + let start = follow_table_op("create", table, before); |
| 1518 | + let files = follow_table_append_files(table, appended); |
| 1519 | + assert_eq!( |
| 1520 | + files.len(), |
| 1521 | + appended.len(), |
| 1522 | + "expected one data file per row" |
| 1523 | + ); |
| 1524 | + let last = files.last().unwrap(); |
| 1525 | + follow_table_file_op("break-file", table, last); |
| 1526 | + |
| 1527 | + let mut config = rest_follow_config_with(table, "follow", "snapshot_id", json!(start)); |
| 1528 | + config["num_parsers"] = json!(1); |
| 1529 | + with_follow_pipeline_cfg(engine, config, |pipeline, out_path| { |
| 1530 | + wait(|| input_is_unhealthy(pipeline), TIMEOUT_MS) |
| 1531 | + .expect("the read of the broken file never failed"); |
| 1532 | + follow_table_file_op("restore-file", table, last); |
| 1533 | + |
| 1534 | + // A repeated row shows up in the output before the last row does. |
| 1535 | + let expected = follow_expected_zset(appended); |
| 1536 | + let _ = wait(|| follow_output_zset(out_path) == expected, TIMEOUT_MS); |
| 1537 | + assert_eq!(follow_output_zset(out_path), expected); |
| 1538 | + assert_eq!(ingested_records(pipeline), appended.len() as u64); |
| 1539 | + }); |
| 1540 | +} |
| 1541 | + |
1482 | 1542 | /// Number of complete `insert_delete` records written to the output file so |
1483 | 1543 | /// far (one JSON value per line). |
1484 | 1544 | fn output_record_count(path: &Path) -> usize { |
|