You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "zhuqi-lucas (via GitHub)" <gi...@apache.org> on 2025/12/02 08:09:43 UTC
[PR] Issue 18976 [datafusion]
zhuqi-lucas opened a new pull request, #19042:
URL: https://github.com/apache/datafusion/pull/19042
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
- Closes #.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
## Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#issuecomment-3610842761
> LGTM, thank you!
>
> I tested locally and it's working as expected. I have left several minor advices for cleanup.
Thank you @2010YOUY01 for review, addressed all comments in latest PR!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2594654703
##########
benchmarks/bench.sh:
##########
@@ -1197,10 +1206,105 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
Review Comment:
```rust
Resources exhausted: Additional allocation failed for ExternalSorterMerge[1] with top memory consumers (across reservations) as:
ExternalSorter[2]#13(can spill: true) consumed 3.7 GB, peak 4.8 GB,
ExternalSorter[3]#15(can spill: true) consumed 3.5 GB, peak 4.4 GB,
ExternalSorterMerge[2]#14(can spill: false) consumed 2.3 GB, peak 2.3 GB,
ExternalSorterMerge[1]#12(can spill: false) consumed 1004.2 MB, peak 1694.0 MB,
ExternalSorterMerge[3]#16(can spill: false) consumed 845.7 MB, peak 1798.9 MB.
Error: Failed to allocate additional 12.7 MB for ExternalSorterMerge[1] with 998.9 MB already allocated for this reservation - 689.7 KB remain available for the total pool
\q
```
ExternalSorterMerge seems to cause the Resources exhausted, when we have more than one partition.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2594658232
##########
benchmarks/bench.sh:
##########
@@ -1197,10 +1206,105 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
Review Comment:
Updated, @alamb I add the duration logs in latest PR now for the default behavior (12GB memory, and 1 target partition), the time is fast for it for my local mac, less than 5mins:
```rust
+----------+
| count |
+----------+
| 99997497 |
+----------+
1 row(s) fetched.
Elapsed 278.468 seconds.
\q
End time: 2025-12-06 16:27:54
✓ Successfully created sorted ClickBench dataset
Input: 14095 MB
Output: 36159 MB
Time Statistics:
Total duration: 280 seconds (00:04:40)
Throughput: 50 MB/s
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2599603799
##########
benchmarks/bench.sh:
##########
@@ -1197,10 +1206,105 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
Review Comment:
I filed a ticket to try and make this better
- https://github.com/apache/datafusion/issues/19216
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2581643916
##########
benchmarks/src/clickbench.rs:
##########
@@ -78,6 +78,16 @@ pub struct RunOpt {
/// If present, write results json here
#[structopt(parse(from_os_str), short = "o", long = "output")]
output_path: Option<PathBuf>,
+
+ /// Column name that the data is sorted by (e.g., "EventTime")
+ /// If specified, DataFusion will be informed that the data has this sort order
+ /// using CREATE EXTERNAL TABLE with WITH ORDER clause
+ #[structopt(long = "sorted-by")]
+ sorted_by: Option<String>,
+
+ /// Sort order: ASC or DESC (default: ASC)
+ #[structopt(long = "sort-order", default_value = "ASC")]
Review Comment:
We can improve this as follow-up.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2591927842
##########
benchmarks/bench.sh:
##########
@@ -1197,6 +1206,75 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits_0.parquet (partitioned dataset)
+# The data is sorted by EventTime in ascending order
+# Using hits_0.parquet (~150MB) instead of full hits.parquet (~14GB) for faster testing
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_0_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits_partitioned/hits_0.parquet"
+
+ echo "Creating sorted ClickBench dataset from hits_0.parquet..."
Review Comment:
I did this, but OOM in my local mac, and i tried today with target partiton setting to 1, it works now.
Addressed in latest PR, thanks @alamb !
##########
benchmarks/bench.sh:
##########
@@ -1197,6 +1206,75 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits_0.parquet (partitioned dataset)
+# The data is sorted by EventTime in ascending order
+# Using hits_0.parquet (~150MB) instead of full hits.parquet (~14GB) for faster testing
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_0_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits_partitioned/hits_0.parquet"
+
+ echo "Creating sorted ClickBench dataset from hits_0.parquet..."
Review Comment:
I did this, but OOM in my local mac, and i tried today with target partition setting to 1, it works now.
Addressed in latest PR, thanks @alamb !
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2592056916
##########
benchmarks/bench.sh:
##########
@@ -1197,10 +1206,105 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
Review Comment:
But it works for the huge data set:
```
Running `/Users/zhuqi/arrow-datafusion/target/release/dfbench clickbench --iterations 5 --path /Users/zhuqi/arrow-datafusion/benchmarks/data/hits_sorted.parquet --queries-path /Users/zhuqi/arrow-datafusion/benchmarks/queries/clickbench/queries/sorted_data --sorted-by EventTime -c datafusion.optimizer.prefer_existing_sort=true -o /Users/zhuqi/arrow-datafusion/benchmarks/results/issue_18976/data_sorted_clickbench.json`
Running benchmarks with the following options: RunOpt { query: None, pushdown: false, common: CommonOpt { iterations: 5, partitions: None, batch_size: None, mem_pool_type: "fair", memory_limit: None, sort_spill_reservation_bytes: None, debug: false }, path: "/Users/zhuqi/arrow-datafusion/benchmarks/data/hits_sorted.parquet", queries_path: "/Users/zhuqi/arrow-datafusion/benchmarks/queries/clickbench/queries/sorted_data", output_path: Some("/Users/zhuqi/arrow-datafusion/benchmarks/results/issue_18976/data_sorted_clickbench.json"), sorted_by: Some("EventTime"), sort_order: "ASC", config_options: ["datafusion.optimizer.prefer_existing_sort=true"] }
ℹ️ Data is registered with sort order
Setting config: datafusion.optimizer.prefer_existing_sort = true
Registering table with sort order: EventTime ASC
Executing: CREATE EXTERNAL TABLE hits STORED AS PARQUET LOCATION '/Users/zhuqi/arrow-datafusion/benchmarks/data/hits_sorted.parquet' WITH ORDER ("EventTime" ASC)
Q0: -- Must set for ClickBench hits_partitioned dataset. See https://github.com/apache/datafusion/issues/16591
-- set datafusion.execution.parquet.binary_as_string = true
SELECT * FROM hits ORDER BY "EventTime" DESC limit 10;
Query 0 iteration 0 took 2388.0 ms and returned 10 rows
Query 0 iteration 1 took 1789.9 ms and returned 10 rows
Query 0 iteration 2 took 1844.1 ms and returned 10 rows
Query 0 iteration 3 took 1816.4 ms and returned 10 rows
Query 0 iteration 4 took 1808.9 ms and returned 10 rows
Query 0 avg time: 1929.46 ms
+ set +x
Done
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2594623309
##########
benchmarks/bench.sh:
##########
@@ -1201,6 +1210,101 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
+
+-- Parquet output configuration
+SET datafusion.execution.parquet.max_row_group_size = 65536;
+SET datafusion.execution.parquet.compression = 'uncompressed';
+
+-- Execute sort and write
+COPY (SELECT * FROM '${ORIGINAL_FILE}' ORDER BY "EventTime")
+TO '${SORTED_FILE}'
+STORED AS PARQUET;
+EOF
+
+ local result=$?
+
+ if [ $result -eq 0 ]; then
+ echo "✓ Successfully created sorted ClickBench dataset"
+
+ INPUT_SIZE=$(stat -f%z "${ORIGINAL_FILE}" 2>/dev/null || stat -c%s "${ORIGINAL_FILE}" 2>/dev/null)
+ OUTPUT_SIZE=$(stat -f%z "${SORTED_FILE}" 2>/dev/null || stat -c%s "${SORTED_FILE}" 2>/dev/null)
+ INPUT_MB=$((INPUT_SIZE / 1024 / 1024))
+ OUTPUT_MB=$((OUTPUT_SIZE / 1024 / 1024))
+
+ echo " Input: ${INPUT_MB} MB"
Review Comment:
Yes @alamb , it was suggested by @2010YOUY01 here: https://github.com/apache/datafusion/pull/19042#discussion_r2587928607
Because we want to speed up the sort for data generation, and we don't care about the compression here, so i set to uncompressed to speed up it.
```rust
SET datafusion.execution.parquet.compression = 'uncompressed';
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2581505359
##########
benchmarks/src/clickbench.rs:
##########
@@ -136,10 +158,24 @@ impl RunOpt {
parquet_options.pushdown_filters = true;
parquet_options.reorder_filters = true;
}
+
+ if self.sorted_by.is_some() {
Review Comment:
Yes, it should be separate .
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2587988139
##########
benchmarks/src/clickbench.rs:
##########
@@ -78,6 +78,19 @@ pub struct RunOpt {
/// If present, write results json here
#[structopt(parse(from_os_str), short = "o", long = "output")]
output_path: Option<PathBuf>,
+
+ /// Column name that the data is sorted by (e.g., "EventTime")
+ /// If specified, DataFusion will be informed that the data has this sort order
+ /// using CREATE EXTERNAL TABLE with WITH ORDER clause
+ ///
+ /// This is for `sorted_clickbench` benchmark's internal usage. If you want to specify
Review Comment:
The comment mentions "sorted_clickbench" but the shell function is actually named "data_sorted_clickbench". This inconsistency could confuse users. Consider updating the comment to match the actual function name.
```suggestion
/// This is for `data_sorted_clickbench` benchmark's internal usage. If you want to specify
```
##########
benchmarks/src/clickbench.rs:
##########
@@ -78,6 +78,19 @@ pub struct RunOpt {
/// If present, write results json here
#[structopt(parse(from_os_str), short = "o", long = "output")]
output_path: Option<PathBuf>,
+
+ /// Column name that the data is sorted by (e.g., "EventTime")
+ /// If specified, DataFusion will be informed that the data has this sort order
+ /// using CREATE EXTERNAL TABLE with WITH ORDER clause
+ ///
+ /// This is for `sorted_clickbench` benchmark's internal usage. If you want to specify
+ /// a different order, make sure the dataset is generated with the same sort order.
+ #[structopt(long = "sorted-by")]
+ sorted_by: Option<String>,
+
+ /// Sort order: ASC or DESC (default: ASC)
+ #[structopt(long = "sort-order", default_value = "ASC")]
+ sort_order: String,
Review Comment:
The `sort_order` parameter should be validated to ensure it's either "ASC" or "DESC". Currently, any value can be passed which could lead to invalid SQL generation. Consider adding validation or using an enum type in structopt.
##########
benchmarks/src/clickbench.rs:
##########
@@ -214,17 +253,54 @@ impl RunOpt {
}
/// Registers the `hits.parquet` as a table named `hits`
+ /// If sorted_by is specified, uses CREATE EXTERNAL TABLE with WITH ORDER
async fn register_hits(&self, ctx: &SessionContext) -> Result<()> {
- let options = Default::default();
let path = self.path.as_os_str().to_str().unwrap();
- ctx.register_parquet("hits", path, options)
- .await
- .map_err(|e| {
- DataFusionError::Context(
- format!("Registering 'hits' as {path}"),
- Box::new(e),
- )
- })
+
+ // If sorted_by is specified, use CREATE EXTERNAL TABLE with WITH ORDER
+ if let Some(ref sort_column) = self.sorted_by {
+ println!(
+ "Registering table with sort order: {} {}",
+ sort_column, self.sort_order
+ );
+
+ // Escape column name with double quotes
+ let escaped_column = if sort_column.contains('"') {
+ sort_column.clone()
+ } else {
+ format!("\"{sort_column}\"")
+ };
Review Comment:
The column name escaping logic is incomplete. If a column name already contains double quotes, simply cloning it won't properly escape them. Column names with quotes should have them doubled (e.g., `My"Column` should become `"My""Column"`). Consider using proper SQL identifier escaping.
```suggestion
// Escape column name for SQL identifier: double embedded quotes and wrap in double quotes
let escaped_column = format!("\"{}\"", sort_column.replace('"', "\"\""));
```
##########
benchmarks/src/clickbench.rs:
##########
@@ -214,17 +253,54 @@ impl RunOpt {
}
/// Registers the `hits.parquet` as a table named `hits`
+ /// If sorted_by is specified, uses CREATE EXTERNAL TABLE with WITH ORDER
async fn register_hits(&self, ctx: &SessionContext) -> Result<()> {
- let options = Default::default();
let path = self.path.as_os_str().to_str().unwrap();
- ctx.register_parquet("hits", path, options)
- .await
- .map_err(|e| {
- DataFusionError::Context(
- format!("Registering 'hits' as {path}"),
- Box::new(e),
- )
- })
+
+ // If sorted_by is specified, use CREATE EXTERNAL TABLE with WITH ORDER
+ if let Some(ref sort_column) = self.sorted_by {
+ println!(
+ "Registering table with sort order: {} {}",
+ sort_column, self.sort_order
+ );
+
+ // Escape column name with double quotes
+ let escaped_column = if sort_column.contains('"') {
+ sort_column.clone()
+ } else {
+ format!("\"{sort_column}\"")
+ };
+
+ // Build CREATE EXTERNAL TABLE DDL with WITH ORDER clause
+ // Schema will be automatically inferred from the Parquet file
+ let create_table_sql = format!(
+ "CREATE EXTERNAL TABLE hits \
+ STORED AS PARQUET \
+ LOCATION '{}' \
+ WITH ORDER ({} {})",
+ path,
+ escaped_column,
+ self.sort_order.to_uppercase()
Review Comment:
The `to_uppercase()` call is applied to the user-provided `sort_order` without validation. If an invalid value is provided (e.g., "invalid"), this will generate invalid SQL. This validation should happen earlier, ideally when parsing command-line arguments.
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,254 @@
+#!/usr/bin/env python3
+
+# 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.
+
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 64 * 1024, # 64k rows default
+ compression: str = 'zstd',
+ compression_level: int = 3,
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=compression_level, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print("\n✓ Successfully created sorted file!")
+ print(f" Input: {input_size_mb:.1f} MB")
+ print(f" Output: {output_size_mb:.1f} MB")
+ print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
Review Comment:
Potential division by zero: if `output_size_mb` is 0 (empty or very small file), this will raise a `ZeroDivisionError`. Add a check before the division or handle the case where the output file is empty.
```suggestion
if output_size_mb > 0:
print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
else:
print(" Compression ratio: N/A (output file size is zero)")
```
##########
benchmarks/bench.sh:
##########
@@ -1197,6 +1206,80 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits_0.parquet (partitioned dataset)
+# The data is sorted by EventTime in ascending order
+# Using hits_0.parquet (~150MB) instead of full hits.parquet (~14GB) for faster testing
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_0_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits_partitioned/hits_0.parquet"
+
+ echo "Creating sorted ClickBench dataset from hits_0.parquet..."
+
+ # Check if partitioned data exists
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits_partitioned/hits_0.parquet not found. Running data_clickbench_partitioned first..."
+ data_clickbench_partitioned
+ fi
+
+ # Check if sorted file already exists
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits_0.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits_0.parquet by EventTime (this takes ~10 seconds)..."
+
+ # Ensure virtual environment exists and has pyarrow
+ if [ ! -d "$VIRTUAL_ENV" ]; then
+ echo "Creating virtual environment at $VIRTUAL_ENV..."
+ python3 -m venv "$VIRTUAL_ENV"
+ fi
+
+ # Activate virtual environment
+ source "$VIRTUAL_ENV/bin/activate"
+
+ # Check and install pyarrow if needed
+ if ! python3 -c "import pyarrow" 2>/dev/null; then
+ echo "Installing pyarrow (this may take a minute)..."
+ pip install --quiet pyarrow
+ fi
+
+ # Use the standalone Python script to sort
+ python3 "${SCRIPT_DIR}"/sort_clickbench.py "${ORIGINAL_FILE}" "${SORTED_FILE}"
+ local result=$?
+
+ # Deactivate virtual environment
+ deactivate
Review Comment:
The `deactivate` command may fail or not exist if the virtual environment activation failed. This should be wrapped in a check to prevent script errors. Consider using `if command -v deactivate &> /dev/null; then deactivate; fi` or checking if the activation was successful before deactivating.
```suggestion
if command -v deactivate &> /dev/null; then deactivate; fi
```
##########
benchmarks/bench.sh:
##########
@@ -99,6 +99,9 @@ clickbench_partitioned: ClickBench queries against partitioned (100 files) parqu
clickbench_pushdown: ClickBench queries against partitioned (100 files) parquet w/ filter_pushdown enabled
clickbench_extended: ClickBench \"inspired\" queries against a single parquet (DataFusion specific)
+# Sorted Data Benchmarks (ORDER BY Optimization)
+data_sorted_clickbench: ClickBench queries on pre-sorted data WITH sort order info (tests sort elimination optimization)
+
Review Comment:
The new `data_sorted_clickbench` benchmark is not documented in the README.md file. Users won't know how to use this new feature without documentation. Consider adding a section describing how to generate sorted data and run the sorted data benchmark, similar to how other benchmarks are documented.
```suggestion
# Documentation:
# To use the data_sorted_clickbench benchmark:
# 1. Generate sorted ClickBench data with sort order information:
# Run: ./bench.sh generate_sorted_clickbench_data
# This will create the required sorted parquet files in the data directory.
# 2. Run the benchmark:
# Run: ./bench.sh data_sorted_clickbench
# This will execute the ClickBench queries on the pre-sorted data and test sort elimination optimization.
# For more details, see the README.md file.
```
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,254 @@
+#!/usr/bin/env python3
+
+# 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.
+
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 64 * 1024, # 64k rows default
+ compression: str = 'zstd',
+ compression_level: int = 3,
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=compression_level, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print("\n✓ Successfully created sorted file!")
+ print(f" Input: {input_size_mb:.1f} MB")
+ print(f" Output: {output_size_mb:.1f} MB")
+ print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
+ print("\nRow Group Statistics:")
+ print(f" Total row groups: {num_row_groups}")
+ print(f" Total rows: {len(sorted_table):,}")
+
+ # Show row group details
+ for i in range(min(3, num_row_groups)):
+ rg_metadata = parquet_file.metadata.row_group(i)
+ print(f" Row group {i}: {rg_metadata.num_rows:,} rows, {rg_metadata.total_byte_size / 1024**2:.1f} MB")
+
+ if num_row_groups > 3:
+ print(f" ... and {num_row_groups - 3} more row groups")
+
+ avg_rows_per_group = len(sorted_table) / num_row_groups if num_row_groups > 0 else 0
Review Comment:
The division by zero check `if num_row_groups > 0` is done after the division, making it ineffective. This will cause a ZeroDivisionError if `num_row_groups` is 0. Move the check before the division: `avg_rows_per_group = len(sorted_table) / num_row_groups if num_row_groups > 0 else 0`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2591972014
##########
benchmarks/bench.sh:
##########
@@ -1197,10 +1206,105 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
Review Comment:
@alamb I tried locally for target_partitions, it only not OOM for setting to 1, even for 2 it will OOM, so i setting 1 here. I am not sure why.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2587976106
##########
benchmarks/src/clickbench.rs:
##########
@@ -78,6 +78,16 @@ pub struct RunOpt {
/// If present, write results json here
#[structopt(parse(from_os_str), short = "o", long = "output")]
output_path: Option<PathBuf>,
+
+ /// Column name that the data is sorted by (e.g., "EventTime")
+ /// If specified, DataFusion will be informed that the data has this sort order
+ /// using CREATE EXTERNAL TABLE with WITH ORDER clause
Review Comment:
Good suggestion @2010YOUY01 , addressed.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#issuecomment-3602553504
Thank you @martin-g for review.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#issuecomment-3629364796
> Thanks @zhuqi-lucas -- I re-tested this benchmark and it seems to work great!
Thank you @alamb , i will investigate the follow-up, i am on leave this week.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "martin-g (via GitHub)" <gi...@apache.org>.
martin-g commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2581331565
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 1024 * 1024, # 1M rows default
+ compression: str = 'snappy',
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=None, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print(f"\n✓ Successfully created sorted file!")
+ print(f" Input: {input_size_mb:.1f} MB")
+ print(f" Output: {output_size_mb:.1f} MB")
+ print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
+ print(f"\nRow Group Statistics:")
+ print(f" Total row groups: {num_row_groups}")
+ print(f" Total rows: {len(sorted_table):,}")
+
+ # Show row group details
+ for i in range(min(3, num_row_groups)):
+ rg_metadata = parquet_file.metadata.row_group(i)
+ print(f" Row group {i}: {rg_metadata.num_rows:,} rows, {rg_metadata.total_byte_size / 1024**2:.1f} MB")
+
+ if num_row_groups > 3:
+ print(f" ... and {num_row_groups - 3} more row groups")
+
+ avg_rows_per_group = len(sorted_table) / num_row_groups if num_row_groups > 0 else 0
+ print(f" Average rows per group: {avg_rows_per_group:,.0f}")
+
+ except Exception as e:
+ print(f"Error: {e}")
+ import traceback
+ traceback.print_exc()
+ sys.exit(1)
+
+
+def verify_sorted_file(file_path: Path):
+ """Verify that a parquet file is sorted by EventTime."""
+ try:
+ print(f"Verifying sorted file: {file_path}")
+ parquet_file = pq.ParquetFile(str(file_path))
+
+ num_row_groups = parquet_file.num_row_groups
+ file_size_mb = file_path.stat().st_size / (1024**2)
+
+ print(f" File size: {file_size_mb:.1f} MB")
+ print(f" Row groups: {num_row_groups}")
+
+ # Read first and last row group to verify sort
+ first_rg = parquet_file.read_row_group(0)
+ last_rg = parquet_file.read_row_group(num_row_groups - 1)
+
+ first_time = first_rg.column('EventTime')[0].as_py()
+ last_time = last_rg.column('EventTime')[-1].as_py()
+
+ print(f" First EventTime: {first_time}")
+ print(f" Last EventTime: {last_time}")
+ print(f" Sorted: {'✓ YES' if first_time <= last_time else '✗ NO'}")
+
+ except Exception as e:
+ print(f"Error during verification: {e}")
+
+
+def main():
+ parser = argparse.ArgumentParser(
+ description='Sort ClickBench parquet file by EventTime',
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="""
+Examples:
+ # Basic usage (1M rows per group)
+ %(prog)s input.parquet output.parquet
+
+ # Custom row group size (2M rows)
+ %(prog)s input.parquet output.parquet --row-group-size 2097152
+
+ # Use zstd compression
+ %(prog)s input.parquet output.parquet --compression zstd
+
+ # Verify existing file
+ %(prog)s --verify-only output.parquet
+ """
+ )
+
+ parser.add_argument(
+ 'input',
+ nargs='?',
+ help='Input parquet file path'
+ )
+ parser.add_argument(
+ 'output',
+ nargs='?',
+ help='Output sorted parquet file path'
+ )
+ parser.add_argument(
+ '--row-group-size',
+ type=int,
+ default=64 * 1024, # 64K rows
+ help='Number of rows per row group (default: 65536 = 64K)'
+ )
+ parser.add_argument(
+ '--compression',
+ choices=['snappy', 'gzip', 'brotli', 'lz4', 'zstd', 'none'],
+ default='zstd',
+ help='Compression codec (default: zstd)'
+ )
+ parser.add_argument(
+ '--compression-level',
+ type=int,
+ default=3,
+ help='Compression level (default: 3 for zstd)'
+ )
+ parser.add_argument(
+ '--no-verify',
+ action='store_true',
+ help='Skip sort verification'
+ )
+ parser.add_argument(
+ '--verify-only',
+ action='store_true',
+ help='Only verify an existing sorted file (no sorting)'
+ )
+
+ args = parser.parse_args()
+
+ if args.verify_only:
+ if not args.input:
+ parser.error("--verify-only requires input file")
+ verify_sorted_file(Path(args.input))
+ return
+
+ if not args.input or not args.output:
+ parser.error("input and output paths are required")
+
+ sort_clickbench_data(
+ args.input,
+ args.output,
+ row_group_size=args.row_group_size,
+ compression=args.compression,
+ verify=not args.no_verify
+ )
+
+
+if __name__ == '__main__':
+ if len(sys.argv) == 1:
+ print("Usage: python3 sort_clickbench_enhanced.py <input_file> <output_file>")
Review Comment:
```suggestion
print("Usage: python3 sort_clickbench.py <input_file> <output_file>")
```
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
Review Comment:
```suggestion
```
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 1024 * 1024, # 1M rows default
+ compression: str = 'snappy',
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=None, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print(f"\n✓ Successfully created sorted file!")
+ print(f" Input: {input_size_mb:.1f} MB")
+ print(f" Output: {output_size_mb:.1f} MB")
+ print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
+ print(f"\nRow Group Statistics:")
+ print(f" Total row groups: {num_row_groups}")
+ print(f" Total rows: {len(sorted_table):,}")
+
+ # Show row group details
+ for i in range(min(3, num_row_groups)):
+ rg_metadata = parquet_file.metadata.row_group(i)
+ print(f" Row group {i}: {rg_metadata.num_rows:,} rows, {rg_metadata.total_byte_size / 1024**2:.1f} MB")
+
+ if num_row_groups > 3:
+ print(f" ... and {num_row_groups - 3} more row groups")
+
+ avg_rows_per_group = len(sorted_table) / num_row_groups if num_row_groups > 0 else 0
+ print(f" Average rows per group: {avg_rows_per_group:,.0f}")
+
+ except Exception as e:
+ print(f"Error: {e}")
+ import traceback
+ traceback.print_exc()
+ sys.exit(1)
+
+
+def verify_sorted_file(file_path: Path):
+ """Verify that a parquet file is sorted by EventTime."""
+ try:
+ print(f"Verifying sorted file: {file_path}")
+ parquet_file = pq.ParquetFile(str(file_path))
+
+ num_row_groups = parquet_file.num_row_groups
+ file_size_mb = file_path.stat().st_size / (1024**2)
+
+ print(f" File size: {file_size_mb:.1f} MB")
+ print(f" Row groups: {num_row_groups}")
+
+ # Read first and last row group to verify sort
+ first_rg = parquet_file.read_row_group(0)
+ last_rg = parquet_file.read_row_group(num_row_groups - 1)
+
+ first_time = first_rg.column('EventTime')[0].as_py()
+ last_time = last_rg.column('EventTime')[-1].as_py()
+
+ print(f" First EventTime: {first_time}")
+ print(f" Last EventTime: {last_time}")
+ print(f" Sorted: {'✓ YES' if first_time <= last_time else '✗ NO'}")
+
+ except Exception as e:
+ print(f"Error during verification: {e}")
+
+
+def main():
+ parser = argparse.ArgumentParser(
+ description='Sort ClickBench parquet file by EventTime',
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="""
+Examples:
+ # Basic usage (1M rows per group)
+ %(prog)s input.parquet output.parquet
+
+ # Custom row group size (2M rows)
+ %(prog)s input.parquet output.parquet --row-group-size 2097152
+
+ # Use zstd compression
+ %(prog)s input.parquet output.parquet --compression zstd
+
+ # Verify existing file
+ %(prog)s --verify-only output.parquet
+ """
+ )
+
+ parser.add_argument(
+ 'input',
+ nargs='?',
+ help='Input parquet file path'
+ )
+ parser.add_argument(
+ 'output',
+ nargs='?',
+ help='Output sorted parquet file path'
+ )
+ parser.add_argument(
+ '--row-group-size',
+ type=int,
+ default=64 * 1024, # 64K rows
+ help='Number of rows per row group (default: 65536 = 64K)'
+ )
+ parser.add_argument(
+ '--compression',
+ choices=['snappy', 'gzip', 'brotli', 'lz4', 'zstd', 'none'],
+ default='zstd',
+ help='Compression codec (default: zstd)'
+ )
+ parser.add_argument(
+ '--compression-level',
Review Comment:
The compression_level is ignored and None is always used - https://github.com/apache/datafusion/pull/19042/files#diff-e69296826ec4a87d5995601df87bb71ea4346532b1b4055641ca8959febe510aR104
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 1024 * 1024, # 1M rows default
+ compression: str = 'snappy',
Review Comment:
```suggestion
compression: str = 'zstd',
```
below it uses `zstd` as default - https://github.com/apache/datafusion/pull/19042/files#diff-e69296826ec4a87d5995601df87bb71ea4346532b1b4055641ca8959febe510aR207
##########
benchmarks/bench.sh:
##########
@@ -1197,6 +1206,86 @@ compare_benchmarks() {
}
+# Sorted Data Benchmark Functions (Optimized for hits_0.parquet)
+# Add these functions to bench.sh
+
+# Creates sorted ClickBench data from hits_0.parquet (partitioned dataset)
+# The data is sorted by EventTime in ascending order
+# Using hits_0.parquet (~150MB) instead of full hits.parquet (~14GB) for faster testing
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_0_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits_partitioned/hits_0.parquet"
+
+ echo "Creating sorted ClickBench dataset from hits_0.parquet..."
+
+ # Check if partitioned data exists
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits_partitioned/hits_0.parquet not found. Running data_clickbench_partitioned first..."
+ data_clickbench_partitioned
+ fi
+
+ # Check if sorted file already exists
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits_0.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits_0.parquet by EventTime (this takes ~10 seconds)..."
+
+ # Ensure virtual environment exists and has pyarrow
+ if [ ! -d "$VIRTUAL_ENV" ]; then
+ echo "Creating virtual environment at $VIRTUAL_ENV..."
+ python3 -m venv "$VIRTUAL_ENV"
+ fi
+
+ # Activate virtual environment
+ source "$VIRTUAL_ENV/bin/activate"
+
+ # Check and install pyarrow if needed
+ if ! python3 -c "import pyarrow" 2>/dev/null; then
+ echo "Installing pyarrow (this may take a minute)..."
+ pip install --quiet pyarrow
+ fi
+
+ # Use the standalone Python script to sort
+ python3 "${SCRIPT_DIR}"/sort_clickbench.py "${ORIGINAL_FILE}" "${SORTED_FILE}"
+ local result=$?
+
+ # Deactivate virtual environment
+ deactivate
+
+ if [ $result -eq 0 ]; then
+ echo "✓ Successfully created sorted ClickBench dataset"
+ return 0
+ else
+ echo "✗ Error: Failed to create sorted dataset"
+ return 1
+ fi
+}
+
+# Sorted Data Benchmark Functions for bench.sh
+# Add these functions to your bench.sh script
Review Comment:
What is the purpose of this comment ?
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 1024 * 1024, # 1M rows default
+ compression: str = 'snappy',
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=None, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print(f"\n✓ Successfully created sorted file!")
+ print(f" Input: {input_size_mb:.1f} MB")
+ print(f" Output: {output_size_mb:.1f} MB")
+ print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
+ print(f"\nRow Group Statistics:")
Review Comment:
```suggestion
print("\nRow Group Statistics:")
```
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 1024 * 1024, # 1M rows default
Review Comment:
Below it says that the default is 64K - https://github.com/apache/datafusion/pull/19042/files#diff-e69296826ec4a87d5995601df87bb71ea4346532b1b4055641ca8959febe510aR201
##########
benchmarks/src/clickbench.rs:
##########
@@ -78,6 +78,16 @@ pub struct RunOpt {
/// If present, write results json here
#[structopt(parse(from_os_str), short = "o", long = "output")]
output_path: Option<PathBuf>,
+
+ /// Column name that the data is sorted by (e.g., "EventTime")
+ /// If specified, DataFusion will be informed that the data has this sort order
+ /// using CREATE EXTERNAL TABLE with WITH ORDER clause
+ #[structopt(long = "sorted-by")]
+ sorted_by: Option<String>,
+
+ /// Sort order: ASC or DESC (default: ASC)
+ #[structopt(long = "sort-order", default_value = "ASC")]
Review Comment:
The user may provide any value here.
Consider validating it with a custom parse function.
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 1024 * 1024, # 1M rows default
+ compression: str = 'snappy',
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=None, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print(f"\n✓ Successfully created sorted file!")
+ print(f" Input: {input_size_mb:.1f} MB")
+ print(f" Output: {output_size_mb:.1f} MB")
+ print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
+ print(f"\nRow Group Statistics:")
+ print(f" Total row groups: {num_row_groups}")
+ print(f" Total rows: {len(sorted_table):,}")
+
+ # Show row group details
+ for i in range(min(3, num_row_groups)):
+ rg_metadata = parquet_file.metadata.row_group(i)
+ print(f" Row group {i}: {rg_metadata.num_rows:,} rows, {rg_metadata.total_byte_size / 1024**2:.1f} MB")
+
+ if num_row_groups > 3:
+ print(f" ... and {num_row_groups - 3} more row groups")
+
+ avg_rows_per_group = len(sorted_table) / num_row_groups if num_row_groups > 0 else 0
+ print(f" Average rows per group: {avg_rows_per_group:,.0f}")
+
+ except Exception as e:
+ print(f"Error: {e}")
+ import traceback
+ traceback.print_exc()
+ sys.exit(1)
+
+
+def verify_sorted_file(file_path: Path):
+ """Verify that a parquet file is sorted by EventTime."""
+ try:
+ print(f"Verifying sorted file: {file_path}")
+ parquet_file = pq.ParquetFile(str(file_path))
+
+ num_row_groups = parquet_file.num_row_groups
+ file_size_mb = file_path.stat().st_size / (1024**2)
+
+ print(f" File size: {file_size_mb:.1f} MB")
+ print(f" Row groups: {num_row_groups}")
+
+ # Read first and last row group to verify sort
+ first_rg = parquet_file.read_row_group(0)
+ last_rg = parquet_file.read_row_group(num_row_groups - 1)
+
+ first_time = first_rg.column('EventTime')[0].as_py()
+ last_time = last_rg.column('EventTime')[-1].as_py()
+
+ print(f" First EventTime: {first_time}")
+ print(f" Last EventTime: {last_time}")
+ print(f" Sorted: {'✓ YES' if first_time <= last_time else '✗ NO'}")
+
+ except Exception as e:
+ print(f"Error during verification: {e}")
+
+
+def main():
+ parser = argparse.ArgumentParser(
+ description='Sort ClickBench parquet file by EventTime',
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="""
+Examples:
+ # Basic usage (1M rows per group)
Review Comment:
Actually the default is 64K - https://github.com/apache/datafusion/pull/19042/files#diff-e69296826ec4a87d5995601df87bb71ea4346532b1b4055641ca8959febe510aR201
##########
benchmarks/src/clickbench.rs:
##########
@@ -136,10 +158,24 @@ impl RunOpt {
parquet_options.pushdown_filters = true;
parquet_options.reorder_filters = true;
}
+
+ if self.sorted_by.is_some() {
Review Comment:
```suggestion
else if self.sorted_by.is_some() {
```
nit: I guess you kept them separate intentionally, so feel free to ignore this one
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 1024 * 1024, # 1M rows default
+ compression: str = 'snappy',
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=None, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print(f"\n✓ Successfully created sorted file!")
Review Comment:
```suggestion
print("\n✓ Successfully created sorted file!")
```
##########
benchmarks/bench.sh:
##########
@@ -1197,6 +1206,86 @@ compare_benchmarks() {
}
+# Sorted Data Benchmark Functions (Optimized for hits_0.parquet)
+# Add these functions to bench.sh
Review Comment:
What is the purpose of this comment ?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2594623309
##########
benchmarks/bench.sh:
##########
@@ -1201,6 +1210,101 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
+
+-- Parquet output configuration
+SET datafusion.execution.parquet.max_row_group_size = 65536;
+SET datafusion.execution.parquet.compression = 'uncompressed';
+
+-- Execute sort and write
+COPY (SELECT * FROM '${ORIGINAL_FILE}' ORDER BY "EventTime")
+TO '${SORTED_FILE}'
+STORED AS PARQUET;
+EOF
+
+ local result=$?
+
+ if [ $result -eq 0 ]; then
+ echo "✓ Successfully created sorted ClickBench dataset"
+
+ INPUT_SIZE=$(stat -f%z "${ORIGINAL_FILE}" 2>/dev/null || stat -c%s "${ORIGINAL_FILE}" 2>/dev/null)
+ OUTPUT_SIZE=$(stat -f%z "${SORTED_FILE}" 2>/dev/null || stat -c%s "${SORTED_FILE}" 2>/dev/null)
+ INPUT_MB=$((INPUT_SIZE / 1024 / 1024))
+ OUTPUT_MB=$((OUTPUT_SIZE / 1024 / 1024))
+
+ echo " Input: ${INPUT_MB} MB"
Review Comment:
It was suggested by @2010YOUY01 because we want to speed up the sort for data generation, and we don't care about the compression here, so i set to uncompressed to speed up it.
```rust
SET datafusion.execution.parquet.compression = 'uncompressed';
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2594654703
##########
benchmarks/bench.sh:
##########
@@ -1197,10 +1206,105 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
Review Comment:
```rust
Resources exhausted: Additional allocation failed for ExternalSorterMerge[1] with top memory consumers (across reservations) as:
ExternalSorter[2]#13(can spill: true) consumed 3.7 GB, peak 4.8 GB,
ExternalSorter[3]#15(can spill: true) consumed 3.5 GB, peak 4.4 GB,
ExternalSorterMerge[2]#14(can spill: false) consumed 2.3 GB, peak 2.3 GB,
ExternalSorterMerge[1]#12(can spill: false) consumed 1004.2 MB, peak 1694.0 MB,
ExternalSorterMerge[3]#16(can spill: false) consumed 845.7 MB, peak 1798.9 MB.
Error: Failed to allocate additional 12.7 MB for ExternalSorterMerge[1] with 998.9 MB already allocated for this reservation - 689.7 KB remain available for the total pool
\q
```
ExternalSorterMerge seems to cause the Resources exhausted.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2581523202
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 1024 * 1024, # 1M rows default
+ compression: str = 'snappy',
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=None, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print(f"\n✓ Successfully created sorted file!")
+ print(f" Input: {input_size_mb:.1f} MB")
+ print(f" Output: {output_size_mb:.1f} MB")
+ print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
+ print(f"\nRow Group Statistics:")
+ print(f" Total row groups: {num_row_groups}")
+ print(f" Total rows: {len(sorted_table):,}")
+
+ # Show row group details
+ for i in range(min(3, num_row_groups)):
+ rg_metadata = parquet_file.metadata.row_group(i)
+ print(f" Row group {i}: {rg_metadata.num_rows:,} rows, {rg_metadata.total_byte_size / 1024**2:.1f} MB")
+
+ if num_row_groups > 3:
+ print(f" ... and {num_row_groups - 3} more row groups")
+
+ avg_rows_per_group = len(sorted_table) / num_row_groups if num_row_groups > 0 else 0
+ print(f" Average rows per group: {avg_rows_per_group:,.0f}")
+
+ except Exception as e:
+ print(f"Error: {e}")
+ import traceback
+ traceback.print_exc()
+ sys.exit(1)
+
+
+def verify_sorted_file(file_path: Path):
+ """Verify that a parquet file is sorted by EventTime."""
+ try:
+ print(f"Verifying sorted file: {file_path}")
+ parquet_file = pq.ParquetFile(str(file_path))
+
+ num_row_groups = parquet_file.num_row_groups
+ file_size_mb = file_path.stat().st_size / (1024**2)
+
+ print(f" File size: {file_size_mb:.1f} MB")
+ print(f" Row groups: {num_row_groups}")
+
+ # Read first and last row group to verify sort
+ first_rg = parquet_file.read_row_group(0)
+ last_rg = parquet_file.read_row_group(num_row_groups - 1)
+
+ first_time = first_rg.column('EventTime')[0].as_py()
+ last_time = last_rg.column('EventTime')[-1].as_py()
+
+ print(f" First EventTime: {first_time}")
+ print(f" Last EventTime: {last_time}")
+ print(f" Sorted: {'✓ YES' if first_time <= last_time else '✗ NO'}")
+
+ except Exception as e:
+ print(f"Error during verification: {e}")
+
+
+def main():
+ parser = argparse.ArgumentParser(
+ description='Sort ClickBench parquet file by EventTime',
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="""
+Examples:
+ # Basic usage (1M rows per group)
Review Comment:
Yes.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2581632808
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 1024 * 1024, # 1M rows default
+ compression: str = 'snappy',
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=None, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print(f"\n✓ Successfully created sorted file!")
+ print(f" Input: {input_size_mb:.1f} MB")
+ print(f" Output: {output_size_mb:.1f} MB")
+ print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
+ print(f"\nRow Group Statistics:")
+ print(f" Total row groups: {num_row_groups}")
+ print(f" Total rows: {len(sorted_table):,}")
+
+ # Show row group details
+ for i in range(min(3, num_row_groups)):
+ rg_metadata = parquet_file.metadata.row_group(i)
+ print(f" Row group {i}: {rg_metadata.num_rows:,} rows, {rg_metadata.total_byte_size / 1024**2:.1f} MB")
+
+ if num_row_groups > 3:
+ print(f" ... and {num_row_groups - 3} more row groups")
+
+ avg_rows_per_group = len(sorted_table) / num_row_groups if num_row_groups > 0 else 0
+ print(f" Average rows per group: {avg_rows_per_group:,.0f}")
+
+ except Exception as e:
+ print(f"Error: {e}")
+ import traceback
+ traceback.print_exc()
+ sys.exit(1)
+
+
+def verify_sorted_file(file_path: Path):
+ """Verify that a parquet file is sorted by EventTime."""
+ try:
+ print(f"Verifying sorted file: {file_path}")
+ parquet_file = pq.ParquetFile(str(file_path))
+
+ num_row_groups = parquet_file.num_row_groups
+ file_size_mb = file_path.stat().st_size / (1024**2)
+
+ print(f" File size: {file_size_mb:.1f} MB")
+ print(f" Row groups: {num_row_groups}")
+
+ # Read first and last row group to verify sort
+ first_rg = parquet_file.read_row_group(0)
+ last_rg = parquet_file.read_row_group(num_row_groups - 1)
+
+ first_time = first_rg.column('EventTime')[0].as_py()
+ last_time = last_rg.column('EventTime')[-1].as_py()
+
+ print(f" First EventTime: {first_time}")
+ print(f" Last EventTime: {last_time}")
+ print(f" Sorted: {'✓ YES' if first_time <= last_time else '✗ NO'}")
+
+ except Exception as e:
+ print(f"Error during verification: {e}")
+
+
+def main():
+ parser = argparse.ArgumentParser(
+ description='Sort ClickBench parquet file by EventTime',
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="""
+Examples:
+ # Basic usage (1M rows per group)
+ %(prog)s input.parquet output.parquet
+
+ # Custom row group size (2M rows)
+ %(prog)s input.parquet output.parquet --row-group-size 2097152
+
+ # Use zstd compression
+ %(prog)s input.parquet output.parquet --compression zstd
+
+ # Verify existing file
+ %(prog)s --verify-only output.parquet
+ """
+ )
+
+ parser.add_argument(
+ 'input',
+ nargs='?',
+ help='Input parquet file path'
+ )
+ parser.add_argument(
+ 'output',
+ nargs='?',
+ help='Output sorted parquet file path'
+ )
+ parser.add_argument(
+ '--row-group-size',
+ type=int,
+ default=64 * 1024, # 64K rows
+ help='Number of rows per row group (default: 65536 = 64K)'
+ )
+ parser.add_argument(
+ '--compression',
+ choices=['snappy', 'gzip', 'brotli', 'lz4', 'zstd', 'none'],
+ default='zstd',
+ help='Compression codec (default: zstd)'
+ )
+ parser.add_argument(
+ '--compression-level',
Review Comment:
Changed to a para to pass already in latest PR.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2591923940
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,244 @@
+#!/usr/bin/env python3
+
+# 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.
+
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 64 * 1024, # 64k rows default
+ compression: str = 'zstd',
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
Review Comment:
Great idea @alamb , addressed in latest PR, thanks!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2594622690
##########
benchmarks/bench.sh:
##########
@@ -1197,10 +1206,105 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
Review Comment:
I already set the memory limit here:
SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
I think we can keep target partition for the first step, i can investigate more as follow-up, may be we can speed up it.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2590170963
##########
benchmarks/README.md:
##########
@@ -804,3 +804,40 @@ Getting results...
cancelling thread
done dropping runtime in 83.531417ms
```
+
+## Sorted Data Benchmarks
+
+### Data Sorted ClickBench
+
+Benchmark for queries on pre-sorted data to test sort order optimization.
+
+This benchmark uses a subset of the ClickBench dataset (hits_0.parquet, ~150MB) that has been pre-sorted by the EventTime column. The queries are designed to test DataFusion's performance when the data is already sorted.
Review Comment:
```suggestion
This benchmark uses a subset of the ClickBench dataset (hits_0.parquet, ~150MB) that has been pre-sorted by the EventTime column. The queries are designed to test DataFusion's performance when the data is already sorted as is common in timeseries workloads.
```
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,244 @@
+#!/usr/bin/env python3
+
+# 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.
+
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 64 * 1024, # 64k rows default
+ compression: str = 'zstd',
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
Review Comment:
rather than using pq, would it be possible to use `datafusion-cli` for this (to reduce the number of dependencoes)?
https://datafusion.apache.org/user-guide/sql/dml.html#copy
For example, something like
```sql
> COPY (SELECT * from 'hits.parquet' ORDER BY "EventTime") TO 'output.parquet' OPTIONS (MAX_ROW_GROUP_SIZE 64000);
```
I think that is pretty equialent
##########
benchmarks/bench.sh:
##########
@@ -1197,6 +1206,75 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits_0.parquet (partitioned dataset)
+# The data is sorted by EventTime in ascending order
+# Using hits_0.parquet (~150MB) instead of full hits.parquet (~14GB) for faster testing
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_0_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits_partitioned/hits_0.parquet"
+
+ echo "Creating sorted ClickBench dataset from hits_0.parquet..."
Review Comment:
Another thing you could do is sort hits.parquet (the entire dataset) rather than just 1% of the data.
##########
benchmarks/src/clickbench.rs:
##########
@@ -125,6 +138,18 @@ impl RunOpt {
// configure parquet options
let mut config = self.common.config()?;
+
+ // CRITICAL: If sorted_by is specified, force target_partitions=1
Review Comment:
I recommend using a different option, `datafusion.optimizer.prefer_existing_sort ` which I think is more likely what real systems would be using as it has the same effect but still allows more than one core to be used by other parts of the query
https://datafusion.apache.org/user-guide/configs.html
datafusion.optimizer.prefer_existing_sort | false | When true, DataFusion will opportunistically remove sorts when the data is already sorted, (i.e. setting preserve_order to true on RepartitionExec and using SortPreservingMergeExec) When false, DataFusion will maximize plan parallelism using RepartitionExec even if this requires subsequently resorting data using a SortExec.
-- | -- | --
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#issuecomment-3615963568
> Thank you @zhuqi-lucas @2010YOUY01 and @xudong963 -- this is really nice 👌
>
> I tested it locally like
>
> ```shell
> ./bench.sh data data_sorted_clickbench
> ```
>
> And then
>
> ```shell
> ./bench.sh run data_sorted_clickbench
> ```
>
> It seems to have worked great
>
> ```
> ⚠️ Overriding target_partitions=1 to preserve sort order
> ⚠️ (Because we want to get the pure performance benefit of sorted data to compare)
> 📊 Session config target_partitions: 1
> Registering table with sort order: EventTime ASC
> Executing: CREATE EXTERNAL TABLE hits STORED AS PARQUET LOCATION '/Users/andrewlamb/Software/datafusion2/benchmarks/data/hits_0_sorted.parquet' WITH ORDER ("EventTime" ASC)
> Q0: -- Must set for ClickBench hits_partitioned dataset. See https://github.com/apache/datafusion/issues/16591
> -- set datafusion.execution.parquet.binary_as_string = true
> SELECT * FROM hits ORDER BY "EventTime" DESC limit 10;
>
> Query 0 iteration 0 took 228.9 ms and returned 10 rows
> Query 0 iteration 1 took 178.1 ms and returned 10 rows
> Query 0 iteration 2 took 178.5 ms and returned 10 rows
> Query 0 iteration 3 took 177.8 ms and returned 10 rows
> Query 0 iteration 4 took 179.1 ms and returned 10 rows
> Query 0 avg time: 188.49 ms
> + set +x
> Done
> ```
>
> i have a few suggestions but nothing that is necessary in my opinion.
Thank you @alamb for review, addressed the comments in latest PR.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2593265058
##########
benchmarks/bench.sh:
##########
@@ -99,6 +99,9 @@ clickbench_partitioned: ClickBench queries against partitioned (100 files) parqu
clickbench_pushdown: ClickBench queries against partitioned (100 files) parquet w/ filter_pushdown enabled
clickbench_extended: ClickBench \"inspired\" queries against a single parquet (DataFusion specific)
+# Sorted Data Benchmarks (ORDER BY Optimization)
+data_sorted_clickbench: ClickBench queries on pre-sorted data using prefer_existing_sort (tests sort elimination optimization)
Review Comment:
I realize it is a bit late, but maybe calling this `clickbench_sorted` would be more consistent with the benchmarks above
##########
benchmarks/bench.sh:
##########
@@ -1201,6 +1210,101 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
+
+-- Parquet output configuration
+SET datafusion.execution.parquet.max_row_group_size = 65536;
+SET datafusion.execution.parquet.compression = 'uncompressed';
+
+-- Execute sort and write
+COPY (SELECT * FROM '${ORIGINAL_FILE}' ORDER BY "EventTime")
+TO '${SORTED_FILE}'
+STORED AS PARQUET;
+EOF
+
+ local result=$?
+
+ if [ $result -eq 0 ]; then
+ echo "✓ Successfully created sorted ClickBench dataset"
+
+ INPUT_SIZE=$(stat -f%z "${ORIGINAL_FILE}" 2>/dev/null || stat -c%s "${ORIGINAL_FILE}" 2>/dev/null)
+ OUTPUT_SIZE=$(stat -f%z "${SORTED_FILE}" 2>/dev/null || stat -c%s "${SORTED_FILE}" 2>/dev/null)
+ INPUT_MB=$((INPUT_SIZE / 1024 / 1024))
+ OUTPUT_MB=$((OUTPUT_SIZE / 1024 / 1024))
+
+ echo " Input: ${INPUT_MB} MB"
Review Comment:
I ran this and it showed
```shell
✓ Successfully created sorted ClickBench dataset
Input: 14095 MB
Output: 36159 MB
```
I think that is due to the lack of compression
##########
benchmarks/bench.sh:
##########
@@ -1197,10 +1206,105 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
Review Comment:
> @alamb I tried locally for target_partitions, it only not OOM for setting to 1, even for 2 it will OOM, so i setting 1 here. I am not sure why.
I think by default there is no memory limit.
```
andrewlamb@Andrews-MacBook-Pro-3:~/Software/datafusion$ datafusion-cli --help
Command Line Client for DataFusion query engine.
...
-m, --memory-limit <MEMORY_LIMIT>
The memory pool limitation (e.g. '10g'), default to None (no limit)
```
You can potentially limit the memory usage like this
```shell
diff --git a/benchmarks/bench.sh b/benchmarks/bench.sh
index e79cca1a2a..d227fffde6 100755
--- a/benchmarks/bench.sh
+++ b/benchmarks/bench.sh
@@ -1244,8 +1244,8 @@ data_sorted_clickbench() {
DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
popd > /dev/null
- echo "Using datafusion-cli to create sorted parquet file..."
- "${DATAFUSION_CLI}" << EOF
+ echo "Using datafusion-cli (4GB memory) to create sorted parquet file..."
+ "${DATAFUSION_CLI}" -m 4 << EOF
-- Memory and performance configuration
SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
SET datafusion.execution.spill_compression = 'uncompressed';
```
However, whenI tried that it still wasn't able to re-sort the data 🤔
##########
benchmarks/bench.sh:
##########
@@ -1201,6 +1210,101 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
Review Comment:
nice
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2581507269
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 1024 * 1024, # 1M rows default
Review Comment:
And we already passed 64k above, but better to make it consistent.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2588004724
##########
benchmarks/bench.sh:
##########
@@ -1197,6 +1206,80 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits_0.parquet (partitioned dataset)
+# The data is sorted by EventTime in ascending order
+# Using hits_0.parquet (~150MB) instead of full hits.parquet (~14GB) for faster testing
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_0_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits_partitioned/hits_0.parquet"
+
+ echo "Creating sorted ClickBench dataset from hits_0.parquet..."
+
+ # Check if partitioned data exists
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits_partitioned/hits_0.parquet not found. Running data_clickbench_partitioned first..."
+ data_clickbench_partitioned
+ fi
+
+ # Check if sorted file already exists
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits_0.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits_0.parquet by EventTime (this takes ~10 seconds)..."
+
+ # Ensure virtual environment exists and has pyarrow
+ if [ ! -d "$VIRTUAL_ENV" ]; then
+ echo "Creating virtual environment at $VIRTUAL_ENV..."
+ python3 -m venv "$VIRTUAL_ENV"
+ fi
+
+ # Activate virtual environment
+ source "$VIRTUAL_ENV/bin/activate"
+
+ # Check and install pyarrow if needed
+ if ! python3 -c "import pyarrow" 2>/dev/null; then
+ echo "Installing pyarrow (this may take a minute)..."
+ pip install --quiet pyarrow
+ fi
+
+ # Use the standalone Python script to sort
+ python3 "${SCRIPT_DIR}"/sort_clickbench.py "${ORIGINAL_FILE}" "${SORTED_FILE}"
+ local result=$?
+
+ # Deactivate virtual environment
+ deactivate
Review Comment:
Good idea, i will do this!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2587981805
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,254 @@
+#!/usr/bin/env python3
+
+# 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.
+
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 64 * 1024, # 64k rows default
+ compression: str = 'zstd',
+ compression_level: int = 3,
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=compression_level, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print("\n✓ Successfully created sorted file!")
+ print(f" Input: {input_size_mb:.1f} MB")
+ print(f" Output: {output_size_mb:.1f} MB")
+ print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
+ print("\nRow Group Statistics:")
+ print(f" Total row groups: {num_row_groups}")
+ print(f" Total rows: {len(sorted_table):,}")
+
+ # Show row group details
+ for i in range(min(3, num_row_groups)):
+ rg_metadata = parquet_file.metadata.row_group(i)
+ print(f" Row group {i}: {rg_metadata.num_rows:,} rows, {rg_metadata.total_byte_size / 1024**2:.1f} MB")
+
+ if num_row_groups > 3:
+ print(f" ... and {num_row_groups - 3} more row groups")
+
+ avg_rows_per_group = len(sorted_table) / num_row_groups if num_row_groups > 0 else 0
+ print(f" Average rows per group: {avg_rows_per_group:,.0f}")
+
+ except Exception as e:
+ print(f"Error: {e}")
+ import traceback
+ traceback.print_exc()
+ sys.exit(1)
+
+
+def verify_sorted_file(file_path: Path):
+ """Verify that a parquet file is sorted by EventTime."""
+ try:
+ print(f"Verifying sorted file: {file_path}")
+ parquet_file = pq.ParquetFile(str(file_path))
+
+ num_row_groups = parquet_file.num_row_groups
+ file_size_mb = file_path.stat().st_size / (1024**2)
+
+ print(f" File size: {file_size_mb:.1f} MB")
+ print(f" Row groups: {num_row_groups}")
+
+ # Read first and last row group to verify sort
+ first_rg = parquet_file.read_row_group(0)
+ last_rg = parquet_file.read_row_group(num_row_groups - 1)
+
+ first_time = first_rg.column('EventTime')[0].as_py()
+ last_time = last_rg.column('EventTime')[-1].as_py()
+
+ print(f" First EventTime: {first_time}")
+ print(f" Last EventTime: {last_time}")
+ print(f" Sorted: {'✓ YES' if first_time <= last_time else '✗ NO'}")
+
+ except Exception as e:
+ print(f"Error during verification: {e}")
+
+
+def main():
+ parser = argparse.ArgumentParser(
+ description='Sort ClickBench parquet file by EventTime',
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="""
+Examples:
+ # Basic usage (64k rows per group)
+ %(prog)s input.parquet output.parquet
+
+ # Custom row group size (2M rows)
+ %(prog)s input.parquet output.parquet --row-group-size 2097152
+
+ # Use zstd compression
+ %(prog)s input.parquet output.parquet --compression zstd
+
+ # Verify existing file
+ %(prog)s --verify-only output.parquet
+ """
+ )
+
+ parser.add_argument(
+ 'input',
+ nargs='?',
+ help='Input parquet file path'
+ )
+ parser.add_argument(
+ 'output',
+ nargs='?',
+ help='Output sorted parquet file path'
+ )
+ parser.add_argument(
+ '--row-group-size',
+ type=int,
+ default=64 * 1024, # 64K rows
+ help='Number of rows per row group (default: 65536 = 64K)'
+ )
+ parser.add_argument(
+ '--compression',
+ choices=['snappy', 'gzip', 'brotli', 'lz4', 'zstd', 'none'],
+ default='zstd',
Review Comment:
Good point, i agree.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2594622173
##########
benchmarks/bench.sh:
##########
@@ -1201,6 +1210,101 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
Review Comment:
I already set memory limit here @alamb , i think it's similar to -m limit.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2594624935
##########
benchmarks/bench.sh:
##########
@@ -99,6 +99,9 @@ clickbench_partitioned: ClickBench queries against partitioned (100 files) parqu
clickbench_pushdown: ClickBench queries against partitioned (100 files) parquet w/ filter_pushdown enabled
clickbench_extended: ClickBench \"inspired\" queries against a single parquet (DataFusion specific)
+# Sorted Data Benchmarks (ORDER BY Optimization)
+data_sorted_clickbench: ClickBench queries on pre-sorted data using prefer_existing_sort (tests sort elimination optimization)
Review Comment:
Good point @alamb , it's not late, i addressed in latest PR, thanks!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2581515353
##########
benchmarks/bench.sh:
##########
@@ -1197,6 +1206,86 @@ compare_benchmarks() {
}
+# Sorted Data Benchmark Functions (Optimized for hits_0.parquet)
+# Add these functions to bench.sh
+
+# Creates sorted ClickBench data from hits_0.parquet (partitioned dataset)
+# The data is sorted by EventTime in ascending order
+# Using hits_0.parquet (~150MB) instead of full hits.parquet (~14GB) for faster testing
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_0_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits_partitioned/hits_0.parquet"
+
+ echo "Creating sorted ClickBench dataset from hits_0.parquet..."
+
+ # Check if partitioned data exists
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits_partitioned/hits_0.parquet not found. Running data_clickbench_partitioned first..."
+ data_clickbench_partitioned
+ fi
+
+ # Check if sorted file already exists
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits_0.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits_0.parquet by EventTime (this takes ~10 seconds)..."
+
+ # Ensure virtual environment exists and has pyarrow
+ if [ ! -d "$VIRTUAL_ENV" ]; then
+ echo "Creating virtual environment at $VIRTUAL_ENV..."
+ python3 -m venv "$VIRTUAL_ENV"
+ fi
+
+ # Activate virtual environment
+ source "$VIRTUAL_ENV/bin/activate"
+
+ # Check and install pyarrow if needed
+ if ! python3 -c "import pyarrow" 2>/dev/null; then
+ echo "Installing pyarrow (this may take a minute)..."
+ pip install --quiet pyarrow
+ fi
+
+ # Use the standalone Python script to sort
+ python3 "${SCRIPT_DIR}"/sort_clickbench.py "${ORIGINAL_FILE}" "${SORTED_FILE}"
+ local result=$?
+
+ # Deactivate virtual environment
+ deactivate
+
+ if [ $result -eq 0 ]; then
+ echo "✓ Successfully created sorted ClickBench dataset"
+ return 0
+ else
+ echo "✗ Error: Failed to create sorted dataset"
+ return 1
+ fi
+}
+
+# Sorted Data Benchmark Functions for bench.sh
+# Add these functions to your bench.sh script
Review Comment:
Removed it in latest PR.
##########
benchmarks/bench.sh:
##########
@@ -1197,6 +1206,86 @@ compare_benchmarks() {
}
+# Sorted Data Benchmark Functions (Optimized for hits_0.parquet)
+# Add these functions to bench.sh
Review Comment:
Removed it in latest PR.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2581495220
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+
+# 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.
+
+#!/usr/bin/env python3
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 1024 * 1024, # 1M rows default
Review Comment:
Good catch @martin-g , 64k is better i think.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2587936077
##########
benchmarks/bench.sh:
##########
@@ -1197,6 +1206,80 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits_0.parquet (partitioned dataset)
+# The data is sorted by EventTime in ascending order
+# Using hits_0.parquet (~150MB) instead of full hits.parquet (~14GB) for faster testing
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_0_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits_partitioned/hits_0.parquet"
+
+ echo "Creating sorted ClickBench dataset from hits_0.parquet..."
+
+ # Check if partitioned data exists
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits_partitioned/hits_0.parquet not found. Running data_clickbench_partitioned first..."
+ data_clickbench_partitioned
+ fi
+
+ # Check if sorted file already exists
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits_0.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits_0.parquet by EventTime (this takes ~10 seconds)..."
+
+ # Ensure virtual environment exists and has pyarrow
+ if [ ! -d "$VIRTUAL_ENV" ]; then
+ echo "Creating virtual environment at $VIRTUAL_ENV..."
+ python3 -m venv "$VIRTUAL_ENV"
+ fi
+
+ # Activate virtual environment
+ source "$VIRTUAL_ENV/bin/activate"
+
+ # Check and install pyarrow if needed
+ if ! python3 -c "import pyarrow" 2>/dev/null; then
+ echo "Installing pyarrow (this may take a minute)..."
+ pip install --quiet pyarrow
+ fi
+
+ # Use the standalone Python script to sort
+ python3 "${SCRIPT_DIR}"/sort_clickbench.py "${ORIGINAL_FILE}" "${SORTED_FILE}"
+ local result=$?
+
+ # Deactivate virtual environment
+ deactivate
Review Comment:
I believe users are supposed to activate the `venv` externally, so we only have to add `pyarrow` to `requirements.txt`, and remove the dependency installation steps here.
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,254 @@
+#!/usr/bin/env python3
+
+# 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.
+
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 64 * 1024, # 64k rows default
+ compression: str = 'zstd',
+ compression_level: int = 3,
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=compression_level, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print("\n✓ Successfully created sorted file!")
+ print(f" Input: {input_size_mb:.1f} MB")
+ print(f" Output: {output_size_mb:.1f} MB")
+ print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
+ print("\nRow Group Statistics:")
+ print(f" Total row groups: {num_row_groups}")
+ print(f" Total rows: {len(sorted_table):,}")
+
+ # Show row group details
+ for i in range(min(3, num_row_groups)):
+ rg_metadata = parquet_file.metadata.row_group(i)
+ print(f" Row group {i}: {rg_metadata.num_rows:,} rows, {rg_metadata.total_byte_size / 1024**2:.1f} MB")
+
+ if num_row_groups > 3:
+ print(f" ... and {num_row_groups - 3} more row groups")
+
+ avg_rows_per_group = len(sorted_table) / num_row_groups if num_row_groups > 0 else 0
+ print(f" Average rows per group: {avg_rows_per_group:,.0f}")
+
+ except Exception as e:
+ print(f"Error: {e}")
+ import traceback
+ traceback.print_exc()
+ sys.exit(1)
+
+
+def verify_sorted_file(file_path: Path):
+ """Verify that a parquet file is sorted by EventTime."""
+ try:
+ print(f"Verifying sorted file: {file_path}")
+ parquet_file = pq.ParquetFile(str(file_path))
+
+ num_row_groups = parquet_file.num_row_groups
+ file_size_mb = file_path.stat().st_size / (1024**2)
+
+ print(f" File size: {file_size_mb:.1f} MB")
+ print(f" Row groups: {num_row_groups}")
+
+ # Read first and last row group to verify sort
+ first_rg = parquet_file.read_row_group(0)
+ last_rg = parquet_file.read_row_group(num_row_groups - 1)
+
+ first_time = first_rg.column('EventTime')[0].as_py()
+ last_time = last_rg.column('EventTime')[-1].as_py()
+
+ print(f" First EventTime: {first_time}")
+ print(f" Last EventTime: {last_time}")
+ print(f" Sorted: {'✓ YES' if first_time <= last_time else '✗ NO'}")
+
+ except Exception as e:
+ print(f"Error during verification: {e}")
+
+
+def main():
+ parser = argparse.ArgumentParser(
Review Comment:
Looks like it's not used, and it's using defaults. Should we remove it?
##########
benchmarks/src/clickbench.rs:
##########
@@ -136,10 +158,24 @@ impl RunOpt {
parquet_options.pushdown_filters = true;
parquet_options.reorder_filters = true;
}
+
+ if self.sorted_by.is_some() {
+ // We should compare the dynamic topk optimization when data is sorted, so we make the
+ // assumption that filter pushdown is also enabled in this case.
+ parquet_options.pushdown_filters = true;
+ parquet_options.reorder_filters = true;
+ }
}
let rt_builder = self.common.runtime_env_builder()?;
let ctx = SessionContext::new_with_config_rt(config, rt_builder.build_arc()?);
+
+ // Debug: print actual target_partitions being used
Review Comment:
Looks like it's for debug, should we remove it?
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,254 @@
+#!/usr/bin/env python3
+
+# 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.
+
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
+ import pyarrow.compute as pc
+except ImportError:
+ print("Error: pyarrow is not installed.")
+ print("Please install it with: pip install pyarrow")
+ sys.exit(1)
+
+
+def sort_clickbench_data(
+ input_path: str,
+ output_path: str,
+ row_group_size: int = 64 * 1024, # 64k rows default
+ compression: str = 'zstd',
+ compression_level: int = 3,
+ verify: bool = True
+):
+ """Sort parquet file by EventTime column with optimized settings."""
+
+ input_file = Path(input_path)
+ output_file = Path(output_path)
+
+ if not input_file.exists():
+ print(f"Error: Input file not found: {input_file}")
+ sys.exit(1)
+
+ if output_file.exists():
+ print(f"Sorted file already exists: {output_file}")
+ if verify:
+ verify_sorted_file(output_file)
+ return
+
+ try:
+ print(f"Reading {input_file.name}...")
+ table = pq.read_table(str(input_file))
+
+ print(f"Original table has {len(table):,} rows")
+ print("Sorting by EventTime...")
+
+ # Sort the table by EventTime
+ sorted_indices = pc.sort_indices(table, sort_keys=[("EventTime", "ascending")])
+ sorted_table = pc.take(table, sorted_indices)
+
+ print(f"Sorted table has {len(sorted_table):,} rows")
+
+ # Verify sort
+ event_times = sorted_table.column('EventTime').to_pylist()
+ if event_times and verify:
+ print(f"First EventTime: {event_times[0]}")
+ print(f"Last EventTime: {event_times[-1]}")
+ # Verify ascending order
+ is_sorted = all(event_times[i] <= event_times[i+1] for i in range(min(1000, len(event_times)-1)))
+ print(f"Sort verification (first 1000 rows): {'✓ PASS' if is_sorted else '✗ FAIL'}")
+
+ print(f"Writing sorted file to {output_file}...")
+ print(f" Row group size: {row_group_size:,} rows")
+ print(f" Compression: {compression}")
+
+ # Write sorted table with optimized settings
+ pq.write_table(
+ sorted_table,
+ str(output_file),
+ compression=compression,
+ use_dictionary=True,
+ write_statistics=True,
+ # Optimize row group size for better performance
+ row_group_size=row_group_size,
+ # Set data page size (1MB is good for most cases)
+ data_page_size=1024 * 1024,
+ # Use v2 data page format for better compression
+ use_deprecated_int96_timestamps=False,
+ coerce_timestamps='us', # Use microsecond precision
+ # Batch size for writing
+ write_batch_size=min(row_group_size, 1024 * 64),
+ # Enable compression for all columns
+ compression_level=compression_level, # Use default compression level
+ )
+
+ # Report results
+ input_size_mb = input_file.stat().st_size / (1024**2)
+ output_size_mb = output_file.stat().st_size / (1024**2)
+
+ # Read metadata to verify row groups
+ parquet_file = pq.ParquetFile(str(output_file))
+ num_row_groups = parquet_file.num_row_groups
+
+ print("\n✓ Successfully created sorted file!")
+ print(f" Input: {input_size_mb:.1f} MB")
+ print(f" Output: {output_size_mb:.1f} MB")
+ print(f" Compression ratio: {input_size_mb/output_size_mb:.2f}x")
+ print("\nRow Group Statistics:")
+ print(f" Total row groups: {num_row_groups}")
+ print(f" Total rows: {len(sorted_table):,}")
+
+ # Show row group details
+ for i in range(min(3, num_row_groups)):
+ rg_metadata = parquet_file.metadata.row_group(i)
+ print(f" Row group {i}: {rg_metadata.num_rows:,} rows, {rg_metadata.total_byte_size / 1024**2:.1f} MB")
+
+ if num_row_groups > 3:
+ print(f" ... and {num_row_groups - 3} more row groups")
+
+ avg_rows_per_group = len(sorted_table) / num_row_groups if num_row_groups > 0 else 0
+ print(f" Average rows per group: {avg_rows_per_group:,.0f}")
+
+ except Exception as e:
+ print(f"Error: {e}")
+ import traceback
+ traceback.print_exc()
+ sys.exit(1)
+
+
+def verify_sorted_file(file_path: Path):
+ """Verify that a parquet file is sorted by EventTime."""
+ try:
+ print(f"Verifying sorted file: {file_path}")
+ parquet_file = pq.ParquetFile(str(file_path))
+
+ num_row_groups = parquet_file.num_row_groups
+ file_size_mb = file_path.stat().st_size / (1024**2)
+
+ print(f" File size: {file_size_mb:.1f} MB")
+ print(f" Row groups: {num_row_groups}")
+
+ # Read first and last row group to verify sort
+ first_rg = parquet_file.read_row_group(0)
+ last_rg = parquet_file.read_row_group(num_row_groups - 1)
+
+ first_time = first_rg.column('EventTime')[0].as_py()
+ last_time = last_rg.column('EventTime')[-1].as_py()
+
+ print(f" First EventTime: {first_time}")
+ print(f" Last EventTime: {last_time}")
+ print(f" Sorted: {'✓ YES' if first_time <= last_time else '✗ NO'}")
+
+ except Exception as e:
+ print(f"Error during verification: {e}")
+
+
+def main():
+ parser = argparse.ArgumentParser(
+ description='Sort ClickBench parquet file by EventTime',
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="""
+Examples:
+ # Basic usage (64k rows per group)
+ %(prog)s input.parquet output.parquet
+
+ # Custom row group size (2M rows)
+ %(prog)s input.parquet output.parquet --row-group-size 2097152
+
+ # Use zstd compression
+ %(prog)s input.parquet output.parquet --compression zstd
+
+ # Verify existing file
+ %(prog)s --verify-only output.parquet
+ """
+ )
+
+ parser.add_argument(
+ 'input',
+ nargs='?',
+ help='Input parquet file path'
+ )
+ parser.add_argument(
+ 'output',
+ nargs='?',
+ help='Output sorted parquet file path'
+ )
+ parser.add_argument(
+ '--row-group-size',
+ type=int,
+ default=64 * 1024, # 64K rows
+ help='Number of rows per row group (default: 65536 = 64K)'
+ )
+ parser.add_argument(
+ '--compression',
+ choices=['snappy', 'gzip', 'brotli', 'lz4', 'zstd', 'none'],
+ default='zstd',
Review Comment:
I suggest to use `none` here, `zstd` is quite heavy weight, and significant time will be spent decompression, here I believe we want to focus on the sort part
##########
benchmarks/sort_clickbench.py:
##########
@@ -0,0 +1,254 @@
+#!/usr/bin/env python3
+
+# 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.
+
+"""
+Sort ClickBench data by EventTime for reverse scan benchmark.
+Enhanced version with configurable row group size and optimization options.
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+try:
+ import pyarrow.parquet as pq
Review Comment:
We can add the dependencies to venv like https://github.com/apache/datafusion/pull/10894
##########
benchmarks/src/clickbench.rs:
##########
@@ -78,6 +78,16 @@ pub struct RunOpt {
/// If present, write results json here
#[structopt(parse(from_os_str), short = "o", long = "output")]
output_path: Option<PathBuf>,
+
+ /// Column name that the data is sorted by (e.g., "EventTime")
+ /// If specified, DataFusion will be informed that the data has this sort order
+ /// using CREATE EXTERNAL TABLE with WITH ORDER clause
Review Comment:
```suggestion
/// using CREATE EXTERNAL TABLE with WITH ORDER clause
///
/// This is for `sorted_clickbench` benchmark's internal usage. If you want to specify
/// a different order, make sure the dataset is generated with the same sort order.
```
##########
benchmarks/src/clickbench.rs:
##########
@@ -125,6 +135,18 @@ impl RunOpt {
// configure parquet options
let mut config = self.common.config()?;
+
+ // CRITICAL: If sorted_by is specified, force target_partitions=1
+ // This ensures the file is not split into multiple partitions, we
+ // can get the pure performance benefit of sorted data to compare.
+ if self.sorted_by.is_some() {
+ println!("⚠️ Forcing target_partitions=1 to preserve sort order");
Review Comment:
```suggestion
println!("⚠️ Overriding target_partitions=1 to preserve sort order");
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#issuecomment-3610741404
and update the doc here 👉🏼 https://github.com/apache/datafusion/blob/main/benchmarks/README.md
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2591930569
##########
benchmarks/src/clickbench.rs:
##########
@@ -125,6 +138,18 @@ impl RunOpt {
// configure parquet options
let mut config = self.common.config()?;
+
+ // CRITICAL: If sorted_by is specified, force target_partitions=1
Review Comment:
Good point @alamb , addressed in latest PR, thanks!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add sorted data benchmark. [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #19042:
URL: https://github.com/apache/datafusion/pull/19042#discussion_r2594622690
##########
benchmarks/bench.sh:
##########
@@ -1197,10 +1206,105 @@ compare_benchmarks() {
}
+# Creates sorted ClickBench data from hits.parquet (full dataset)
+# The data is sorted by EventTime in ascending order
+# Uses datafusion-cli to reduce dependencies
+data_sorted_clickbench() {
+ SORTED_FILE="${DATA_DIR}/hits_sorted.parquet"
+ ORIGINAL_FILE="${DATA_DIR}/hits.parquet"
+
+ # Default memory limit is 12GB, can be overridden with DATAFUSION_MEMORY_GB env var
+ MEMORY_LIMIT_GB=${DATAFUSION_MEMORY_GB:-12}
+
+ echo "Creating sorted ClickBench dataset from hits.parquet..."
+ echo "Configuration:"
+ echo " Memory limit: ${MEMORY_LIMIT_GB}G"
+ echo " Row group size: 64K rows"
+ echo " Compression: uncompressed"
+
+ if [ ! -f "${ORIGINAL_FILE}" ]; then
+ echo "hits.parquet not found. Running data_clickbench_1 first..."
+ data_clickbench_1
+ fi
+
+ if [ -f "${SORTED_FILE}" ]; then
+ echo "Sorted hits.parquet already exists at ${SORTED_FILE}"
+ return 0
+ fi
+
+ echo "Sorting hits.parquet by EventTime (this may take several minutes)..."
+
+ pushd "${DATAFUSION_DIR}" > /dev/null
+ echo "Building datafusion-cli..."
+ cargo build --release --bin datafusion-cli
+ DATAFUSION_CLI="${DATAFUSION_DIR}/target/release/datafusion-cli"
+ popd > /dev/null
+
+ echo "Using datafusion-cli to create sorted parquet file..."
+ "${DATAFUSION_CLI}" << EOF
+-- Memory and performance configuration
+SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
+SET datafusion.execution.spill_compression = 'uncompressed';
+SET datafusion.execution.sort_spill_reservation_bytes = 10485760; -- 10MB
+SET datafusion.execution.batch_size = 8192;
+SET datafusion.execution.target_partitions = 1;
Review Comment:
I already set the memory limit here:
SET datafusion.runtime.memory_limit = '${MEMORY_LIMIT_GB}G';
I think we can keep target partition 1 for the first step, i can investigate more as follow-up, may be we can speed up it.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org