NoahKusaba commented on PR #2638: URL: https://github.com/apache/datafusion-ballista/pull/2638#issuecomment-6882333592
Here's the battery, if you'd like to rerun it. It runs 81 queries over Iceberg tables with standalone Ballista and with plain DataFusion, under both planners, with the default task sizing and with 1 and 2 partitions per task, and reports each query whose rows differ. To run it, save it as `integrations/iceberg/tests/datafusion_parity.rs`, then: ```shell cd integrations/iceberg cargo test --test datafusion_parity -- --nocapture ``` It takes about 5 minutes in a debug build, 90 s of which is the three hung joins waiting out their 30 s timeout. Results on 50a7dac7: | Planner | Partitions per task | Match DataFusion | Differences | |---|---|---|---| | static | default | 81/81 | | | static | 1 | 81/81 | | | static | 2 | 81/81 | | | adaptive | default | 80/81 | `join-full-empty` hangs (#2865) | | adaptive | 1 | 71/81 | `join-full-empty` hangs (#2865); 7 unions fail (#2866); `union-all-order` and `union-global-aggs` return extra rows (#2854) | | adaptive | 2 | 80/81 | `join-full-empty` hangs (#2865) | I'd like to add it as a test in a follow-up PR with the DataFusion 56 bump, failing on any difference other than the known ones. <details> <summary><code>datafusion_parity.rs</code></summary> ```rust // Licensed to the Apache Software Foundation (ASF) under one // or more contributor license agreements. See the NOTICE file // distributed with this work for additional information // regarding copyright ownership. The ASF licenses this file // to you under the Apache License, Version 2.0 (the // "License"); you may not use this file except in compliance // with the License. You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, // software distributed under the License is distributed on an // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. //! Runs a battery of queries over Iceberg tables with standalone Ballista and //! with plain DataFusion, and reports every query whose rows differ. //! //! Each query runs under both planners, with the default task sizing and with //! every stage split into tasks of one and of two partitions. The report lists //! each mismatch, error or timeout; it doesn't fail on them. //! //! ```text //! cd integrations/iceberg //! cargo test --test datafusion_parity -- --nocapture //! ``` use std::collections::HashMap; use std::path::Path; use std::sync::Arc; use std::time::Duration; use ballista::prelude::{SessionConfigExt, SessionContextExt}; use ballista_core::config::{ BALLISTA_ADAPTIVE_PLANNER_ENABLED, BALLISTA_SCHEDULER_MAX_PARTITIONS_PER_TASK, }; use ballista_iceberg::register_iceberg; use datafusion::arrow::util::pretty::pretty_format_batches; use datafusion::error::Result; use datafusion::execution::SessionStateBuilder; use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext}; use iceberg::io::LocalFsStorageFactory; use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder}; use iceberg::spec::{NestedField, PartitionSpec, PrimitiveType, Schema, Transform, Type}; use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation, TableIdent}; use iceberg_datafusion::{IcebergCatalogProvider, IcebergStaticTableProvider}; /// How long a query may run before it counts as hung. const TIMEOUT: Duration = Duration::from_secs(30); /// Each configuration: whether the adaptive planner is on, and the most /// partitions a task may run, or `None` for the default. const CONFIGS: [(bool, Option<usize>); 6] = [ (false, None), (false, Some(1)), (false, Some(2)), (true, None), (true, Some(1)), (true, Some(2)), ]; /// `(name, sql, ordered)`. `{o}` is `orders` (1000 rows written in three /// commits), `{op}` the same rows partitioned by `status`, `{c}` `customers`, /// `{e}` an empty table, `o_v1` `orders` as of its first commit, and `pq` a /// Parquet table. Rows are compared in order when `ordered`, sorted otherwise. #[rustfmt::skip] const QUERIES: &[(&str, &str, bool)] = &[ // Scans and filters ("scan", "SELECT * FROM {o}", false), ("filter-lt", "SELECT id, amount FROM {o} WHERE id < 10", false), ("filter-eq-string", "SELECT id FROM {o} WHERE status = 'b'", false), ("filter-null-string", "SELECT id FROM {o} WHERE status IS NULL", false), ("filter-null-long", "SELECT id FROM {o} WHERE cust IS NULL", false), ("filter-between", "SELECT * FROM {o} WHERE id BETWEEN 100 AND 120", false), ("filter-in-and", "SELECT id, qty FROM {o} WHERE status IN ('a', 'c') AND qty > 3", false), ("filter-not-in", "SELECT count(*) FROM {o} WHERE status NOT IN ('a', 'b')", false), ("filter-like", "SELECT * FROM {c} WHERE name LIKE 'cust_1%'", false), ("filter-starts-with", "SELECT id FROM {c} WHERE starts_with(name, 'cust_9')", false), ("filter-date", "SELECT id, d FROM {o} WHERE d >= DATE '2024-01-25'", false), ("filter-bool", "SELECT id FROM {o} WHERE flag", false), ("filter-or", "SELECT id FROM {o} WHERE NOT flag OR amount > 110", false), ("filter-expr", "SELECT id FROM {o} WHERE id % 7 = 0", false), ("filter-double", "SELECT id FROM {o} WHERE amount > 100.5 AND amount < 105", false), ("filter-cast", "SELECT id FROM {o} WHERE CAST(qty AS BIGINT) = 8 AND id < 100", false), ("filter-no-rows", "SELECT * FROM {o} WHERE id = 999999", false), ("filter-in-list", "SELECT count(*) FROM {o} WHERE id IN (1, 5, 9, 200, 450, 999, 1000)", false), // Limits and ordering ("limit", "SELECT count(*) FROM (SELECT * FROM {o} LIMIT 5)", false), ("order-limit", "SELECT id FROM {o} ORDER BY id LIMIT 7", true), ("order-limit-offset", "SELECT id, amount FROM {o} ORDER BY amount DESC, id LIMIT 10 OFFSET 5", true), ("filter-order-limit", "SELECT id FROM {o} WHERE status = 'a' ORDER BY id DESC LIMIT 3", true), ("order-nulls-first", "SELECT cust, id FROM {o} WHERE id < 40 ORDER BY cust NULLS FIRST, id", true), ("order", "SELECT * FROM {o} ORDER BY id", true), // Aggregates ("agg-global", "SELECT count(*), count(cust), sum(qty), round(sum(amount), 2), min(d), max(d), min(status), max(status) FROM {o}", false), ("agg-no-rows", "SELECT count(*), sum(qty) FROM {o} WHERE id < 0", false), ("agg-group", "SELECT status, count(*), sum(qty) FROM {o} GROUP BY status", false), ("agg-having", "SELECT status, flag, count(*) FROM {o} GROUP BY status, flag HAVING count(*) > 60", false), ("agg-distinct", "SELECT count(DISTINCT cust), count(DISTINCT status) FROM {o}", false), ("agg-avg-ordered", "SELECT cust, round(avg(amount), 4) FROM {o} GROUP BY cust ORDER BY cust NULLS LAST", true), ("agg-date", "SELECT d, count(*) FROM {o} GROUP BY d ORDER BY d", true), ("agg-grouping-sets", "SELECT status, flag, count(*) FROM {o} GROUP BY GROUPING SETS ((status), (flag), ())", false), ("agg-string-agg", "SELECT region, string_agg(name, ',' ORDER BY id) FROM {c} WHERE id < 30 GROUP BY region", false), ("agg-filter-clause", "SELECT count(*) FILTER (WHERE flag), sum(qty) FILTER (WHERE status = 'c') FROM {o}", false), // Joins ("join-inner", "SELECT c.region, count(*), sum(o.qty) FROM {o} o JOIN {c} c ON o.cust = c.id GROUP BY c.region", false), ("join-left", "SELECT c.id, count(o.id) FROM {c} c LEFT JOIN {o} o ON o.cust = c.id GROUP BY c.id", false), ("join-self", "SELECT count(*) FROM {o} a JOIN {o} b ON a.cust = b.cust AND a.id < b.id", false), ("join-parquet", "SELECT count(*), sum(p.v) FROM {o} o JOIN pq p ON o.id = p.id", false), ("join-full-empty", "SELECT c.id, e.id FROM {c} c FULL OUTER JOIN {e} e ON c.id = e.id WHERE c.id < 5", false), ("join-cross", "SELECT count(*) FROM {c} CROSS JOIN (SELECT * FROM {o} WHERE id < 5) s", false), ("join-semi", "SELECT count(*) FROM {o} WHERE cust IN (SELECT id FROM {c} WHERE region = 'north')", false), ("join-anti", "SELECT count(*) FROM {c} c WHERE NOT EXISTS (SELECT 1 FROM {o} o WHERE o.cust = c.id AND o.qty = 8)", false), ("join-scalar-subquery", "SELECT count(*) FROM {o} WHERE amount > (SELECT avg(amount) FROM {o})", false), ("join-correlated", "SELECT id FROM {c} c WHERE (SELECT count(*) FROM {o} o WHERE o.cust = c.id) > 9", false), ("join-partitioned", "SELECT count(*), sum(a.qty) FROM {o} a JOIN {op} b ON a.id = b.id AND a.status = b.status", false), ("join-three-way", "SELECT c.region, count(*), sum(p.v) FROM {o} o JOIN {c} c ON o.cust = c.id JOIN pq p ON p.id = c.id GROUP BY c.region", false), ("join-derived", "SELECT c.region, s.n FROM {c} c JOIN (SELECT cust, count(*) AS n FROM {o} GROUP BY cust) s ON s.cust = c.id WHERE c.id < 20", false), // Window functions and CTEs ("window-row-number", "SELECT * FROM (SELECT id, status, row_number() OVER (PARTITION BY status ORDER BY id) AS rn FROM {o}) WHERE rn <= 2", false), ("window-running-sum", "SELECT id, sum(qty) OVER (ORDER BY id ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) FROM {o} WHERE id < 20", false), ("window-rank", "SELECT id, region, rank() OVER (PARTITION BY region ORDER BY id DESC) FROM {c}", false), ("window-lag", "SELECT id, lag(amount) OVER (ORDER BY id) FROM {o} WHERE id < 15", false), ("cte", "WITH big AS (SELECT cust, sum(amount) AS s FROM {o} GROUP BY cust) SELECT count(*) FROM big WHERE s > 600", false), ("cte-twice", "WITH a AS (SELECT id, qty FROM {o} WHERE status = 'a') SELECT count(*) FROM a x JOIN a y ON x.id = y.id", false), // Set operations ("union-all", "SELECT id, status FROM {o} WHERE id < 20 UNION ALL SELECT id, name FROM {c} WHERE id < 20", false), ("union-all-order", "SELECT id FROM {o} WHERE id < 10 UNION ALL SELECT id FROM {c} WHERE id < 10 ORDER BY id", false), ("union", "SELECT cust FROM {o} UNION SELECT id FROM {c}", false), ("union-parquet", "SELECT id FROM {o} WHERE id < 5 UNION ALL SELECT id FROM pq WHERE id < 5", false), ("union-count", "SELECT count(*) FROM (SELECT id FROM {o} UNION ALL SELECT id FROM {c})", false), ("union-global-aggs", "SELECT count(*) FROM {o} UNION ALL SELECT count(*) FROM {c}", false), ("union-limit-branch", "SELECT count(*) FROM (SELECT id FROM (SELECT id FROM {o} LIMIT 10) UNION ALL SELECT id FROM {c})", false), ("union-same-table", "SELECT id, status FROM {o} UNION ALL SELECT id, status FROM {o}", false), ("union-group-by", "SELECT status, count(*) FROM (SELECT status FROM {o} UNION ALL SELECT status FROM {op}) GROUP BY status", false), ("intersect", "SELECT cust FROM {o} WHERE qty = 1 INTERSECT SELECT id FROM {c} WHERE region = 'east'", false), ("except", "SELECT id FROM {c} EXCEPT SELECT cust FROM {o} WHERE qty > 2", false), // Partitioned table ("partitioned-scan", "SELECT * FROM {op}", false), ("partitioned-prune", "SELECT id FROM {op} WHERE status = 'b'", false), ("partitioned-group", "SELECT status, count(*), sum(amount) FROM {op} GROUP BY status", false), ("partitioned-null", "SELECT count(*) FROM {op} WHERE status IS NULL", false), ("partitioned-column-only", "SELECT status FROM {op}", false), ("partitioned-mixed-filter", "SELECT count(*) FROM {op} WHERE status IN ('c', 'd') AND qty < 4", false), // Time travel ("time-travel", "SELECT count(*), max(id) FROM o_v1", false), ("time-travel-join", "SELECT count(*) FROM o_v1 a JOIN {o} b ON a.id = b.id", false), ("time-travel-except", "SELECT count(*) FROM (SELECT id FROM {o} EXCEPT SELECT id FROM o_v1)", false), // Empty table ("empty-scan", "SELECT * FROM {e}", false), ("empty-count", "SELECT count(*) FROM {e}", false), ("empty-join", "SELECT count(*) FROM {c} c JOIN {e} e ON c.id = e.id", false), // Expressions ("distinct", "SELECT DISTINCT status FROM {o}", false), ("case", "SELECT id, CASE WHEN qty > 5 THEN 'hi' WHEN qty > 2 THEN 'mid' ELSE 'lo' END FROM {o} WHERE id < 30", false), ("string-functions", "SELECT upper(name), length(name), substr(region, 1, 2) FROM {c} WHERE id < 25", false), ("date-functions", "SELECT date_part('day', d) AS day, count(*) FROM {o} GROUP BY day", false), ("coalesce", "SELECT coalesce(status, 'none'), count(*) FROM {o} GROUP BY 1", false), ]; #[tokio::test(flavor = "multi_thread")] async fn report_queries_whose_rows_differ_from_datafusion() -> Result<()> { let dir = tempfile::tempdir()?; let (catalog, first_snapshot) = create_tables(dir.path()).await; let queries: Vec<(&str, String, bool)> = QUERIES .iter() .map(|&(name, sql, ordered)| (name, expand(sql), ordered)) .collect(); let datafusion = SessionContext::new(); register(&datafusion, &catalog, first_snapshot, dir.path()).await; let mut expected = Vec::new(); for (name, sql, ordered) in &queries { let rows = rows(&datafusion, sql, *ordered).await; assert!(rows.is_ok(), "DataFusion failed {name}: {rows:?}"); expected.push(rows); } let mut differences = 0; for (adaptive, max_partitions) in CONFIGS { let mut config = SessionConfig::new_with_ballista() .set_str(BALLISTA_ADAPTIVE_PLANNER_ENABLED, &adaptive.to_string()); if let Some(max) = max_partitions { config = config .set_str(BALLISTA_SCHEDULER_MAX_PARTITIONS_PER_TASK, &max.to_string()); } let state = SessionStateBuilder::new() .with_config(config) .with_default_features() .build(); let ballista = SessionContext::standalone_with_state(register_iceberg(state)).await?; register(&ballista, &catalog, first_snapshot, dir.path()).await; let planner = if adaptive { "adaptive" } else { "static" }; let tasks = max_partitions.map_or("default".to_string(), |max| max.to_string()); let mut matched = 0; for ((name, sql, ordered), expected) in queries.iter().zip(&expected) { let verdict = match (rows(&ballista, sql, *ordered).await, expected) { (Ok(got), Ok(expected)) if &got == expected => { matched += 1; continue; } (Ok(got), Ok(expected)) => format!( "wrong rows: {} lines, DataFusion {}", got.len(), expected.len() ), (Err(e), _) => e, (Ok(_), Err(_)) => unreachable!("checked above"), }; differences += 1; println!( "{planner:<8} max_partitions_per_task={tasks:<7} {name:<24} {verdict}" ); } println!( "{planner:<8} max_partitions_per_task={tasks:<7} {matched}/{} match", queries.len() ); } println!( "{differences} of {} runs differ from DataFusion", queries.len() * CONFIGS.len() ); Ok(()) } /// The query's rows as printed lines, sorted unless `ordered`, or why it /// didn't finish. async fn rows( ctx: &SessionContext, sql: &str, ordered: bool, ) -> std::result::Result<Vec<String>, String> { let batches = tokio::time::timeout(TIMEOUT, async { ctx.sql(sql).await?.collect().await }) .await .map_err(|_| format!("no result after {}s", TIMEOUT.as_secs()))? .map_err(|e| { // Ballista prefixes the cause with the job's id. let e = e.to_string(); let cause = e.split_once("DataFusion error: ").map_or(&*e, |(_, c)| c); cause.chars().take(160).collect::<String>() })?; let printed = pretty_format_batches(&batches) .map_err(|e| e.to_string())? .to_string(); let mut lines: Vec<String> = printed.lines().map(str::to_string).collect(); if !ordered { lines.sort(); } Ok(lines) } fn expand(sql: &str) -> String { sql.replace("{op}", "iceberg.ns.orders_p") .replace("{o}", "iceberg.ns.orders") .replace("{c}", "iceberg.ns.customers") .replace("{e}", "iceberg.ns.empty") } /// Rows of `orders` for ids in a range: some `cust` and `status` are null. const ORDERS: &str = "SELECT value AS id, \ CASE WHEN value % 17 = 0 THEN NULL ELSE value % 100 END AS cust, \ CAST(value % 97 AS DOUBLE) * 1.25 AS amount, \ CAST(value % 9 AS INT) AS qty, \ CASE WHEN value % 23 = 0 THEN NULL ELSE chr(CAST(97 + value % 5 AS INT)) END AS status, \ CAST(CAST(19723 + value % 30 AS INT) AS DATE) AS d, \ value % 3 = 0 AS flag"; /// Creates the tables in a memory catalog in `dir`, and a Parquet table in /// `dir/pq`. Returns the catalog and the first snapshot of `orders`. async fn create_tables(dir: &Path) -> (Arc<dyn Catalog>, i64) { let catalog: Arc<dyn Catalog> = Arc::new( MemoryCatalogBuilder::default() .with_storage_factory(Arc::new(LocalFsStorageFactory)) .load( "memory", HashMap::from([( MEMORY_CATALOG_WAREHOUSE.to_string(), dir.join("warehouse").display().to_string(), )]), ) .await .unwrap(), ); let namespace = NamespaceIdent::new("ns".to_string()); catalog .create_namespace(&namespace, HashMap::new()) .await .unwrap(); let orders = schema(&[ ("id", PrimitiveType::Long), ("cust", PrimitiveType::Long), ("amount", PrimitiveType::Double), ("qty", PrimitiveType::Int), ("status", PrimitiveType::String), ("d", PrimitiveType::Date), ("flag", PrimitiveType::Boolean), ]); let customers = schema(&[ ("id", PrimitiveType::Long), ("name", PrimitiveType::String), ("region", PrimitiveType::String), ]); let by_status = PartitionSpec::builder(orders.clone()) .add_partition_field("status", "status", Transform::Identity) .unwrap() .build() .unwrap(); for (name, schema, spec) in [ ("orders", orders.clone(), None), ("orders_p", orders, Some(by_status)), ("customers", customers.clone(), None), ("empty", customers, None), ] { let mut creation = TableCreation::builder() .name(name.to_string()) .schema(schema) .build(); creation.partition_spec = spec.map(Into::into); catalog.create_table(&namespace, creation).await.unwrap(); } let writer = SessionContext::new(); writer.register_catalog( "iceberg", Arc::new( IcebergCatalogProvider::try_new(catalog.clone()) .await .unwrap(), ), ); let run = async |sql: String| { writer.sql(&sql).await.unwrap().collect().await.unwrap(); }; run(format!( "INSERT INTO iceberg.ns.orders {ORDERS} FROM range(0, 400)" )) .await; let first_snapshot = catalog .load_table(&TableIdent::from_strs(["ns", "orders"]).unwrap()) .await .unwrap() .metadata() .current_snapshot_id() .unwrap(); run(format!( "INSERT INTO iceberg.ns.orders {ORDERS} FROM range(400, 800)" )) .await; run(format!( "INSERT INTO iceberg.ns.orders {ORDERS} FROM range(800, 1000)" )) .await; run("INSERT INTO iceberg.ns.orders_p SELECT * FROM iceberg.ns.orders".to_string()) .await; run("INSERT INTO iceberg.ns.customers SELECT value AS id, \ concat('cust_', CAST(value AS VARCHAR)) AS name, \ CASE WHEN value % 11 = 0 THEN NULL \ ELSE CASE value % 4 WHEN 0 THEN 'north' WHEN 1 THEN 'south' \ WHEN 2 THEN 'east' ELSE 'west' END END AS region \ FROM range(0, 100)" .to_string()) .await; run(format!( "COPY (SELECT value AS id, value * 2 AS v FROM range(0, 50)) \ TO '{}/' STORED AS PARQUET", dir.join("pq").display() )) .await; (catalog, first_snapshot) } /// A schema whose first field is a required `id` and the rest optional. fn schema(fields: &[(&str, PrimitiveType)]) -> Schema { let fields = fields.iter().enumerate().map(|(i, (name, ty))| { let id = i as i32 + 1; let ty = Type::Primitive(ty.clone()); if i == 0 { NestedField::required(id, *name, ty).into() } else { NestedField::optional(id, *name, ty).into() } }); Schema::builder().with_fields(fields).build().unwrap() } /// Mounts `catalog` as `iceberg`, and registers `orders` as of /// `first_snapshot` as `o_v1` and the Parquet table as `pq`. async fn register( ctx: &SessionContext, catalog: &Arc<dyn Catalog>, first_snapshot: i64, dir: &Path, ) { ctx.register_catalog( "iceberg", Arc::new( IcebergCatalogProvider::try_new(catalog.clone()) .await .unwrap(), ), ); let orders = catalog .load_table(&TableIdent::from_strs(["ns", "orders"]).unwrap()) .await .unwrap(); let v1 = IcebergStaticTableProvider::try_new_from_table_snapshot(orders, first_snapshot) .await .unwrap(); ctx.register_table("o_v1", Arc::new(v1)).unwrap(); ctx.register_parquet( "pq", format!("{}/", dir.join("pq").display()), ParquetReadOptions::default(), ) .await .unwrap(); } ``` </details> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
