diff --git a/Cargo.lock b/Cargo.lock index 2cff97c192b..47f7f1040e2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3952,7 +3952,7 @@ dependencies = [ [[package]] name = "dbsp" -version = "0.317.0" +version = "0.318.0" dependencies = [ "anyhow", "arc-swap", @@ -4042,7 +4042,7 @@ dependencies = [ [[package]] name = "dbsp_adapters" -version = "0.317.0" +version = "0.318.0" dependencies = [ "actix", "actix-codec", @@ -4186,7 +4186,7 @@ dependencies = [ [[package]] name = "dbsp_nexmark" -version = "0.317.0" +version = "0.318.0" dependencies = [ "anyhow", "ascii_table", @@ -5170,7 +5170,7 @@ dependencies = [ [[package]] name = "fda" -version = "0.317.0" +version = "0.318.0" dependencies = [ "anyhow", "arrow", @@ -5222,7 +5222,7 @@ dependencies = [ [[package]] name = "feldera-adapterlib" -version = "0.317.0" +version = "0.318.0" dependencies = [ "actix-web", "anyhow", @@ -5256,7 +5256,7 @@ dependencies = [ [[package]] name = "feldera-buffer-cache" -version = "0.317.0" +version = "0.318.0" dependencies = [ "crossbeam-utils", "enum-map", @@ -5284,7 +5284,7 @@ dependencies = [ [[package]] name = "feldera-datagen" -version = "0.317.0" +version = "0.318.0" dependencies = [ "anyhow", "async-channel 2.5.0", @@ -5310,7 +5310,7 @@ dependencies = [ [[package]] name = "feldera-fxp" -version = "0.317.0" +version = "0.318.0" dependencies = [ "bytecheck", "dbsp", @@ -5330,7 +5330,7 @@ dependencies = [ [[package]] name = "feldera-iceberg" -version = "0.317.0" +version = "0.318.0" dependencies = [ "anyhow", "chrono", @@ -5352,7 +5352,7 @@ dependencies = [ [[package]] name = "feldera-ir" -version = "0.317.0" +version = "0.318.0" dependencies = [ "proptest", "proptest-derive", @@ -5364,7 +5364,7 @@ dependencies = [ [[package]] name = "feldera-macros" -version = "0.317.0" +version = "0.318.0" dependencies = [ "prettyplease", "proc-macro2", @@ -5374,7 +5374,7 @@ dependencies = [ [[package]] name = "feldera-observability" -version = "0.317.0" +version = "0.318.0" dependencies = [ "actix-http", "awc", @@ -5389,7 +5389,7 @@ dependencies = [ [[package]] name = "feldera-rest-api" -version = "0.317.0" +version = "0.318.0" dependencies = [ "chrono", "feldera-observability", @@ -5408,7 +5408,7 @@ dependencies = [ [[package]] name = "feldera-samply" -version = "0.317.0" +version = "0.318.0" dependencies = [ "crossbeam", "feldera-size-of", @@ -5444,7 +5444,7 @@ dependencies = [ [[package]] name = "feldera-sqllib" -version = "0.317.0" +version = "0.318.0" dependencies = [ "arcstr", "base58", @@ -5488,7 +5488,7 @@ dependencies = [ [[package]] name = "feldera-storage" -version = "0.317.0" +version = "0.318.0" dependencies = [ "anyhow", "crossbeam", @@ -5511,7 +5511,7 @@ dependencies = [ [[package]] name = "feldera-types" -version = "0.317.0" +version = "0.318.0" dependencies = [ "actix-web", "anyhow", @@ -8774,7 +8774,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" [[package]] name = "pipeline-manager" -version = "0.317.0" +version = "0.318.0" dependencies = [ "actix-cors", "actix-files", @@ -9984,7 +9984,7 @@ dependencies = [ [[package]] name = "readers" -version = "0.317.0" +version = "0.318.0" dependencies = [ "async-std", "csv", @@ -11683,7 +11683,7 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" [[package]] name = "sltsqlvalue" -version = "0.317.0" +version = "0.318.0" dependencies = [ "dbsp", "feldera-sqllib", @@ -12154,7 +12154,7 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" [[package]] name = "storage-test-compat" -version = "0.317.0" +version = "0.318.0" dependencies = [ "dbsp", "derive_more 1.0.0", diff --git a/Cargo.toml b/Cargo.toml index fe00c686930..29197d1193d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [workspace.package] authors = ["Feldera Team "] -version = "0.317.0" +version = "0.318.0" license = "MIT OR Apache-2.0" homepage = "https://github.com/feldera/feldera" repository = "https://github.com/feldera/feldera" @@ -108,7 +108,7 @@ csv = "1.2.2" csv-core = "0.1.10" dashmap = "6.1.0" datafusion = "53.1" -dbsp = { path = "crates/dbsp", version = "0.317.0" } +dbsp = { path = "crates/dbsp", version = "0.318.0" } dbsp_nexmark = { path = "crates/nexmark" } deadpool-postgres = "0.14.1" # Feldera fork of delta-rs: upstream tag `rust-v0.32.3` + 2 patches, on branch @@ -133,20 +133,20 @@ erased-serde = "0.3.31" fake = "2.10" fastbloom = "0.14.0" fdlimit = "0.3.0" -feldera-buffer-cache = { version = "0.317.0", path = "crates/buffer-cache" } +feldera-buffer-cache = { version = "0.318.0", path = "crates/buffer-cache" } feldera-cloud1-client = "0.1.2" feldera-datagen = { path = "crates/datagen" } -feldera-fxp = { version = "0.317.0", path = "crates/fxp", features = ["dbsp"] } +feldera-fxp = { version = "0.318.0", path = "crates/fxp", features = ["dbsp"] } feldera-iceberg = { path = "crates/iceberg" } -feldera-observability = { version = "0.317.0", path = "crates/feldera-observability" } -feldera-macros = { version = "0.317.0", path = "crates/feldera-macros" } -feldera-sqllib = { version = "0.317.0", path = "crates/sqllib" } -feldera-storage = { version = "0.317.0", path = "crates/storage" } -feldera-types = { version = "0.317.0", path = "crates/feldera-types" } -feldera-rest-api = { version = "0.317.0", path = "crates/rest-api" } -feldera-ir = { version = "0.317.0", path = "crates/ir" } -feldera-adapterlib = { version = "0.317.0", path = "crates/adapterlib" } -feldera-samply = { version = "0.317.0", path = "crates/samply" } +feldera-observability = { version = "0.318.0", path = "crates/feldera-observability" } +feldera-macros = { version = "0.318.0", path = "crates/feldera-macros" } +feldera-sqllib = { version = "0.318.0", path = "crates/sqllib" } +feldera-storage = { version = "0.318.0", path = "crates/storage" } +feldera-types = { version = "0.318.0", path = "crates/feldera-types" } +feldera-rest-api = { version = "0.318.0", path = "crates/rest-api" } +feldera-ir = { version = "0.318.0", path = "crates/ir" } +feldera-adapterlib = { version = "0.318.0", path = "crates/adapterlib" } +feldera-samply = { version = "0.318.0", path = "crates/samply" } flate2 = "1.1.0" form_urlencoded = "1.2.0" fs_extra = "1.3.0" diff --git a/crates/adapters/src/integrated/delta_table/input.rs b/crates/adapters/src/integrated/delta_table/input.rs index fb6ed653422..d804ada5745 100644 --- a/crates/adapters/src/integrated/delta_table/input.rs +++ b/crates/adapters/src/integrated/delta_table/input.rs @@ -7,7 +7,7 @@ use crate::util::JobQueue; use crate::{ControllerError, InputConsumer, InputReader, PipelineState}; use anyhow::{Error as AnyError, Result as AnyResult, anyhow, bail}; use arrow::array::BooleanArray; -use arrow::datatypes::Schema as ArrowSchema; +use arrow::datatypes::{FieldRef, Schema as ArrowSchema, SchemaRef}; use chrono::{DateTime, Utc}; use datafusion::catalog::TableProvider; use datafusion::common::DataFusionError; @@ -19,10 +19,11 @@ use datafusion::datasource::listing::{ use datafusion::physical_plan::{PhysicalExpr, displayable}; use datafusion::physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner}; use dbsp::circuit::tokio::TOKIO; +use delta_kernel::schema::{ColumnMetadataKey, MetadataValue}; use deltalake::datafusion::dataframe::DataFrame; use deltalake::datafusion::execution::context::SQLOptions; use deltalake::datafusion::logical_expr::SortExpr; -use deltalake::datafusion::prelude::SessionContext; +use deltalake::datafusion::prelude::{Expr, SessionContext, col}; use deltalake::datafusion::sql::sqlparser::dialect::GenericDialect; use deltalake::datafusion::sql::sqlparser::parser::Parser; use deltalake::datafusion::sql::sqlparser::tokenizer::Token; @@ -940,6 +941,20 @@ struct DeltaTableInputEndpointInner { /// even when marked unused, so the expressions can resolve them. Derived /// once from the immutable config. config_referenced_columns: OnceLock, + + /// Table handle whose snapshot supplies the schema used to read the table's + /// data files. Set to the open table once it is loaded, so it is always + /// present before any data is read. The version it reflects only ever moves + /// forward (see [`schema_snapshot`](Self::schema_snapshot)): + /// + /// * Snapshot mode leaves it at the opened version. + /// * While following, each commit's data is read against the schema that was + /// active when that commit was written: [`advance_schema`](Self::advance_schema) + /// adopts the new schema whenever a commit carries a `metaData` action. A + /// renamed column keeps its physical `col-` name, a column added + /// later null-fills for earlier commits, and a dropped column is ignored + /// once gone. + schema_table: Mutex>>, } #[derive(Debug, Clone)] @@ -989,6 +1004,7 @@ impl DeltaTableInputEndpointInner { catchup_follow_state: Mutex::new(CatchupFollowState::default()), used_sql_columns: OnceLock::new(), config_referenced_columns: OnceLock::new(), + schema_table: Mutex::new(None), } } @@ -1206,9 +1222,9 @@ impl DeltaTableInputEndpointInner { /// /// Delta schemas carry no case-sensitivity information, so column names are /// matched both as-is and lowercased (see [`ColumnNameSet`]). - fn used_columns(&self, table: &DeltaTable) -> Vec { + fn used_columns(&self) -> Vec { let used = self.used_sql_columns(); - table + self.schema_snapshot() .snapshot() .expect("Delta table snapshot must be loaded before computing used columns") .schema() @@ -1218,8 +1234,8 @@ impl DeltaTableInputEndpointInner { .collect() } - fn used_column_list(&self, table: &DeltaTable) -> String { - let columns = self.used_columns(table); + fn used_column_list(&self) -> String { + let columns = self.used_columns(); columns .iter() @@ -1344,7 +1360,7 @@ impl DeltaTableInputEndpointInner { input_stream: &mut dyn ArrowStream, receiver: &mut Receiver, ) -> AnyResult { - let column_names = self.used_column_list(table); + let column_names = self.used_column_list(); let mut snapshot_query = format!("select {column_names} from snapshot"); if let Some(filter) = self.effective_snapshot_filter() { @@ -1514,7 +1530,7 @@ impl DeltaTableInputEndpointInner { ) })?; - let columns = self.used_columns(table); + let columns = self.used_columns(); let column_names = columns .iter() .map(quote_sql_identifier) @@ -1660,6 +1676,10 @@ impl DeltaTableInputEndpointInner { let table = Arc::new(table); + // The opened table is the schema source until following pins it to the + // latest version; this makes it available to the snapshot reads below. + *self.schema_table.lock().unwrap() = Some(Arc::clone(&table)); + if let Err(e) = self.prepare_snapshot_query(&table).await { let _ = init_status_sender.send(Err(e)).await; return; @@ -1732,6 +1752,11 @@ impl DeltaTableInputEndpointInner { // Start following the table if required by the configuration. if self.config.follow() { + // The schema source stays at the version we opened at; each commit's + // schema change advances it as the loop reaches that commit (see + // `advance_schema`), so every commit's data is read against the + // schema that was active when it was written. + // Log the appropriate message based on whether we just completed a snapshot if self.config.snapshot() && snapshot_incomplete { // We just completed reading a snapshot, now switching to follow mode @@ -2695,6 +2720,10 @@ impl DeltaTableInputEndpointInner { ); } + // Adopt a schema change carried by this commit before reading its data, + // so the files are interpreted against the right schema. + self.advance_schema(new_version, actions).await?; + // Use the time when we _started_ reading transaction data as the ingestion timestamp. let timestamp = Utc::now(); @@ -2712,7 +2741,7 @@ impl DeltaTableInputEndpointInner { // Compute the projected read set once for the whole transaction; each // `process_action` borrows the `&str` view rather than re-collecting it // per Add/Remove. `used_column_names` owns the strings the view points at. - let used_column_names = self.used_columns(table); + let used_column_names = self.used_columns(); let used_columns: Vec<&str> = used_column_names.iter().map(String::as_str).collect(); // TODO: consider processing all Add actions and all Remove actions in one @@ -2891,21 +2920,163 @@ impl DeltaTableInputEndpointInner { Ok(()) } - /// Create a table provider from a list of Parquet files. + /// The table handle whose snapshot defines the current schema (see the + /// [`schema_table`](Self#structfield.schema_table) field). Set before any + /// data is read. + fn schema_snapshot(&self) -> Arc { + Arc::clone( + self.schema_table + .lock() + .unwrap() + .as_ref() + .expect("schema table must be set before reading the table schema"), + ) + } + + /// Reload the table at `version` and adopt it as the schema source, unless + /// the current schema is already at `version`. Mirrors + /// [`open_table`](Self::open_table)'s storage configuration without its + /// resume/version-selection logic. + async fn pin_schema_to_version(&self, version: u64) -> AnyResult<()> { + if self.schema_snapshot().version() == Some(version) { + return Ok(()); + } + let url = ensure_table_uri(&self.config.uri) + .map_err(|e| anyhow!("invalid Delta table uri '{}': {e}", &self.config.uri))?; + let table = self + .retry( + &format!("error loading Delta table schema at version {version}"), + Some("delta-schema-load"), + || async { + // DeltaTableBuilder isn't Clone, so build it anew on each retry. + DeltaTableBuilder::from_url(url.clone()) + .map_err(|e| anyhow!("invalid Delta table URL '{url}': {e}"))? + .with_storage_options(self.config.object_store_config.clone()) + .with_io_runtime(IORuntime::RT(TOKIO_DEDICATED_IO.handle().clone())) + .with_version(version) + .load() + .await + .map_err(|e| anyhow!("{e}")) + }, + ) + .await?; + *self.schema_table.lock().unwrap() = Some(Arc::new(table)); + Ok(()) + } + + /// Advance the schema source when a commit carries a schema change. + /// + /// A commit that changes the schema writes a `metaData` action holding the + /// schema effective from that commit on. We reload the table at that version + /// and adopt it, so this commit's data, and every later commit's, is read + /// against the schema that was active when it was written. Commits without a + /// `metaData` action leave the current schema in place. Moves forward only: + /// the guard skips a version we already hold (the opened or resumed snapshot + /// may already reflect it), since the schema never moves backward. + async fn advance_schema(&self, new_version: i64, actions: &[Action]) -> AnyResult<()> { + let changes_schema = actions.iter().any(|a| matches!(a, Action::Metadata(_))); + let current_version = self.schema_snapshot().version(); + if !changes_schema || current_version.is_some_and(|v| new_version <= v as i64) { + return Ok(()); + } + self.pin_schema_to_version(new_version as u64).await + } + + /// Logical-to-physical column-name pairs under Delta column mapping. + /// + /// With `delta.columnMapping.mode = 'name'` or `'id'` each column lives on + /// disk under an opaque physical name (`col-`, from the field's + /// `physicalName` metadata) while the SQL schema and connector expressions + /// use logical names. Returns one pair per renamed field; empty when mapping + /// is off, so callers skip remapping. + fn column_mapping(&self) -> AnyResult> { + let schema_table = self.schema_snapshot(); + let state = schema_table + .snapshot() + .map_err(|e| anyhow!("error accessing Delta table snapshot: {e}"))?; + Ok(state + .schema() + .fields() + .filter_map(|field| { + match field.get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName) { + Some(MetadataValue::String(physical)) if physical != field.name() => { + Some((field.name().to_string(), physical.clone())) + } + _ => None, + } + }) + .collect()) + } + + /// Returns the Arrow schema to use when reading the raw data files, named as + /// they appear on disk: the table's logical schema restricted to the columns + /// `keep` accepts (in schema order, so unions with another read side line + /// up), with each column-mapped field renamed to its physical (`col-`) + /// name so DataFusion's by-name matching finds them. The same as the kept + /// logical schema when column mapping is off. + fn physical_read_schema(&self, keep: impl Fn(&str) -> bool) -> AnyResult { + let schema_table = self.schema_snapshot(); + let logical = schema_table + .snapshot() + .map_err(|e| anyhow!("error accessing Delta table snapshot: {e}"))? + .snapshot() + .arrow_schema(); + let pairs = self.column_mapping()?; + let to_physical: HashMap<&str, &str> = pairs + .iter() + .map(|(l, p)| (l.as_str(), p.as_str())) + .collect(); + let fields: Vec = logical + .fields() + .iter() + .filter(|f| keep(f.name())) + .map(|f| match to_physical.get(f.name().as_str()) { + Some(physical) => Arc::new(f.as_ref().clone().with_name(*physical)), + None => Arc::clone(f), + }) + .collect(); + Ok(Arc::new( + ArrowSchema::new(fields).with_metadata(logical.metadata().clone()), + )) + } + + /// Renames a DataFrame's physical columns back to their logical names. + /// + /// Follow/CDC reads apply [`physical_read_schema`](Self::physical_read_schema), + /// so a column-mapped DataFrame arrives with physical (`col-`) names, + /// while the rest of the pipeline expects logical names. Unmapped columns + /// (already logical, or Delta metadata like `__feldera_op`) pass through. + /// This function is a no-op when column mapping is off. + fn project_physical_to_logical(&self, df: DataFrame) -> AnyResult { + let pairs = self.column_mapping()?; + if pairs.is_empty() { + return Ok(df); + } + let to_logical: HashMap<&str, &str> = pairs + .iter() + .map(|(l, p)| (p.as_str(), l.as_str())) + .collect(); + let exprs: Vec = df + .schema() + .fields() + .iter() + .map(|f| match to_logical.get(f.name().as_str()) { + Some(logical) => col(f.name()).alias(*logical), + None => col(f.name()), + }) + .collect(); + df.select(exprs) + .map_err(|e| anyhow!("error remapping column-mapped names: {e}")) + } + async fn create_parquet_table( &self, - table: &DeltaTable, files: Vec, description: &str, ) -> AnyResult { - // `DeltaTable::snapshot()` returns the table state; calling `.snapshot()` - // on that state hands back the eager snapshot, which exposes the Arrow - // schema. - let schema = table - .snapshot() - .map_err(|e| anyhow!("error accessing Delta table snapshot: {e}"))? - .snapshot() - .arrow_schema(); + // Use the on-disk (physical) schema so the reader matches each file's + // columns; `project_physical_to_logical` renames them back afterwards. + let schema = self.physical_read_schema(|_| true)?; let mut urls = Vec::with_capacity(files.len()); for file in files.iter() { @@ -2955,25 +3126,9 @@ impl DeltaTableInputEndpointInner { ) .await?; - let snapshot_schema = table - .snapshot() - .map_err(|e| anyhow!("error accessing Delta table snapshot for {description}: {e}"))? - .snapshot() - .arrow_schema(); - - // Restrict the schema to the kept columns, preserving field order so - // unions with the unmasked side line up. - let fields: Vec<_> = snapshot_schema - .fields() - .iter() - .filter(|f| keep_column(f.name())) - .cloned() - .collect(); - let logical_schema = if fields.len() == snapshot_schema.fields().len() { - snapshot_schema - } else { - Arc::new(ArrowSchema::new(fields)) - }; + // The provider declares the kept columns under their physical names so + // it lines up with the unmasked side; the caller renames them back. + let read_schema = self.physical_read_schema(keep_column)?; // `Add.path` is URL-encoded per the Delta spec; decode it into a // real object-store key. @@ -2984,7 +3139,7 @@ impl DeltaTableInputEndpointInner { table.log_store().object_store(None), file_path, bitmap, - logical_schema, + read_schema, ) .await } @@ -3030,11 +3185,11 @@ impl DeltaTableInputEndpointInner { .iter() .map(|p| format!("{}{}", root_url.as_str(), p)) .collect(); - let listing_table = - Arc::new(self.create_parquet_table(table, files, description).await?); + let listing_table = Arc::new(self.create_parquet_table(files, description).await?); let df = self.datafusion.read_table(listing_table).map_err(|e| { anyhow!("internal error processing {description}; {REPORT_ERROR}; error reading Parquet files: {e}") })?; + let df = self.project_physical_to_logical(df)?; // Drop unused columns when `skip_unused_columns` is set, so // DataFusion never reads them off disk. dfs.push(self.project_cdc_columns(df).map_err(|e| { @@ -3055,9 +3210,10 @@ impl DeltaTableInputEndpointInner { description, ) .await?; - dfs.push(self.datafusion.read_table(provider).map_err(|e| { + let df = self.datafusion.read_table(provider).map_err(|e| { anyhow!("internal error processing {description}; {REPORT_ERROR}; error reading masked file '{path}': {e}") - })?); + })?; + dfs.push(self.project_physical_to_logical(df)?); } let mut dfs = dfs.into_iter(); @@ -3115,12 +3271,13 @@ impl DeltaTableInputEndpointInner { } } - // NOTE: Column projection (follow mode here, CDC mode in `process_cdc_transaction`) projects - // against the startup snapshot schema, which `create_parquet_table` forces onto every Parquet - // file we read. This assumes a stable schema across the log versions we follow. DataFusion's - // schema adapter handles additive evolution (new columns ignored, missing columns read as NULL); - // a renamed/dropped column also reads as NULL, and an uncastable type change errors at read time. - // Real per-version evolution would require reloading the table and re-deriving the column set. + // NOTE: Column projection (follow here, CDC in `process_cdc_transaction`) runs against the + // schema in `schema_table`, which `create_parquet_table` applies to every Parquet file we read. + // While following, that schema is the one active when the commit being read was written (see the + // field's docs and `advance_schema`). Column-mapped physical names are stable across a rename, so + // the reader handles each version's files against its own schema: DataFusion's schema adapter + // null-fills columns absent from a file and ignores ones the schema drops; a type change produces + // an error at read time when a value cannot be converted to the target type. #[allow(clippy::too_many_arguments)] async fn add_with_polarity( &self, @@ -3155,21 +3312,20 @@ impl DeltaTableInputEndpointInner { // don't use `object_store_url()` here. let full_path = format!("{}{}", table.log_store().root_url().as_str(), path); Arc::new( - self.create_parquet_table(table, vec![full_path], &description) + self.create_parquet_table(vec![full_path], &description) .await?, ) }; - let df = self - .datafusion - .read_table(provider) - .map_err(|e| { - anyhow!("internal error processing {description}; {REPORT_ERROR}; error reading Parquet file: {e}") - })? - .select_columns(used_columns) - .map_err(|e| { - anyhow!("internal error processing {description}; {REPORT_ERROR}; error selecting columns: {e}") - })?; + let df = self.datafusion.read_table(provider).map_err(|e| { + anyhow!("internal error processing {description}; {REPORT_ERROR}; error reading Parquet file: {e}") + })?; + // Translate column-mapped physical names back to logical before any + // logical-name projection or filter. + let df = self.project_physical_to_logical(df)?; + let df = df.select_columns(used_columns).map_err(|e| { + anyhow!("internal error processing {description}; {REPORT_ERROR}; error selecting columns: {e}") + })?; let df = if let Some(filter) = &self.config.filter { let expr = df diff --git a/openapi.json b/openapi.json index 6e10540482e..c7e65ffc4f9 100644 --- a/openapi.json +++ b/openapi.json @@ -10,7 +10,7 @@ "license": { "name": "MIT OR Apache-2.0" }, - "version": "0.317.0" + "version": "0.318.0" }, "paths": { "/config/authentication": { diff --git a/python/dbt-feldera/pyproject.toml b/python/dbt-feldera/pyproject.toml index d5b063af7d5..112b5f4efd5 100644 --- a/python/dbt-feldera/pyproject.toml +++ b/python/dbt-feldera/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta" name = "dbt-feldera" readme = "README.md" description = "The dbt adapter for Feldera — DBSP-native incremental view maintenance" -version = "0.317.0" +version = "0.318.0" license = "MIT" requires-python = ">=3.10" authors = [ diff --git a/python/felderize/pyproject.toml b/python/felderize/pyproject.toml index d5605195057..6fcdcc898fb 100644 --- a/python/felderize/pyproject.toml +++ b/python/felderize/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta" name = "felderize" readme = "README.md" description = "SQL dialect to Feldera SQL translator agent" -version = "0.317.0" +version = "0.318.0" license = "MIT" requires-python = ">=3.10" authors = [ diff --git a/python/pyproject.toml b/python/pyproject.toml index e163b585257..804216f0f5c 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta" name = "feldera" readme = "README.md" description = "The feldera python client" -version = "0.317.0" +version = "0.318.0" license = "MIT" requires-python = ">=3.10,<3.14" authors = [ diff --git a/python/tests/platform/fixtures/column_mapping.py b/python/tests/platform/fixtures/column_mapping.py index 30a4564d84b..34a034c7e8b 100644 --- a/python/tests/platform/fixtures/column_mapping.py +++ b/python/tests/platform/fixtures/column_mapping.py @@ -2,7 +2,7 @@ ``tests.utils.ensure_delta_spark_fixture`` runs this as a subprocess (``uv run --with "delta-spark>=4.2,<5" python column_mapping.py -``), where ```` is a ``delta.columnMapping.mode`` value: ``name`` +``). ```` is a ``delta.columnMapping.mode`` value: ``name`` (physical Parquet columns are renamed to ``col-``) or ``id`` (columns are matched by Parquet field ID). PySpark is currently the only writer that can produce a Delta table with column mapping enabled *and* perform the rename / @@ -26,9 +26,14 @@ * ``v5`` DROP COLUMN ``amount`` * ``v6`` INSERT one row under the final ``(id, full_name, country)`` schema -A snapshot read at the latest version must therefore return the final logical -schema with ``amount`` gone, ``country`` NULL for the rows written before it -existed, and every value resolved through column mapping. +A snapshot read of the latest version returns the final logical schema with +``amount`` gone, ``country`` NULL for the rows written before it existed, and +every value resolved through column mapping. The follow/CDC paths instead read +each commit's data file as raw Parquet against the schema active when that commit +was written, exercising the connector's own physical-name resolution across the +rename/add/drop history. A replay from v0 therefore differs from the snapshot on +the pre-rename rows: rows 1 and 2 were written under the logical name ``name``, +so ``full_name`` reads NULL for them (see ``_REPLAY_EXPECTED_ROWS`` in the test). """ from __future__ import annotations @@ -48,7 +53,8 @@ def build(table_path: str, mode: str = "name") -> None: - """Create the column-mapped, schema-evolved table at ``table_path``. + """Create a column-mapped table at ``table_path`` with the rename/add/drop + history documented in the module docstring. :param mode: ``delta.columnMapping.mode`` for the table: ``name`` or ``id``. """ @@ -71,15 +77,15 @@ def build(table_path: str, mode: str = "name") -> None: spark.sparkContext.setLogLevel("ERROR") try: t = f"delta.`{table_path}`" + props = ( + f"TBLPROPERTIES ('delta.columnMapping.mode' = '{mode}'," + " 'delta.minReaderVersion' = '2'," + " 'delta.minWriterVersion' = '5')" + ) spark.sql( - f""" - CREATE TABLE {t} (id BIGINT, name STRING, amount DOUBLE) - USING delta - TBLPROPERTIES ('delta.columnMapping.mode' = '{mode}', - 'delta.minReaderVersion' = '2', - 'delta.minWriterVersion' = '5') - """ + f"CREATE TABLE {t} (id BIGINT, name STRING, amount DOUBLE)" + f" USING delta {props}" ) spark.sql(f"INSERT INTO {t} VALUES (1,'alice',10.0),(2,'bob',20.0)") spark.sql(f"ALTER TABLE {t} RENAME COLUMN name TO full_name") diff --git a/python/tests/platform/test_delta_input_column_mapping.py b/python/tests/platform/test_delta_input_column_mapping.py index 970685933ce..298eee771ff 100644 --- a/python/tests/platform/test_delta_input_column_mapping.py +++ b/python/tests/platform/test_delta_input_column_mapping.py @@ -1,13 +1,20 @@ -"""Snapshot read of a column-mapped Delta table after schema evolution. +"""Read a column-mapped, schema-evolved Delta table in every ingest mode. A table with ``delta.columnMapping.mode = 'name'`` stores opaque physical Parquet column names (``col-``) that differ from the logical names; with ``mode = 'id'`` columns are matched by Parquet field ID instead. Either way, a -correct snapshot read must resolve every value through the column-mapping -metadata in the Delta log. PySpark seeds the fixture (it is currently the only -writer that supports column-mapping rename/drop); the builder lives in -``fixtures/column_mapping.py`` and runs via ``uv run --with delta-spark`` so -the Spark/JVM wheels are only fetched on cache miss. +correct read must resolve every value through the column-mapping metadata in +the Delta log. snapshot reads let delta-rs resolve this against the latest +schema; the follow and CDC paths read each commit's data file as raw Parquet and +must resolve the physical names themselves, against the schema active when that +commit was written. Because of that, replaying the rename/add/drop history from +v0 does not converge to a snapshot of the latest table: rows written before a +column was renamed resolve under the old logical name (NULL in the SQL relation). + +PySpark seeds the fixture (it is currently the only writer that supports +column-mapping rename/drop); the builder lives in ``fixtures/column_mapping.py`` +and runs via ``uv run --with delta-spark`` so the Spark/JVM wheels are only +fetched on cache miss. """ from __future__ import annotations @@ -26,32 +33,39 @@ TABLE = "t" CONNECTOR = "delta_in" # Bump to invalidate cached MinIO copies when the fixture definition changes. -FIXTURE_VERSION = "v1" +FIXTURE_VERSION = "v2" # Spark builder that writes the column-mapped, schema-evolved table. It runs in # a subprocess (see ensure_delta_spark_fixture) rather than being imported here. _FIXTURE_BUILDER = Path(__file__).parent / "fixtures" / "column_mapping.py" -def _build_sql(loc: DeltaTestLocation) -> str: +# The final, post-evolution logical columns. Used unless a test reads only part +# of the history (see the end_version test, which sees the pre-drop schema). +_FINAL_COLUMNS = "id BIGINT NOT NULL, full_name VARCHAR, country VARCHAR" + + +def _build_sql( + loc: DeltaTestLocation, + extra_config: dict | None = None, + columns: str = _FINAL_COLUMNS, +) -> str: + config = dict(loc.connector_config) + config.update(extra_config or {}) connectors = json.dumps( [ { "name": CONNECTOR, "transport": { "name": "delta_table_input", - "config": dict(loc.connector_config), + "config": config, }, } ] ).replace("'", "''") - # The SQL schema declares the final, post-evolution logical columns. return ( - f"CREATE TABLE {TABLE} (" - "id BIGINT NOT NULL," - "full_name VARCHAR," - "country VARCHAR" - f") WITH ('materialized' = 'true', 'connectors' = '{connectors}');" + f"CREATE TABLE {TABLE} ({columns})" + f" WITH ('materialized' = 'true', 'connectors' = '{connectors}');" ) @@ -66,18 +80,54 @@ def _snapshot_rows(pipeline) -> list[dict]: ) -def _run_column_mapping_snapshot_test(pipeline_name: str, mapping_mode: str) -> None: - """A snapshot read of a column-mapped, schema-evolved table returns the - final logical schema and correctly resolves physical column names. +# Replay the whole fixture history (v0 CREATE through v6) in follow/CDC mode. +# ``version`` is the already-consumed baseline (the empty v0 table), so the +# connector applies every later commit: the three inserts and the interleaved +# rename/add/drop. +_REPLAY_CONFIG = {"version": 0, "end_version": 6} + + +# Follow/CDC read each commit's data against the schema active when that commit +# was written, so a replay from v0 does NOT converge to a snapshot of the latest +# table on renamed columns. The v1 rows (1, 2) were written under the original +# logical name ``name``; at v1 the connector resolves them to ``name``, which the +# SQL relation (``full_name``) does not have, so ``full_name`` is NULL for them. +# Rows written at v4/v6 (after the v2 rename) carry ``full_name`` and resolve +# normally. Contrast ``EXPECTED_ROWS``, which a snapshot read of v6 produces by +# resolving the whole table against its final schema. +_REPLAY_EXPECTED_ROWS = [ + {"id": 1, "full_name": None, "country": None}, + {"id": 2, "full_name": None, "country": None}, + {"id": 3, "full_name": "carol", "country": "US"}, + {"id": 4, "full_name": "dave", "country": "UK"}, + {"id": 5, "full_name": "erin", "country": "FR"}, +] + + +def _run_column_mapping_test( + pipeline_name: str, + mapping_mode: str, + *, + mode: str, + extra_config: dict | None = None, +) -> None: + """Ingest the column-mapped table in ``mode`` and check that every value + resolves through the column-mapping metadata. - Driven by one wrapper test per ``delta.columnMapping.mode`` (rather than - ``pytest.mark.parametrize``) so each case gets a distinct pipeline name — - the ``pipeline_name`` fixture derives the name from the test function, and + snapshot mode reads the latest version directly, resolving the whole table + against its final schema (``EXPECTED_ROWS``). follow/CDC replay the log from + v0 and read each commit against the schema active when it was written, so + the pre-rename rows surface their old logical name and ``full_name`` is NULL + for them (``_REPLAY_EXPECTED_ROWS``). + + One wrapper test per (ingest mode, ``delta.columnMapping.mode``) pair rather + than ``pytest.mark.parametrize`` so each gets a distinct pipeline name — the + ``pipeline_name`` fixture derives the name from the test function, and parametrized cases sharing one name could collide under ``pytest -n``. """ loc = DeltaTestLocation.create( pipeline_name, - mode="snapshot", + mode=mode, stable_subpath=f"column_mapping_{mapping_mode}_{FIXTURE_VERSION}", ) try: @@ -86,7 +136,7 @@ def _run_column_mapping_snapshot_test(pipeline_name: str, mapping_mode: str) -> pipeline = PipelineBuilder( TEST_CLIENT, pipeline_name, - sql=_build_sql(loc), + sql=_build_sql(loc, extra_config), runtime_config=RuntimeConfig( workers=FELDERA_TEST_NUM_WORKERS, hosts=FELDERA_TEST_NUM_HOSTS, @@ -97,10 +147,12 @@ def _run_column_mapping_snapshot_test(pipeline_name: str, mapping_mode: str) -> pipeline.wait_for_completion(force_stop=False, timeout_s=600) rows = _snapshot_rows(pipeline) - assert rows == EXPECTED_ROWS, ( - "snapshot read must resolve column-mapped, schema-evolved data: " - "dropped 'amount' gone, 'country' NULL for pre-add rows; " - f"got {rows}" + expected = EXPECTED_ROWS if mode == "snapshot" else _REPLAY_EXPECTED_ROWS + assert rows == expected, ( + f"{mode} read must resolve column-mapped, schema-evolved data: " + "dropped 'amount' gone, 'country' NULL for pre-add rows, and " + "(follow/CDC only) 'full_name' NULL for rows written before the " + f"rename; got {rows}" ) pipeline.stop(force=True) @@ -110,9 +162,221 @@ def _run_column_mapping_snapshot_test(pipeline_name: str, mapping_mode: str) -> def test_delta_input_column_mapping_name_snapshot(pipeline_name): """Snapshot read with ``delta.columnMapping.mode = 'name'``.""" - _run_column_mapping_snapshot_test(pipeline_name, "name") + _run_column_mapping_test(pipeline_name, "name", mode="snapshot") def test_delta_input_column_mapping_id_snapshot(pipeline_name): """Snapshot read with ``delta.columnMapping.mode = 'id'``.""" - _run_column_mapping_snapshot_test(pipeline_name, "id") + _run_column_mapping_test(pipeline_name, "id", mode="snapshot") + + +def test_delta_input_column_mapping_name_follow(pipeline_name): + """Follow a column-mapped (``mode = 'name'``) table across schema evolution. + + The follow path reads each commit's data file as raw Parquet, resolving the + physical (``col-``) names against the schema active when that commit + was written. The v1 rows predate the ``name`` -> ``full_name`` rename, so + they resolve to ``name`` and read NULL for the SQL ``full_name`` column; + rows written after the rename resolve normally (``_REPLAY_EXPECTED_ROWS``). + """ + _run_column_mapping_test( + pipeline_name, + "name", + mode="snapshot_and_follow", + extra_config=_REPLAY_CONFIG, + ) + + +def test_delta_input_column_mapping_id_follow(pipeline_name): + """Follow a column-mapped (``mode = 'id'``) table across schema evolution.""" + _run_column_mapping_test( + pipeline_name, + "id", + mode="snapshot_and_follow", + extra_config=_REPLAY_CONFIG, + ) + + +# CDC mode requires a delete filter and an order-by; the fixture has no +# delete-marker column, so use an always-false predicate on a real logical +# column. Every row stays an insert, and resolving the filter and order-by +# against logical names confirms they survive the physical-name remap. +_CDC_CONFIG = { + **_REPLAY_CONFIG, + "cdc_delete_filter": "id < 0", + "cdc_order_by": "id asc", +} + + +def test_delta_input_column_mapping_name_cdc(pipeline_name): + """CDC-replay a column-mapped (``mode = 'name'``) table across evolution.""" + _run_column_mapping_test( + pipeline_name, "name", mode="cdc", extra_config=_CDC_CONFIG + ) + + +def test_delta_input_column_mapping_id_cdc(pipeline_name): + """CDC-replay a column-mapped (``mode = 'id'``) table across evolution.""" + _run_column_mapping_test(pipeline_name, "id", mode="cdc", extra_config=_CDC_CONFIG) + + +# Replaying a window that ends at v4 (after the rename and ADD country, before +# the v5 DROP amount) reads each commit against its own schema. The v1 rows +# (1, 2) were written under ``(id, name, amount)``: ``amount`` is present (10/20), +# but the column is logically ``name`` there, so the SQL ``full_name`` is NULL +# for them; ``country`` did not yet exist, so it is NULL too. The v4 rows (3, 4) +# were written after the rename and ADD, so every column resolves. ``amount`` +# survives for all rows because it is dropped only at v5, past this window. +_V4_COLUMNS = "id BIGINT NOT NULL, full_name VARCHAR, amount DOUBLE, country VARCHAR" +_V4_EXPECTED_ROWS = [ + {"id": 1, "full_name": None, "amount": 10.0, "country": None}, + {"id": 2, "full_name": None, "amount": 20.0, "country": None}, + {"id": 3, "full_name": "carol", "amount": 30.0, "country": "US"}, + {"id": 4, "full_name": "dave", "amount": 40.0, "country": "UK"}, +] + + +def test_delta_input_column_mapping_follow_end_version(pipeline_name): + """Follow a window ending at v4: each commit resolves against its own schema.""" + loc = DeltaTestLocation.create( + pipeline_name, + mode="snapshot_and_follow", + stable_subpath=f"column_mapping_name_{FIXTURE_VERSION}", + ) + try: + ensure_delta_spark_fixture(loc, _FIXTURE_BUILDER, builder_args=("name",)) + + pipeline = PipelineBuilder( + TEST_CLIENT, + pipeline_name, + sql=_build_sql( + loc, + extra_config={"version": 0, "end_version": 4}, + columns=_V4_COLUMNS, + ), + runtime_config=RuntimeConfig( + workers=FELDERA_TEST_NUM_WORKERS, + hosts=FELDERA_TEST_NUM_HOSTS, + logging="debug", + ), + ).create_or_replace() + pipeline.start() + pipeline.wait_for_completion(force_stop=False, timeout_s=600) + + rows = sorted( + ( + { + "id": r["id"], + "full_name": r["full_name"], + "amount": r["amount"], + "country": r["country"], + } + for r in pipeline.query( + f"SELECT id, full_name, amount, country FROM {TABLE}" + ) + ), + key=lambda r: r["id"], + ) + assert rows == _V4_EXPECTED_ROWS, ( + "replay up to v4 must read each commit against its own schema: " + "'amount' present (dropped only at v5), 'full_name' NULL for the " + f"pre-rename v1 rows; got {rows}" + ) + + pipeline.stop(force=True) + finally: + loc.cleanup() + + +# snapshot_and_follow from v2: snapshot the table as of v2, then follow later +# commits. The v2 snapshot carries rows 1 and 2 (written pre-rename under +# ``name``) with full_name resolved via column mapping; country is NULL, added +# only at v3. Following v3..v6 then adds rows 3, 4, 5. +_MID_HISTORY_EXPECTED_ROWS = [ + {"id": 1, "full_name": "alice", "country": None}, + {"id": 2, "full_name": "bob", "country": None}, + {"id": 3, "full_name": "carol", "country": "US"}, + {"id": 4, "full_name": "dave", "country": "UK"}, + {"id": 5, "full_name": "erin", "country": "FR"}, +] + + +def test_delta_input_column_mapping_resume_mid_history(pipeline_name): + """Snapshot at a non-zero version, then follow: the snapshot resolves the + pre-rename rows through column mapping and later commits are replayed.""" + loc = DeltaTestLocation.create( + pipeline_name, + mode="snapshot_and_follow", + stable_subpath=f"column_mapping_name_{FIXTURE_VERSION}", + ) + try: + ensure_delta_spark_fixture(loc, _FIXTURE_BUILDER, builder_args=("name",)) + + pipeline = PipelineBuilder( + TEST_CLIENT, + pipeline_name, + sql=_build_sql(loc, extra_config={"version": 2, "end_version": 6}), + runtime_config=RuntimeConfig( + workers=FELDERA_TEST_NUM_WORKERS, + hosts=FELDERA_TEST_NUM_HOSTS, + logging="debug", + ), + ).create_or_replace() + pipeline.start() + pipeline.wait_for_completion(force_stop=False, timeout_s=600) + + rows = _snapshot_rows(pipeline) + assert rows == _MID_HISTORY_EXPECTED_ROWS, ( + "snapshot at v2 must resolve the pre-rename rows 1 and 2 through " + "column mapping (full_name populated, country NULL), then follow " + f"v3..v6 for rows 3, 4, 5; got {rows}" + ) + + pipeline.stop(force=True) + finally: + loc.cleanup() + + +# Pure follow (no snapshot) from v2: read only the commits after v2 as raw +# Parquet. Rows 1 and 2 never appear (no snapshot, and their insert predates v2); +# v3..v6 give rows 3, 4, 5, all post-rename so full_name and country resolve. +_FOLLOW_FROM_VERSION_EXPECTED_ROWS = [ + {"id": 3, "full_name": "carol", "country": "US"}, + {"id": 4, "full_name": "dave", "country": "UK"}, + {"id": 5, "full_name": "erin", "country": "FR"}, +] + + +def test_delta_input_column_mapping_follow_from_version(pipeline_name): + """Pure follow from a non-zero version: replay only the commits after it.""" + loc = DeltaTestLocation.create( + pipeline_name, + mode="follow", + stable_subpath=f"column_mapping_name_{FIXTURE_VERSION}", + ) + try: + ensure_delta_spark_fixture(loc, _FIXTURE_BUILDER, builder_args=("name",)) + + pipeline = PipelineBuilder( + TEST_CLIENT, + pipeline_name, + sql=_build_sql(loc, extra_config={"version": 2, "end_version": 6}), + runtime_config=RuntimeConfig( + workers=FELDERA_TEST_NUM_WORKERS, + hosts=FELDERA_TEST_NUM_HOSTS, + logging="debug", + ), + ).create_or_replace() + pipeline.start() + pipeline.wait_for_completion(force_stop=False, timeout_s=600) + + rows = _snapshot_rows(pipeline) + assert rows == _FOLLOW_FROM_VERSION_EXPECTED_ROWS, ( + "pure follow from v2 must replay only v3..v6 as raw Parquet (no " + "snapshot, so rows 1 and 2 never appear), resolving full_name and " + f"country for rows 3, 4, 5 against each commit's schema; got {rows}" + ) + + pipeline.stop(force=True) + finally: + loc.cleanup() diff --git a/python/tests/utils.py b/python/tests/utils.py index a279a606f0b..cb95be5e540 100644 --- a/python/tests/utils.py +++ b/python/tests/utils.py @@ -310,6 +310,7 @@ def ensure_delta_spark_fixture( *, delta_spark_spec: str = "delta-spark>=4.2,<5", is_present: Callable[[DeltaTestLocation], bool] | None = None, + max_attempts: int = 3, ) -> None: """Ensure a PySpark-authored Delta fixture exists at ``loc`` (cached). @@ -336,6 +337,11 @@ def ensure_delta_spark_fixture( primitives; ``None`` would become the literal string ``"None"``. :param is_present: Predicate deciding whether the fixture already exists; also re-checked after upload to catch partial uploads. + :param max_attempts: How many times to run the builder before giving up. + Spark resolves the delta-spark dependency tree from Maven Central on + this path; that download is prone to transient failures (0-byte + artifacts, half-written Ivy cache files, gateway timeouts) that clear on + a fresh attempt, so retry rather than fail the test on infra flakiness. """ present = ( is_present if is_present is not None else DeltaTestLocation.delta_log_exists @@ -349,25 +355,39 @@ def ensure_delta_spark_fixture( f"(builder runs via `uv run --with {delta_spark_spec}`)." ) - staging = pathlib.Path(tempfile.mkdtemp(prefix="feldera_delta_fixture_")) - try: - subprocess.run( - [ - "uv", - "run", - "--no-project", - "--with", - delta_spark_spec, - "python", - str(builder_script), - str(staging), - *(str(arg) for arg in builder_args), - ], - check=True, - ) - loc._place_tree(staging) - finally: - shutil.rmtree(staging, ignore_errors=True) + for attempt in range(1, max_attempts + 1): + # Fresh staging each attempt: a builder that died mid-write must not + # leak a partial tree into the next attempt or the upload. + staging = pathlib.Path(tempfile.mkdtemp(prefix="feldera_delta_fixture_")) + try: + subprocess.run( + [ + "uv", + "run", + "--no-project", + "--with", + delta_spark_spec, + "python", + str(builder_script), + str(staging), + *(str(arg) for arg in builder_args), + ], + check=True, + ) + loc._place_tree(staging) + break + except subprocess.CalledProcessError: + if attempt == max_attempts: + raise + logging.warning( + "PySpark Delta fixture build failed (attempt %d/%d); retrying. " + "Usually a transient Maven/Ivy download failure.", + attempt, + max_attempts, + ) + time.sleep(5 * attempt) + finally: + shutil.rmtree(staging, ignore_errors=True) if not present(loc): raise RuntimeError( diff --git a/python/uv.lock b/python/uv.lock index 081ef9c8bdc..8e080bc7c0d 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -9,7 +9,7 @@ resolution-markers = [ ] [options] -exclude-newer = "2026-06-26T04:07:02.16602316Z" +exclude-newer = "2026-06-29T10:49:25.023863352Z" exclude-newer-span = "P1W" [[package]] @@ -273,7 +273,7 @@ wheels = [ [[package]] name = "feldera" -version = "0.317.0" +version = "0.318.0" source = { editable = "." } dependencies = [ { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, diff --git a/sql-to-dbsp-compiler/SQL-compiler/pom.xml b/sql-to-dbsp-compiler/SQL-compiler/pom.xml index 831aa57521f..495b8eeb196 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/pom.xml +++ b/sql-to-dbsp-compiler/SQL-compiler/pom.xml @@ -19,7 +19,7 @@ 1.43.0 2.22.0 3.1.12 - 0.317.0 + 0.318.0 1.28.0 diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/MaxCaseToCountRule.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/MaxCaseToCountRule.java index 93ac1fb682a..693879d6fa1 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/MaxCaseToCountRule.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/MaxCaseToCountRule.java @@ -30,15 +30,34 @@ /** * Rule that converts MAX(CASE ... THEN 1 ELSE 0 END) - * into COUNT(...) FILTER (WHERE ...) - * Rewrite: + * or MAX(CASE ... THEN 1 ELSE NULL END) (equivalently, no ELSE clause) + * into COUNT(...) FILTER (WHERE ...). + * + *

Rewrite for ELSE 0: + * 

  * LogicalAggregate(group=[{0}], [MAX($1)]
  *   LogicalProject([$0], $f1=[CASE(=(...), 1, 0)])
- * into
+ *
+ * becomes + * 
  * LogicalProject([$0], [CASE(>($1, 0), 1, 0)])
  *   LogicalAggregate(group=[{0}], agg#0=[COUNT(*) FILTER $1])
  *     LogicalProject([$0], $f1=[IS TRUE(...)])
- * This works only if there is a GROUP BY.
+ *
+ * + *

Rewrite for ELSE NULL (or no ELSE): + * 

+ * LogicalAggregate(group=[{0}], [MAX($1)]
+ *   LogicalProject([$0], $f1=[CASE(=(...), 1, null)])
+ *
+ * becomes + * 
+ * LogicalProject([$0], [CASE(>($1, 0), 1, null)])
+ *   LogicalAggregate(group=[{0}], agg#0=[COUNT(*) FILTER $1])
+ *     LogicalProject([$0], $f1=[IS TRUE(...)])
+ *
+ * + *

This works only if there is a GROUP BY. */ public class MaxCaseToCountRule extends RelRule> @@ -66,16 +85,17 @@ public void onMatch(RelOptRuleCall call) { RelDataType bigInt = cluster.getTypeFactory().createSqlType(SqlTypeName.BIGINT); bigInt = cluster.getTypeFactory().createTypeWithNullability(bigInt, true); - RexLiteral one = rexBuilder.makeLiteral(BigDecimal.ONE, bigInt); - newProjects.add(one); for (int i = 0; i < aggregate.getGroupCount(); i++) // Preserve the keys in the postProjects postProjects.add(new RexInputRef(i, resultType.getFieldList().get(i).getType())); + final int groupCount = aggregate.getGroupCount(); for (AggregateCall aggregateCall : aggregate.getAggCallList()) { AggregateCall newCall = transform(aggregateCall, project, newProjects, postProjects, bigInt); if (newCall == null) { - postProjects.add(rexBuilder.makeInputRef(aggregateCall.getType(), newAggregates.size())); + // Aggregate output: [group_keys (0..groupCount-1), agg_0, agg_1, ...] + postProjects.add(rexBuilder.makeInputRef( + aggregateCall.getType(), groupCount + newAggregates.size())); newAggregates.add(aggregateCall); } else { newAggregates.add(newCall); @@ -147,11 +167,17 @@ public static Long getLiteralValue(RexNode node) { if (op1Value == null) return null; Long op2Value = getLiteralValue(op2); - if (op2Value == null) + if (op2Value == null && !(op2 instanceof RexLiteral)) return null; + boolean elseIsNull = op2Value == null; - boolean flip; - if (op1Value == 0L && op2Value == 1L) { + final boolean flip; + if (elseIsNull) { + // Only THEN 1 ELSE NULL is supported: MAX = 1 if any row matches, NULL otherwise. + if (op1Value != 1L) + return null; + flip = false; + } else if (op1Value == 0L && op2Value == 1L) { flip = true; } else if (op1Value == 1L && op2Value == 0L) { flip = false; @@ -179,27 +205,26 @@ public static Long getLiteralValue(RexNode node) { false, call.rexList, ImmutableList.of(), newProjects.size() - 1, null, RelCollations.EMPTY, bigInt, call.getName()); - // CASE(WHEN $1 IS NULL THEN NULL WHEN $1 = 0 THEN 0 ELSE 1 END) - RelDataType nullable = - cluster.getTypeFactory().createTypeWithNullability(op1.getType(), call.type.isNullable()); - List postOperands = new ArrayList<>(); - // WHEN $1 IS NULL RexInputRef ref = rexBuilder.makeInputRef(bigInt, postProjects.size()); - RexNode isNull = rexBuilder.makeCall(pos, SqlStdOperatorTable.IS_NULL, ref); - postOperands.add(isNull); - // THEN NULL - RexLiteral nullLit = rexBuilder.makeNullLiteral(nullable); - postOperands.add(nullLit); - // WHEN $1 = 0 - RexLiteral biggerZero = rexBuilder.makeLiteral(BigDecimal.ZERO, bigInt); - postOperands.add(rexBuilder.makeCall(pos, SqlStdOperatorTable.EQUALS, ref, biggerZero)); - // THEN 0 - RexNode zero = rexBuilder.makeLiteral(BigDecimal.ZERO, nullable, true); - postOperands.add(zero); - // ELSE one - RexNode one = rexBuilder.makeLiteral(BigDecimal.ONE, nullable, true); - postOperands.add(one); - final RexNode postCondition = rexBuilder.makeCall(pos, SqlStdOperatorTable.CASE, postOperands); + List postOperands = new ArrayList<>(); + RelDataType nullable = cluster.getTypeFactory().createTypeWithNullability( + op1.getType(), call.type.isNullable()); + RexLiteral zero = rexBuilder.makeLiteral(BigDecimal.ZERO, bigInt); + final RexNode postCondition; + if (elseIsNull) { + // CASE WHEN count > 0 THEN 1 ELSE NULL END + postOperands.add(rexBuilder.makeCall(pos, SqlStdOperatorTable.GREATER_THAN, ref, zero)); + postOperands.add(rexBuilder.makeLiteral(BigDecimal.ONE, nullable, true)); + postOperands.add(rexBuilder.makeNullLiteral(nullable)); + } else { + // CASE WHEN count IS NULL THEN NULL WHEN count = 0 THEN 0 ELSE 1 END + postOperands.add(rexBuilder.makeCall(pos, SqlStdOperatorTable.IS_NULL, ref)); + postOperands.add(rexBuilder.makeNullLiteral(nullable)); + postOperands.add(rexBuilder.makeCall(pos, SqlStdOperatorTable.EQUALS, ref, zero)); + postOperands.add(rexBuilder.makeLiteral(BigDecimal.ZERO, nullable, true)); + postOperands.add(rexBuilder.makeLiteral(BigDecimal.ONE, nullable, true)); + } + postCondition = rexBuilder.makeCall(pos, SqlStdOperatorTable.CASE, postOperands); postProjects.add(postCondition); return result; } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java index cd913406d64..20c025fd6c7 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java @@ -980,13 +980,14 @@ public void issue5927() { ccs.stepWeightOne("", """ g | max ---------"""); - ccs.stepWeightOne("INSERT INTO T VALUES ('a', 1), ('b', 3), ('c', 3), ('c', 5), ('d', 1), ('d', 5);", """ + ccs.stepWeightOne("INSERT INTO T VALUES ('a', 1), ('b', 3), ('c', 3), ('c', 5), ('d', 1), ('d', 5), ('e', NULL);", """ g | max --------- a| 0 b| 1 c| 1 - d| 1"""); + d| 1 + e| 0"""); ccs.visit(this.findLinear(ccs.compiler)); } @@ -1042,6 +1043,81 @@ public void issue5927a() { ccs.visit(this.findLinear(ccs.compiler)); } + @Test + public void issue6590() { + // MAX(CASE WHEN cond THEN 1 ELSE NULL END) — ELSE NULL variant + var ccs = this.getCCS(""" + CREATE TABLE T(g VARCHAR, x INT); + CREATE VIEW V AS SELECT g, MAX(CASE WHEN x > 2 THEN 1 ELSE NULL END) FROM T GROUP BY g;"""); + // Validated on Postgres: NULL when no row satisfies cond, 1 otherwise. + ccs.stepWeightOne("", """ + g | max + ---------"""); + ccs.stepWeightOne(""" + INSERT INTO T VALUES + ('a', 1), ('b', 3), ('c', 3), ('c', 5), ('d', 1), ('d', 5), ('e', NULL); + """, """ + g | max + --------- + a| NULL + b| 1 + c| 1 + d| 1 + e| NULL"""); + ccs.visit(this.findLinear(ccs.compiler)); + } + + @Test + public void issue6590a() { + // MAX(CASE WHEN cond THEN 1 END) — no ELSE clause (equivalent to ELSE NULL) + var ccs = this.getCCS(""" + CREATE TABLE T(g VARCHAR, x INT); + CREATE VIEW V AS SELECT g, MAX(CASE WHEN x > 2 THEN 1 END) FROM T GROUP BY g;"""); + // Validated on Postgres: NULL when no row satisfies cond, 1 otherwise. + ccs.stepWeightOne("", """ + g | max + ---------"""); + ccs.stepWeightOne(""" + INSERT INTO T VALUES + ('a', 1), ('b', 3), ('c', 3), ('c', 5), ('d', 1), ('d', 5), + ('e', NULL); + """, """ + g | max + --------- + a| NULL + b| 1 + c| 1 + d| 1 + e| NULL"""); + ccs.visit(this.findLinear(ccs.compiler)); + } + + @Test + public void issue6590b() { + // MAX(CASE...) mixed with SUM in the same aggregate. + // Exercises the post-project index path for untransformed aggregate calls. + var ccs = this.getCCS(""" + CREATE TABLE T(g VARCHAR, x INT); + CREATE VIEW V AS SELECT g, MAX(CASE WHEN x > 2 THEN 1 ELSE 0 END), SUM(x) FROM T GROUP BY g;"""); + // Validated on Postgres + ccs.stepWeightOne("", """ + g | max | sum + --------------"""); + ccs.stepWeightOne(""" + INSERT INTO T VALUES + ('a', 1), ('b', 3), ('c', 3), ('c', 5), ('d', 1), ('d', 5), + ('e', NULL); + """, """ + g | max | sum + -------------- + a| 0| 1 + b| 1| 3 + c| 1| 8 + d| 1| 6 + e| 0| NULL"""); + ccs.visit(this.findLinear(ccs.compiler)); + } + @Test public void calciteIssue7501() { this.getCC("""