You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "Copilot (via GitHub)" <gi...@apache.org> on 2026/04/13 04:37:36 UTC
Re: [PR] perf: improve datafusion integration [hudi-rs]
Copilot commented on code in PR #548:
URL: https://github.com/apache/hudi-rs/pull/548#discussion_r3070896041
##########
crates/core/src/table/mod.rs:
##########
@@ -812,6 +841,74 @@ impl Table {
Ok(Box::pin(combined_stream))
}
+
+ /// Compute estimated table-level statistics from the metadata table for scan planning.
+ ///
+ /// Returns `(estimated_num_rows, estimated_total_byte_size)` where byte size is the
+ /// estimated uncompressed in-memory size.
+ ///
+ /// The approach:
+ /// 1. Read MDT files partition to get all active base files with on-disk sizes
+ /// 2. Read ONE Parquet footer from a sampled file to derive compression ratio
+ /// 3. Infer total rows and uncompressed byte size for all base files
+ ///
+ /// Only base files are counted (log files are excluded).
+ ///
+ /// TODO: support including log files in the estimation for MOR tables for scan planning.
+ ///
+ /// Returns `None` if metadata table is not enabled or if statistics cannot be computed.
+ pub async fn compute_table_stats(&self) -> Option<(usize, usize)> {
+ if !self.is_metadata_table_enabled() {
+ return None;
+ }
+
+ // Step 1: Get MDT files partition records (cached on the MDT instance).
+ let partition_schema = self.get_partition_schema().await.ok()?;
+ let partition_pruner =
+ PartitionPruner::new(&[], &partition_schema, self.hudi_configs.as_ref()).ok()?;
+ let mdt = self.get_or_init_metadata_table().await.ok()?;
+ let records = mdt
+ .fetch_files_partition_records(&partition_pruner)
+ .await
+ .ok()?;
+
+ // Only count base files (exclude log files which start with '.').
+ let base_file_format: String = self
+ .hudi_configs
+ .get_or_default(HudiTableConfig::BaseFileFormat)
+ .into();
+ let base_file_suffix = format!(".{base_file_format}");
+ let total_on_disk_size: u64 = records
+ .values()
+ .filter(|r| !r.is_all_partitions())
+ .flat_map(|r| r.active_files_with_sizes())
+ .filter(|(name, _)| name.ends_with(&base_file_suffix))
+ .map(|(_, size)| size)
+ .sum();
+ let sample_file_path = FileSystemView::find_sample_parquet_path_from_records(&records);
+
+ if total_on_disk_size == 0 {
+ return Some((0, 0));
+ }
+
+ // Step 2: Use the cached estimator (or init it now).
+ // For non-Parquet tables (no sample file found), fall back to on-disk size.
+ match self
+ .file_system_view
+ .get_or_init_estimator(sample_file_path.as_deref())
+ .await
+ {
+ Some(estimator) => {
+ let (estimated_total_byte_size, estimated_total_rows) =
+ estimator.estimate(total_on_disk_size);
+ Some((
+ estimated_total_rows as usize,
+ estimated_total_byte_size as usize,
+ ))
+ }
+ None => Some((0, total_on_disk_size as usize)),
+ }
Review Comment:
`compute_table_stats()` returns `Some((0, total_on_disk_size))` when it can't initialize the Parquet-based estimator (e.g., non-Parquet base file format or footer read failure). Returning an inexact `num_rows = 0` can mislead DataFusion join planning (0 rows implies empty). Consider returning `None` in this branch, or changing the API to represent unknown row count (e.g., `Precision::Absent` for `num_rows`) while still optionally providing `total_byte_size`.
##########
benchmark/tpch/run.sh:
##########
@@ -62,23 +55,29 @@ setup_spark_native() {
cp "$SCRIPT_DIR/infra/spark/log4j2.properties" "$SPARK_HOME/conf/log4j2.properties"
}
+is_cloud_url() {
+ case "$1" in
+ s3://*|s3a://*|gs://*|wasb://*|wasbs://*|az://*) return 0 ;;
+ *) return 1 ;;
+ esac
+}
+
usage() {
cat <<EOF
Usage: $0 <command> [options]
Commands:
generate Generate TPC-H parquet data
- create-tables Create Hudi COW tables from parquet via Spark (Docker)
+ create-tables Create Hudi COW tables from parquet via Spark SQL
bench-spark Run TPC-H queries against Hudi tables via Spark SQL
bench-datafusion Run TPC-H queries against Hudi tables via DataFusion
compare Compare persisted benchmark results with bar charts
-Environment:
- MODE Execution mode: docker (default) or native
-
Options:
--scale-factor N TPC-H scale factor (default: $DEFAULT_SCALE_FACTOR)
--format F Table format: hudi or parquet (default: auto)
+ --hudi-dir D Hudi data directory or cloud URL (default: data/sf{N}-hudi)
+ --parquet-dir D Parquet data directory or cloud URL (default: data/sf{N}-parquet)
--queries Q Comma-separated query numbers (default: all 22)
Review Comment:
The usage text advertises `--hudi-dir` / `--parquet-dir` options, but `create-tables` currently only parses `--scale-factor` and will error on these flags. Either update `cmd_create_tables` to accept these options (and handle cloud URLs appropriately) or scope the help text to only the commands that support them.
##########
benchmark/tpch/run.sh:
##########
@@ -138,32 +137,23 @@ cmd_create_tables() {
fi
build_tpch
- build_docker
+ setup_spark
mkdir -p "$hudi_dir"
local sql_file
sql_file="$(mktemp)"
"$TPCH_BIN" render-ctas --scale-factor "$sf" \
- --parquet-base /opt/parquet --hudi-base /opt/hudi > "$sql_file"
+ --parquet-base "$parquet_dir" --hudi-base "$hudi_dir" > "$sql_file"
read_spark_args --scale-factor "$sf" --command create-tables
echo "Creating Hudi COW tables from parquet (sf$sf)..."
- local docker_exit=0
- docker run --rm \
- -v "$parquet_dir:/opt/parquet:ro" \
- -v "$hudi_dir:/opt/hudi" \
- -v "$sql_file:/opt/spark/work-dir/create_hudi_tables.sql:ro" \
- "$DOCKER_IMAGE" \
- /opt/spark/bin/spark-sql "${SPARK_ARGS[@]}" \
- -f /opt/spark/work-dir/create_hudi_tables.sql \
- || docker_exit=$?
+ "$SPARK_HOME/bin/spark-sql" \
+ --packages org.apache.hudi:hudi-spark3.5-bundle_2.12:1.1.1 \
+ "${SPARK_ARGS[@]}" \
+ -f "$sql_file"
Review Comment:
Because the script runs with `set -e`, a `spark-sql` failure will exit before `rm -f "$sql_file"`, leaving temp files behind. Consider adding a `trap` to always clean up the generated SQL file (and possibly any temp dirs) on exit/error.
##########
crates/datafusion/src/lib.rs:
##########
@@ -142,10 +146,30 @@ impl HudiDataSource {
}
};
+ // Compute table-level statistics for join ordering and broadcast decisions.
+ // Uses MDT files partition + latest commit stats + 1 Parquet footer to infer
+ // row counts and byte sizes without loading all file groups.
+ // Falls back to None if metadata table is not enabled.
+ let cached_stats = match table.compute_table_stats().await {
+ Some((num_rows, total_byte_size)) => {
+ let num_fields = schema.fields().len();
+ Some(Statistics {
+ num_rows: Precision::Inexact(num_rows),
Review Comment:
`compute_table_stats()` may return `(0, total_byte_size)` for non-Parquet tables / estimator failures, but this branch still publishes `num_rows: Precision::Inexact(0)`. That can cause DataFusion to treat a non-empty table as empty during join selection/costing. It would be safer to avoid returning table statistics when row count is unknown, or publish `num_rows` as absent/unknown while keeping `total_byte_size` inexact.
```suggestion
let num_rows = if num_rows == 0 && total_byte_size > 0 {
Precision::Absent
} else {
Precision::Inexact(num_rows)
};
Some(Statistics {
num_rows,
```
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org