From 86677ee18b6e331dbd96a3e1fa20d76ec7296932 Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Fri, 10 Jul 2026 02:50:05 +0530 Subject: [PATCH] [connectors] transaction support for Iceberg snapshots Add a transaction_mode option (none | snapshot) to the Iceberg connector. In snapshot mode the initial snapshot is ingested as Feldera transactions: the whole snapshot in one transaction when timestamp_column is unset, or one per lateness range when it is set. Each execute_df call maps to one transaction. Parser entries carry the start label (the input queue starts the transaction on the first flushed entry) and a commit entry is pushed once the dataframe drains. This rides the existing non-fault-tolerant queue() path, which already honors the transaction fields on queue entries. Input-buffer staging is deferred: it needs the flush-based FT queue path that follow mode will introduce. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- crates/adapters/src/test/iceberg.rs | 155 +++++++++--------- crates/feldera-types/src/transport/iceberg.rs | 48 ++++++ crates/iceberg/src/input.rs | 65 +++++++- crates/pipeline-manager/src/api/main.rs | 1 + .../docs/connectors/sources/iceberg.md | 83 ++++++---- openapi.json | 11 ++ 6 files changed, 247 insertions(+), 116 deletions(-) diff --git a/crates/adapters/src/test/iceberg.rs b/crates/adapters/src/test/iceberg.rs index 990056c55c5..263278f4a3c 100644 --- a/crates/adapters/src/test/iceberg.rs +++ b/crates/adapters/src/test/iceberg.rs @@ -15,7 +15,7 @@ use feldera_types::{ }; use serde_json::json; -use std::{collections::HashMap, time::Instant}; +use std::time::Instant; use tempfile::NamedTempFile; use tracing::info; use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitExt}; @@ -64,7 +64,10 @@ fn data_to_ndjson(data: Vec) -> NamedTempFile { } /// Read a snapshot of an Iceberg table with records of type `T` to a temporary JSON file. -fn iceberg_snapshot_to_json(schema: &[Field], config: &HashMap) -> NamedTempFile +/// +/// `config` is the connector's transport config as a JSON object. This function +/// forces `mode = snapshot`. +fn iceberg_snapshot_to_json(schema: &[Field], config: serde_json::Value) -> NamedTempFile where T: DBData + SerializeWithContext @@ -78,11 +81,14 @@ where json_file.path().display() ); - let mut config = config.clone(); - config.insert("mode".to_string(), "snapshot".to_string()); + let mut config = config; + config + .as_object_mut() + .expect("iceberg connector config must be a JSON object") + .insert("mode".to_string(), json!("snapshot")); let (input_pipeline, err_receiver) = - iceberg_input_pipeline::(schema, &config, &json_file.path().display().to_string()); + iceberg_input_pipeline::(schema, config, &json_file.path().display().to_string()); input_pipeline.start(); wait( || input_pipeline.status().pipeline_complete() || err_receiver.len() > 0, @@ -102,7 +108,7 @@ where /// Build a pipeline that reads from an Iceberg table and writes to a JSON file. fn iceberg_input_pipeline( schema: &[Field], - config: &HashMap, + config: serde_json::Value, output_file_path: &str, ) -> (Controller, Receiver) where @@ -198,36 +204,63 @@ fn data(n_records: usize) -> Vec { #[test] #[cfg(feature = "iceberg-tests-fs")] fn iceberg_localfs_input_test_unordered() { - iceberg_localfs_input_test(&[], &|_| true); + iceberg_localfs_input_test(1_000_000, json!({}), &|_| true); } #[test] #[cfg(feature = "iceberg-tests-fs")] fn iceberg_localfs_input_test_ordered() { - iceberg_localfs_input_test( - &[("timestamp_column".to_string(), "ts".to_string())], - &|_| true, - ); + iceberg_localfs_input_test(1_000_000, json!({ "timestamp_column": "ts" }), &|_| true); } #[test] #[cfg(feature = "iceberg-tests-fs")] fn iceberg_localfs_input_test_ordered_with_filter() { iceberg_localfs_input_test( - &[ - ("timestamp_column".to_string(), "ts".to_string()), - ("snapshot_filter".to_string(), "i >= 10000".to_string()), - ], + 1_000_000, + json!({ "timestamp_column": "ts", "snapshot_filter": "i >= 10000" }), &|x| x.i >= 10000, ); } +/// A single parser task must ingest the whole snapshot correctly (the parallel +/// path defaults to 4 parsers and is covered by the tests above). +#[test] +#[cfg(feature = "iceberg-tests-fs")] +fn iceberg_localfs_input_test_single_parser() { + iceberg_localfs_input_test(100_000, json!({ "num_parsers": 1 }), &|_| true); +} + +/// `transaction_mode = snapshot` on an unordered read ingests the whole snapshot +/// in one Feldera transaction; the ingested data must be identical to a +/// non-transactional read. +#[test] +#[cfg(feature = "iceberg-tests-fs")] +fn iceberg_localfs_input_test_transactional() { + iceberg_localfs_input_test(100_000, json!({ "transaction_mode": "snapshot" }), &|_| { + true + }); +} + +/// `transaction_mode = snapshot` on an ordered read ingests one Feldera +/// transaction per lateness range; the ingested data must still be complete. +#[test] +#[cfg(feature = "iceberg-tests-fs")] +fn iceberg_localfs_input_test_ordered_transactional() { + iceberg_localfs_input_test( + 100_000, + json!({ "timestamp_column": "ts", "transaction_mode": "snapshot" }), + &|_| true, + ); +} + #[cfg(feature = "iceberg-tests-fs")] fn iceberg_localfs_input_test( - extra_config: &[(String, String)], + n_records: usize, + extra_config: serde_json::Value, filter: &dyn Fn(&IcebergTestStruct) -> bool, ) { - let data = data(1_000_000); + let data = data(n_records); let table_dir = tempfile::TempDir::new().unwrap(); let table_path = table_dir.path().display().to_string(); @@ -269,12 +302,18 @@ fn iceberg_localfs_input_test( .unwrap() .to_string(); + let mut config = json!({ "metadata_location": metadata_path }); + let config_obj = config.as_object_mut().unwrap(); + for (key, value) in extra_config + .as_object() + .expect("extra_config must be a JSON object") + { + config_obj.insert(key.clone(), value.clone()); + } + let mut json_file = iceberg_snapshot_to_json::( &IcebergTestStruct::schema_with_lateness(), - &[("metadata_location".to_string(), metadata_path.to_string())] - .into_iter() - .chain(extra_config.into_iter().cloned()) - .collect::>(), + config, ); let expected_zset = dbsp::OrdZSet::from_tuples( @@ -297,37 +336,17 @@ fn iceberg_glue_s3_input_test() { // Read delta table unordered. let mut json_file = iceberg_snapshot_to_json::( &IcebergTestStruct::schema_with_lateness(), - &[ - ("catalog_type".to_string(), "glue".to_string()), - ( - "glue.warehouse".to_string(), - "s3://feldera-iceberg-test/".to_string(), - ), - ( - "table_name".to_string(), - "iceberg_test.test_table".to_string(), - ), - ( - "glue.access-key-id".to_string(), - std::env::var("ICEBERG_TEST_AWS_ACCESS_KEY_ID").unwrap(), - ), - ( - "glue.secret-access-key".to_string(), - std::env::var("ICEBERG_TEST_AWS_SECRET_ACCESS_KEY").unwrap(), - ), - ("glue.region".to_string(), "us-east-1".to_string()), - ( - "s3.access-key-id".to_string(), - std::env::var("ICEBERG_TEST_AWS_ACCESS_KEY_ID").unwrap(), - ), - ( - "s3.secret-access-key".to_string(), - std::env::var("ICEBERG_TEST_AWS_SECRET_ACCESS_KEY").unwrap(), - ), - ("s3.region".to_string(), "us-east-1".to_string()), - ] - .into_iter() - .collect::>(), + json!({ + "catalog_type": "glue", + "glue.warehouse": "s3://feldera-iceberg-test/", + "table_name": "iceberg_test.test_table", + "glue.access-key-id": std::env::var("ICEBERG_TEST_AWS_ACCESS_KEY_ID").unwrap(), + "glue.secret-access-key": std::env::var("ICEBERG_TEST_AWS_SECRET_ACCESS_KEY").unwrap(), + "glue.region": "us-east-1", + "s3.access-key-id": std::env::var("ICEBERG_TEST_AWS_ACCESS_KEY_ID").unwrap(), + "s3.secret-access-key": std::env::var("ICEBERG_TEST_AWS_SECRET_ACCESS_KEY").unwrap(), + "s3.region": "us-east-1", + }), ); let zset = file_to_zset::(json_file.as_file_mut()); @@ -381,29 +400,15 @@ fn iceberg_rest_s3_input_test() { // Read delta table unordered. let mut json_file = iceberg_snapshot_to_json::( &IcebergTestStruct::schema_with_lateness(), - &[ - ("catalog_type".to_string(), "rest".to_string()), - ("rest.uri".to_string(), "http://localhost:8181".to_string()), - ( - "rest.warehouse".to_string(), - "s3://feldera-iceberg-test/".to_string(), - ), - ( - "table_name".to_string(), - "iceberg_test.test_table".to_string(), - ), - ( - "s3.access-key-id".to_string(), - std::env::var("ICEBERG_TEST_AWS_ACCESS_KEY_ID").unwrap(), - ), - ( - "s3.secret-access-key".to_string(), - std::env::var("ICEBERG_TEST_AWS_SECRET_ACCESS_KEY").unwrap(), - ), - ("s3.region".to_string(), "us-east-1".to_string()), - ] - .into_iter() - .collect::>(), + json!({ + "catalog_type": "rest", + "rest.uri": "http://localhost:8181", + "rest.warehouse": "s3://feldera-iceberg-test/", + "table_name": "iceberg_test.test_table", + "s3.access-key-id": std::env::var("ICEBERG_TEST_AWS_ACCESS_KEY_ID").unwrap(), + "s3.secret-access-key": std::env::var("ICEBERG_TEST_AWS_SECRET_ACCESS_KEY").unwrap(), + "s3.region": "us-east-1", + }), ); let zset = file_to_zset::(json_file.as_file_mut()); diff --git a/crates/feldera-types/src/transport/iceberg.rs b/crates/feldera-types/src/transport/iceberg.rs index 07af03760ae..9234879f2b6 100644 --- a/crates/feldera-types/src/transport/iceberg.rs +++ b/crates/feldera-types/src/transport/iceberg.rs @@ -48,6 +48,29 @@ pub enum IcebergCatalogType { S3Tables, } +/// Iceberg table transaction mode. +/// +/// Determines how the connector breaks up its input into Feldera transactions. +/// +/// * `none` - the connector does not break up its input into transactions. +/// * `snapshot` - ingest the initial snapshot of the table in one or several transactions. +/// +/// # How the table snapshot is ingested using transactions +/// +/// When `transaction_mode` is set to `snapshot`, the connector ingests the snapshot in one +/// or several transactions, depending on `timestamp_column`. If `timestamp_column` is not set, +/// the whole snapshot is ingested in a single Feldera transaction. If `timestamp_column` is set, +/// the connector ingests the snapshot in a series of timestamp ranges of width equal to the +/// `LATENESS` attribute of the column, each range in a separate transaction. +#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema, Default)] +pub enum IcebergTransactionMode { + #[default] + #[serde(rename = "none")] + None, + #[serde(rename = "snapshot")] + Snapshot, +} + /// AWS Glue catalog config. #[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)] pub struct GlueCatalogConfig { @@ -186,6 +209,13 @@ pub struct IcebergReaderConfig { /// Table read mode. pub mode: IcebergIngestMode, + /// Transaction mode. + /// + /// Determines how the connector breaks up its input into Feldera transactions. + /// See [`IcebergTransactionMode`]. Defaults to [`IcebergTransactionMode::None`]. + #[serde(default)] + pub transaction_mode: IcebergTransactionMode, + /// Table column that serves as an event timestamp. /// /// When this option is specified, and `mode` is one of `snapshot` or `snapshot_and_follow`, @@ -575,6 +605,24 @@ mod test { assert_eq!(config.max_retries(), 0); } + #[test] + fn transaction_mode_defaults_to_none() { + let config: IcebergReaderConfig = serde_json::from_str( + r#"{"mode":"snapshot","metadata_location":"file:///tmp/t/metadata.json"}"#, + ) + .unwrap(); + assert_eq!(config.transaction_mode, IcebergTransactionMode::None); + } + + #[test] + fn transaction_mode_snapshot_parses() { + let config: IcebergReaderConfig = serde_json::from_str( + r#"{"mode":"snapshot","metadata_location":"file:///tmp/t/metadata.json","transaction_mode":"snapshot"}"#, + ) + .unwrap(); + assert_eq!(config.transaction_mode, IcebergTransactionMode::Snapshot); + } + #[test] fn reader_config_roundtrips() { let config: IcebergReaderConfig = serde_json::from_str( diff --git a/crates/iceberg/src/input.rs b/crates/iceberg/src/input.rs index 6221c8f870c..2c3d8dc49ff 100644 --- a/crates/iceberg/src/input.rs +++ b/crates/iceberg/src/input.rs @@ -9,7 +9,7 @@ use feldera_adapterlib::{ errors::journal::ControllerError, format::{InputBuffer, ParseError}, transport::{ - InputConsumer, InputEndpoint, InputQueue, InputReader, InputReaderCommand, + InputConsumer, InputEndpoint, InputQueue, InputQueueEntry, InputReader, InputReaderCommand, IntegratedInputEndpoint, NonFtInputReaderCommand, }, utils::backoff::calculate_backoff_delay, @@ -24,7 +24,7 @@ use feldera_types::adapter_stats::ConnectorHealth; use feldera_types::{ config::{FtModel, PipelineConfig}, program_schema::Relation, - transport::iceberg::{IcebergCatalogType, IcebergReaderConfig}, + transport::iceberg::{IcebergCatalogType, IcebergReaderConfig, IcebergTransactionMode}, }; use futures_util::StreamExt; use iceberg::CatalogBuilder; @@ -48,6 +48,7 @@ use iceberg_catalog_s3tables::{ use iceberg_datafusion::IcebergStaticTableProvider; use iceberg_storage_opendal::OpenDalResolvingStorageFactory; use log::{debug, info, trace, warn}; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::{sync::Arc, thread}; use tokio::{ select, @@ -216,6 +217,8 @@ struct IcebergInputEndpointInner { consumer: Box, datafusion: SessionContext, queue: Arc, + /// Monotonic counter used to label snapshot transactions for observability. + transaction_index: AtomicUsize, } impl IcebergInputEndpointInner { @@ -237,6 +240,22 @@ impl IcebergInputEndpointInner { consumer, datafusion, queue, + transaction_index: AtomicUsize::new(0), + } + } + + /// Allocate a transaction for the next snapshot chunk. + /// + /// Returns `None` when `transaction_mode` is `none`, meaning the chunk is not + /// wrapped in a Feldera transaction. Otherwise returns `Some(Some(label))`, + /// where the label identifies the transaction in logs and metrics. + fn allocate_snapshot_transaction(&self) -> Option> { + match self.config.transaction_mode { + IcebergTransactionMode::None => None, + IcebergTransactionMode::Snapshot => { + let index = self.transaction_index.fetch_add(1, Ordering::AcqRel); + Some(Some(format!("snapshot-{index}"))) + } } } @@ -932,10 +951,15 @@ impl IcebergInputEndpointInner { } }; + // Each snapshot chunk is its own Feldera transaction (or none, depending on + // `transaction_mode`): the whole snapshot for an unordered read, one range + // for an ordered read. + let transaction = self.allocate_snapshot_transaction(); + // On terminal failure `execute_df` has already reported the error to the // consumer, which stops ingestion; nothing more to do here. let _ = self - .execute_df(df, true, &descr, input_stream, receiver) + .execute_df(df, true, &descr, transaction, input_stream, receiver) .await; } @@ -953,6 +977,10 @@ impl IcebergInputEndpointInner { /// /// * `descr` - dataframe description used to construct error message. /// + /// * `transaction` - when `Some`, the dataframe's records are wrapped in a + /// Feldera transaction: entries carry the start label and a commit entry is + /// pushed once the dataframe completes. + /// /// * `input_stream` - handle to push updates to. /// /// * `receiver` - used to block the function until the endpoint is unpaused. @@ -961,17 +989,35 @@ impl IcebergInputEndpointInner { dataframe: DataFrame, polarity: bool, descr: &str, + transaction: Option>, input_stream: &mut dyn ArrowStream, receiver: &mut Receiver, ) -> Result { + let is_transactional = transaction.is_some(); let max_retries = self.config.max_retries(); let mut retry_count = 0; loop { match self - .execute_df_inner(dataframe.clone(), polarity, input_stream, receiver) + .execute_df_inner( + dataframe.clone(), + polarity, + transaction.clone(), + input_stream, + receiver, + ) .await { Ok(total_records) => { + // Close the transaction once all records have been queued. The + // non-empty-buffer entries above start it lazily at flush time; + // this empty entry commits it after the last one flushes. + if is_transactional { + self.queue.push_entry( + InputQueueEntry::new_with_aux(Utc::now(), ()) + .with_commit_transaction(true), + Vec::new(), + ); + } self.consumer .update_connector_health(ConnectorHealth::healthy()); return Ok(total_records); @@ -1009,6 +1055,7 @@ impl IcebergInputEndpointInner { &self, dataframe: DataFrame, polarity: bool, + transaction: Option>, input_stream: &mut dyn ArrowStream, receiver: &mut Receiver, ) -> Result { @@ -1058,7 +1105,15 @@ impl IcebergInputEndpointInner { }) }, move |(buffer, errors, timestamp)| { - queue.push((buffer, errors), timestamp); + // Setting the start label on every entry is idempotent: the input + // queue starts the transaction on the first flushed entry and + // ignores the label thereafter. + queue.push_entry( + InputQueueEntry::new_with_aux(timestamp, ()) + .with_buffer(buffer) + .with_start_transaction(transaction.clone()), + errors, + ); }, ); diff --git a/crates/pipeline-manager/src/api/main.rs b/crates/pipeline-manager/src/api/main.rs index e7f26ed1509..c667211fa26 100644 --- a/crates/pipeline-manager/src/api/main.rs +++ b/crates/pipeline-manager/src/api/main.rs @@ -417,6 +417,7 @@ It contains the following fields: feldera_types::transport::iceberg::RestCatalogConfig, feldera_types::transport::iceberg::GlueCatalogConfig, feldera_types::transport::iceberg::S3TablesCatalogConfig, + feldera_types::transport::iceberg::IcebergTransactionMode, feldera_types::transport::postgres::PostgresReaderConfig, feldera_types::transport::postgres::PostgresCdcReaderConfig, feldera_types::transport::postgres::PostgresWriterConfig, diff --git a/docs.feldera.com/docs/connectors/sources/iceberg.md b/docs.feldera.com/docs/connectors/sources/iceberg.md index 1b8e175ad0c..a8957046afd 100644 --- a/docs.feldera.com/docs/connectors/sources/iceberg.md +++ b/docs.feldera.com/docs/connectors/sources/iceberg.md @@ -29,6 +29,8 @@ The Iceberg input connector does not yet support [fault tolerance](/pipelines/fa | Property | Type | Description | |-----------------------------|--------|---------------| | `mode`* | enum | Table read mode. Currently, the only supported mode is `snapshot`, in which the connector reads a snapshot of the table and stops.| +| `transaction_mode` | enum | Determines how the connector breaks up its input into transactions. Supported values are `none` (default) and `snapshot`. See [below](#transactions) for details. | +| `timestamp_column` | string | Table column that serves as an event timestamp. When this option is specified, table rows are ingested in the timestamp order, respecting the [`LATENESS`](/sql/streaming#lateness-expressions) property of the column: each ingested row has a timestamp no more than `LATENESS` time units earlier than the most recent timestamp of any previously ingested row. See details [below](#ingesting-time-series-data-from-iceberg). | | `snapshot_filter` | string |

