@@ -13,7 +13,7 @@ use crate::{ |
13 | 13 | }, |
14 | 14 | }; |
15 | 15 | use arrow::{ |
16 | | - array::{AsArray, RecordBatch}, |
| 16 | + array::{ArrayRef, AsArray, BooleanArray, Int64Array, RecordBatch, StringArray}, |
17 | 17 | compute::cast, |
18 | 18 | datatypes::{DataType as ArrowDataType, FieldRef, Int64Type, Schema as ArrowSchema}, |
19 | 19 | }; |
@@ -35,6 +35,7 @@ use feldera_adapterlib::{ |
35 | 35 | #[cfg(feature = "delta-unity-test")] |
36 | 36 | use feldera_storage::tokio::TOKIO; |
37 | 37 | use feldera_types::{ |
| 38 | + adapter_stats::ExternalTransactionPhase, |
38 | 39 | config::{PipelineConfig, TransportConfig}, |
39 | 40 | constants::DATAFUSION_TEMP_DIR, |
40 | 41 | program_schema::{ColumnType, Field, Relation, SqlIdentifier}, |
@@ -232,8 +233,13 @@ fn assert_read_change_data(pipeline: &Controller) { |
232 | 233 | } |
233 | 234 |
|
234 | 235 | fn delta_connector_gauge(pipeline: &Controller, metric_name: &str) -> f64 { |
| 236 | + delta_endpoint_gauge(pipeline, DELTA_TEST_INPUT_ENDPOINT, metric_name) |
| 237 | +} |
| 238 | + |
| 239 | +/// [`delta_connector_gauge`] of the input endpoint named `endpoint`. |
| 240 | +fn delta_endpoint_gauge(pipeline: &Controller, endpoint: &str, metric_name: &str) -> f64 { |
235 | 241 | let endpoint_id = pipeline |
236 | | - .input_endpoint_id_by_name(DELTA_TEST_INPUT_ENDPOINT) |
| 242 | + .input_endpoint_id_by_name(endpoint) |
237 | 243 | .expect("delta input endpoint must exist"); |
238 | 244 | let custom_metrics = pipeline |
239 | 245 | .status() |
@@ -4787,6 +4793,248 @@ pub fn catchup_end_version_caps_window_test(engine: &dyn PipelineTestEngine) { |
4787 | 4793 | }) |
4788 | 4794 | } |
4789 | 4795 |
|
| 4796 | +/// Appends one Delta commit of `count` rows of [`basic_relation`], with ids from `first`. |
| 4797 | +/// With `split`, the commit has many small files instead of one. |
| 4798 | +/// |
| 4799 | +/// Builds the columns directly, since JSON rows are too slow for commits of a million rows. |
| 4800 | +async fn append_basic_rows(table: DeltaTable, first: i64, count: i64, split: bool) -> DeltaTable { |
| 4801 | + let relation = basic_relation(); |
| 4802 | + let schema = published_schema(&relation); |
| 4803 | + let ids = first..first + count; |
| 4804 | + let columns: [ArrayRef; 4] = [ |
| 4805 | + Arc::new(Int64Array::from_iter_values(ids.clone())), |
| 4806 | + Arc::new(BooleanArray::from_iter( |
| 4807 | + ids.clone().map(|id| Some(id % 2 == 0)), |
| 4808 | + )), |
| 4809 | + Arc::new(Int64Array::from_iter(ids.clone().map(Some))), |
| 4810 | + Arc::new(StringArray::from_iter_values( |
| 4811 | + ids.map(|id| format!("row {id}")), |
| 4812 | + )), |
| 4813 | + ]; |
| 4814 | + let columns = columns |
| 4815 | + .iter() |
| 4816 | + .zip(schema.fields()) |
| 4817 | + .map(|(column, field)| cast(column, field.data_type()).unwrap()) |
| 4818 | + .collect(); |
| 4819 | + let rows = RecordBatch::try_new(schema, columns).unwrap(); |
| 4820 | + let write = table |
| 4821 | + .write(vec![delta::table_batch(&relation, &rows).unwrap()]) |
| 4822 | + .with_save_mode(SaveMode::Append); |
| 4823 | + // A one-byte target closes a file after each batch that the writer writes. |
| 4824 | + let write = if split { |
| 4825 | + write.with_target_file_size(std::num::NonZeroU64::new(1)) |
| 4826 | + } else { |
| 4827 | + write |
| 4828 | + }; |
| 4829 | + write.await.unwrap() |
| 4830 | +} |
| 4831 | + |
| 4832 | +/// The phase of `endpoint` in the open transaction, or `None` if it takes no part in it. |
| 4833 | +fn transaction_phase(pipeline: &Controller, endpoint: &str) -> Option<ExternalTransactionPhase> { |
| 4834 | + pipeline |
| 4835 | + .api_status(false) |
| 4836 | + .global_metrics |
| 4837 | + .transaction_initiators |
| 4838 | + .initiated_by_connectors |
| 4839 | + .get(endpoint) |
| 4840 | + .map(|connector| connector.phase) |
| 4841 | +} |
| 4842 | + |
| 4843 | +/// Regression test for #7746: a paused reader must not keep its reader permit. |
| 4844 | +/// |
| 4845 | +/// With one permit: |
| 4846 | +/// |
| 4847 | +/// 1. A transaction that the API opens stays open while the connectors read. |
| 4848 | +/// 2. `heavy` votes to commit, then reads a large file and pauses in the middle of it, |
| 4849 | +/// because the controller does not drain a connector that has voted. |
| 4850 | +/// 3. `light` joined the transaction with the first file of a commit and needs the permit |
| 4851 | +/// for the next file. If `heavy` keeps the permit, the transaction never commits. |
| 4852 | +#[engine_test(requires(Storage))] |
| 4853 | +pub fn catchup_paused_reader_holds_permit_test(engine: &dyn PipelineTestEngine) { |
| 4854 | + paused_reader_holds_permit(engine, "catchup"); |
| 4855 | +} |
| 4856 | + |
| 4857 | +/// [`catchup_paused_reader_holds_permit_test`] in `always` mode. |
| 4858 | +#[engine_test(requires(Storage))] |
| 4859 | +pub fn always_paused_reader_holds_permit_test(engine: &dyn PipelineTestEngine) { |
| 4860 | + paused_reader_holds_permit(engine, "always"); |
| 4861 | +} |
| 4862 | + |
| 4863 | +fn paused_reader_holds_permit(engine: &dyn PipelineTestEngine, transaction_mode: &str) { |
| 4864 | + run_async_on(4, async { |
| 4865 | + const HEAVY: &str = "heavy"; |
| 4866 | + const LIGHT: &str = "light"; |
| 4867 | + // Far more rows than the batches in flight between the scan and the buffer, so |
| 4868 | + // that `heavy` sees its pause in the middle of the file. |
| 4869 | + const HEAVY_FILE_ROWS: i64 = 500_000; |
| 4870 | + // Enough rows that `light` still reads them when `heavy` gets the permit. |
| 4871 | + const LIGHT_ROWS: i64 = 1_000_000; |
| 4872 | + const STEP_TIMEOUT_MS: u64 = 300_000; |
| 4873 | + const COMMIT_TIMEOUT_MS: u64 = 60_000; |
| 4874 | + const PHASE: &str = "input_connector_delta_phase"; |
| 4875 | + const LAST_INGESTED: &str = "input_connector_delta_last_ingested_version"; |
| 4876 | + const CATCHUP_TARGET: &str = "input_connector_delta_catchup_target_version"; |
| 4877 | + init_logging(); |
| 4878 | + |
| 4879 | + let relation = basic_relation(); |
| 4880 | + let arrow_schema = ArrowSchema::new(relation_to_arrow_fields( |
| 4881 | + &relation.fields, |
| 4882 | + delta_schema_options(), |
| 4883 | + )); |
| 4884 | + let struct_fields: Vec<StructField> = arrow_schema |
| 4885 | + .fields |
| 4886 | + .iter() |
| 4887 | + .map(|f| { |
| 4888 | + let data_type = DataType::try_from_arrow(f.data_type()).unwrap(); |
| 4889 | + StructField::new(f.name(), data_type, f.is_nullable()) |
| 4890 | + }) |
| 4891 | + .collect(); |
| 4892 | + let heavy_dir = TempDir::new().unwrap(); |
| 4893 | + let light_dir = TempDir::new().unwrap(); |
| 4894 | + let storage_dir = TempDir::new().unwrap(); |
| 4895 | + let heavy_uri = heavy_dir.path().display().to_string(); |
| 4896 | + let light_uri = light_dir.path().display().to_string(); |
| 4897 | + let mut heavy = create_table(&heavy_uri, &HashMap::new(), &struct_fields).await; |
| 4898 | + let light = create_table(&light_uri, &HashMap::new(), &struct_fields).await; |
| 4899 | + |
| 4900 | + let input = |uri: &str| { |
| 4901 | + json!({ |
| 4902 | + "stream": "test_input1", |
| 4903 | + "transport": { |
| 4904 | + "name": "delta_table_input", |
| 4905 | + "config": { |
| 4906 | + "uri": uri, |
| 4907 | + "mode": "follow", |
| 4908 | + "transaction_mode": transaction_mode, |
| 4909 | + "max_concurrent_readers": 1, |
| 4910 | + } |
| 4911 | + } |
| 4912 | + }) |
| 4913 | + }; |
| 4914 | + let mut heavy_input = input(&heavy_uri); |
| 4915 | + heavy_input["max_queued_records"] = 1.into(); |
| 4916 | + let mut inputs = Map::new(); |
| 4917 | + inputs.insert(HEAVY.to_string(), heavy_input); |
| 4918 | + inputs.insert(LIGHT.to_string(), input(&light_uri)); |
| 4919 | + let pipeline_config: PipelineConfig = serde_json::from_value(json!({ |
| 4920 | + "name": "test", |
| 4921 | + "workers": 4, |
| 4922 | + "storage_config": { "path": storage_dir.path() }, |
| 4923 | + "inputs": inputs, |
| 4924 | + })) |
| 4925 | + .unwrap(); |
| 4926 | + |
| 4927 | + let errors: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new())); |
| 4928 | + let pipeline = { |
| 4929 | + let errors = errors.clone(); |
| 4930 | + off_runtime(|| { |
| 4931 | + engine.controller( |
| 4932 | + &identity(&relation, &[Some("output")]), |
| 4933 | + &pipeline_config, |
| 4934 | + Box::new(move |e, _| errors.lock().unwrap().push(e.to_string())), |
| 4935 | + ) |
| 4936 | + }) |
| 4937 | + .unwrap() |
| 4938 | + }; |
| 4939 | + pipeline.start(); |
| 4940 | + for endpoint in [HEAVY, LIGHT] { |
| 4941 | + wait( |
| 4942 | + || delta_endpoint_gauge(&pipeline, endpoint, PHASE) == 1.0, |
| 4943 | + STEP_TIMEOUT_MS, |
| 4944 | + ) |
| 4945 | + .unwrap_or_else(|_| panic!("{endpoint} did not start to follow its table")); |
| 4946 | + } |
| 4947 | + |
| 4948 | + // 1. Open a transaction from the API, so that it stays open while the connectors read. |
| 4949 | + let transaction_id = pipeline.start_transaction().unwrap().transaction_id; |
| 4950 | + |
| 4951 | + // 2. `heavy` reads one row and votes to commit it. |
| 4952 | + heavy = append_basic_rows(heavy, 0, 1, false).await; |
| 4953 | + wait( |
| 4954 | + || transaction_phase(&pipeline, HEAVY) == Some(ExternalTransactionPhase::Committed), |
| 4955 | + STEP_TIMEOUT_MS, |
| 4956 | + ) |
| 4957 | + .expect("heavy did not vote to commit"); |
| 4958 | + |
| 4959 | + // 3. Write the next commits while both connectors are paused. |
| 4960 | + pipeline.pause_input_endpoint(HEAVY).unwrap(); |
| 4961 | + pipeline.pause_input_endpoint(LIGHT).unwrap(); |
| 4962 | + append_basic_rows(heavy, 1_000_000_000, HEAVY_FILE_ROWS, false).await; |
| 4963 | + // Follow mode reads each file in its own query, so `light` needs the permit again |
| 4964 | + // after each file. |
| 4965 | + let light = append_basic_rows(light, 2_000_000_000, LIGHT_ROWS, true).await; |
| 4966 | + let light_target = light.version().unwrap() as i64; |
| 4967 | + assert!(light.get_file_uris().unwrap().count() > 10); |
| 4968 | + |
| 4969 | + // 4. `light` joins the transaction with its first file, and holds the only permit |
| 4970 | + // while it reads it. |
| 4971 | + pipeline.start_input_endpoint(LIGHT).unwrap(); |
| 4972 | + wait( |
| 4973 | + || transaction_phase(&pipeline, LIGHT) == Some(ExternalTransactionPhase::Started), |
| 4974 | + STEP_TIMEOUT_MS, |
| 4975 | + ) |
| 4976 | + .expect("light did not join the transaction"); |
| 4977 | + |
| 4978 | + // 5. `heavy` waits for the permit and gets it between two of `light`'s files. Its |
| 4979 | + // buffer then fills, and the controller does not drain it, because `heavy` has |
| 4980 | + // voted to commit. |
| 4981 | + pipeline.start_input_endpoint(HEAVY).unwrap(); |
| 4982 | + wait( |
| 4983 | + || { |
| 4984 | + pipeline |
| 4985 | + .input_endpoint_status(HEAVY) |
| 4986 | + .is_ok_and(|status| status.metrics.buffered_records > 0) |
| 4987 | + }, |
| 4988 | + STEP_TIMEOUT_MS, |
| 4989 | + ) |
| 4990 | + .expect("heavy did not read its second commit"); |
| 4991 | + let light_last = delta_endpoint_gauge(&pipeline, LIGHT, LAST_INGESTED) as i64; |
| 4992 | + assert!( |
| 4993 | + light_last < light_target, |
| 4994 | + "light read all its files before heavy got the permit, so the case did not set \ |
| 4995 | + up the deadlock; raise LIGHT_ROWS" |
| 4996 | + ); |
| 4997 | + |
| 4998 | + // 6. The transaction can commit only after `light` reads its other files, which |
| 4999 | + // needs the permit. |
| 5000 | + pipeline.start_commit_transaction().unwrap(); |
| 5001 | + let committed = wait( |
| 5002 | + || { |
| 5003 | + pipeline.api_status(false).global_metrics.transaction_id != transaction_id |
| 5004 | + && delta_endpoint_gauge(&pipeline, LIGHT, LAST_INGESTED) as i64 == light_target |
| 5005 | + }, |
| 5006 | + COMMIT_TIMEOUT_MS, |
| 5007 | + ) |
| 5008 | + .is_ok(); |
| 5009 | + |
| 5010 | + let state = [HEAVY, LIGHT] |
| 5011 | + .iter() |
| 5012 | + .map(|endpoint| { |
| 5013 | + let metrics = pipeline.input_endpoint_status(endpoint).unwrap().metrics; |
| 5014 | + format!( |
| 5015 | + "{endpoint}: phase {:?}, buffered {}, total {}, last ingested {}, \ |
| 5016 | + catchup target {}", |
| 5017 | + transaction_phase(&pipeline, endpoint), |
| 5018 | + metrics.buffered_records, |
| 5019 | + metrics.total_records, |
| 5020 | + delta_endpoint_gauge(&pipeline, endpoint, LAST_INGESTED), |
| 5021 | + delta_endpoint_gauge(&pipeline, endpoint, CATCHUP_TARGET), |
| 5022 | + ) |
| 5023 | + }) |
| 5024 | + .collect::<Vec<_>>() |
| 5025 | + .join("\n"); |
| 5026 | + pipeline.stop().unwrap(); |
| 5027 | + |
| 5028 | + let errors = errors.lock().unwrap(); |
| 5029 | + assert!(errors.is_empty(), "connector errors: {errors:?}"); |
| 5030 | + assert!( |
| 5031 | + committed, |
| 5032 | + "transaction {transaction_id} did not commit within {COMMIT_TIMEOUT_MS} ms; \ |
| 5033 | + light needs version {light_target}:\n{state}" |
| 5034 | + ); |
| 5035 | + }) |
| 5036 | +} |
| 5037 | + |
4790 | 5038 | /// Large bursts across two rounds: each round still uses one Feldera transaction. |
4791 | 5039 | #[engine_test(requires(Storage))] |
4792 | 5040 | pub fn catchup_rapid_versions_test(engine: &dyn PipelineTestEngine) { |
|