Optional row filter. When specified, only rows that satisfy the filter condition are included in the snapshot. The condition must be a valid SQL Boolean expression that can be used in the `where` clause of the `select * from snapshot where ..` query.

This option can be used to specify the range of event times to include in the snapshot, e.g.: `ts BETWEEN TIMESTAMP '2005-01-01 00:00:00' AND TIMESTAMP '2010-12-31 23:59:59'`.

| `snapshot_id` | integer|

Optional table snapshot id. When this option is set, the connector reads the specified snapshot of the table.

Note: at most one of `version` and `datetime` options can be specified. When neither of the two options is specified, the latest snapshot of the table is used.

| `datetime` | string |

Optional timestamp for the snapshot in the ISO-8601/RFC-3339 format, e.g., "2024-12-09T16:09:53+00:00". When this option is set, the connector reads the version of the table as of the specified point in time (based on the server time recorded in the transaction log, not the event time encoded in the data).

Note: at most one of `version` and `datetime` options can be specified. When neither of the two options is specified, the latest committed version of the table is used.

| @@ -38,9 +40,6 @@ The Iceberg input connector does not yet support [fault tolerance](/pipelines/fa | `num_parsers` | integer| Number of parallel parsing tasks used to process data read from the table. Increasing this value can improve throughput by parsing record batches concurrently. Recommended range: 1-10. Default: `4`.| | `max_retries` | integer|

Maximum number of retries for reading the table snapshot. When reading the snapshot fails partway through, for example because an object store read times out or is throttled, the connector retries the entire read with exponential backoff. This is in addition to the lower-level retries performed by the object store client.

Defaults to unlimited retries. Set to `0` to disable retries.

| - - - [*]: Required fields ### Rest catalog configuration @@ -158,8 +157,29 @@ The following table lists supported Iceberg data types and corresponding Feldera Types that are currently not supported include Iceberg's nested data types (`struct`s, `list`s and `map`s), `uuid`, and timestamps with time zone. - - - + }]' +); +``` ## Examples diff --git a/openapi.json b/openapi.json index 2b5f12705bb..5d2b0220fe9 100644 --- a/openapi.json +++ b/openapi.json @@ -9919,6 +9919,9 @@ "type": "string", "description": "Table column that serves as an event timestamp.\n\nWhen this option is specified, and `mode` is one of `snapshot` or `snapshot_and_follow`,\ntable rows are ingested in the timestamp order, respecting the\n[`LATENESS`](https://docs.feldera.com/sql/streaming#lateness-expressions)\nproperty of the column: each ingested row has a timestamp no more than `LATENESS`\ntime units earlier than the most recent timestamp of any previously ingested row.\nThe ingestion is performed by partitioning the table into timestamp ranges of width\n`LATENESS`. Each range is processed sequentially, in increasing timestamp order.\n\n# Example\n\nConsider a table with timestamp column of type `TIMESTAMP` and lateness attribute\n`INTERVAL 1 DAY`. Assuming that the oldest timestamp in the table is\n`2024-01-01T00:00:00``, the connector will fetch all records with timestamps\nfrom `2024-01-01`, then all records for `2024-01-02`, `2024-01-03`, etc., until all records\nin the table have been ingested.\n\n# Requirements\n\n* The timestamp column must be of a supported type: integer, `DATE`, or `TIMESTAMP`.\n* The timestamp column must be declared with non-zero `LATENESS`.\n* For efficient ingest, the table must be optimized for timestamp-based\nqueries using partitioning, Z-ordering, or liquid clustering.", "nullable": true + }, + "transaction_mode": { + "$ref": "#/components/schemas/IcebergTransactionMode" } }, "additionalProperties": { @@ -9929,6 +9932,14 @@ ], "description": "Iceberg input connector configuration." }, + "IcebergTransactionMode": { + "type": "string", + "description": "Iceberg table transaction mode.\n\nDetermines how the connector breaks up its input into Feldera transactions.\n\n* `none` - the connector does not break up its input into transactions.\n* `snapshot` - ingest the initial snapshot of the table in one or several transactions.\n\n# How the table snapshot is ingested using transactions\n\nWhen `transaction_mode` is set to `snapshot`, the connector ingests the snapshot in one\nor several transactions, depending on `timestamp_column`. If `timestamp_column` is not set,\nthe whole snapshot is ingested in a single Feldera transaction. If `timestamp_column` is set,\nthe connector ingests the snapshot in a series of timestamp ranges of width equal to the\n`LATENESS` attribute of the column, each range in a separate transaction.", + "enum": [ + "none", + "snapshot" + ] + }, "InputEndpointConfig": { "allOf": [ {