Store example data directly inside the datafusion-examples (#19141) (#19319)

## 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 ##19141.

## 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.
-->

---------

Co-authored-by: Sergey Zhukov <[email protected]>
Co-authored-by: Andrew Lamb <[email protected]>

Author: 3 people

datafusion-examples/Cargo.toml

@@ -35,18 +35,21 @@ rust-version = { workspace = true }
 [lints]
 workspace = true
 
-[dev-dependencies]
+[dependencies]
 arrow = { workspace = true }
-# arrow_schema is required for record_batch! macro :sad:
-arrow-flight = { workspace = true }
 arrow-schema = { workspace = true }
+datafusion = { workspace = true, default-features = true, features = ["parquet_encryption"] }
+datafusion-common = { workspace = true }
+tempfile = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "parking_lot"] }
+
+[dev-dependencies]
+arrow-flight = { workspace = true }
 async-trait = { workspace = true }
 bytes = { workspace = true }
 dashmap = { workspace = true }
 # note only use main datafusion crate for examples
 base64 = "0.22.1"
-datafusion = { workspace = true, default-features = true, features = ["parquet_encryption"] }
-datafusion-common = { workspace = true }
 datafusion-expr = { workspace = true }
 datafusion-physical-expr-adapter = { workspace = true }
 datafusion-proto = { workspace = true }
@@ -62,9 +65,7 @@ rand = { workspace = true }
 serde_json = { workspace = true }
 strum = { workspace = true }
 strum_macros = { workspace = true }
-tempfile = { workspace = true }
 test-utils = { path = "../test-utils" }
-tokio = { workspace = true, features = ["rt-multi-thread", "parking_lot"] }
 tonic = "0.14"
 tracing = { version = "0.1" }
 tracing-subscriber = { version = "0.3" }

datafusion-examples/README.md

@@ -106,10 +106,11 @@ cargo run --example dataframe -- dataframe
 
 #### Category: Single Process
 
-| Subcommand            | File Path                                                                           | Description                                            |
-| --------------------- | ----------------------------------------------------------------------------------- | ------------------------------------------------------ |
-| dataframe             | [`dataframe/dataframe.rs`](examples/dataframe/dataframe.rs)                         | Query DataFrames from various sources and write output |
-| deserialize_to_struct | [`dataframe/deserialize_to_struct.rs`](examples/dataframe/deserialize_to_struct.rs) | Convert Arrow arrays into Rust structs                 |
+| Subcommand            | File Path                                                                           | Description                                             |
+| --------------------- | ----------------------------------------------------------------------------------- | ------------------------------------------------------- |
+| cache_factory         | [`dataframe/cache_factory.rs`](examples/dataframe/cache_factory.rs)                 | Custom lazy caching for DataFrames using `CacheFactory` |
+| dataframe             | [`dataframe/dataframe.rs`](examples/dataframe/dataframe.rs)                         | Query DataFrames from various sources and write output  |
+| deserialize_to_struct | [`dataframe/deserialize_to_struct.rs`](examples/dataframe/deserialize_to_struct.rs) | Convert Arrow arrays into Rust structs                  |
 
 ## Execution Monitoring Examples
 

datafusion-examples/data/README.md

@@ -0,0 +1,25 @@
+<!---
+  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.
+-->
+
+## Example datasets
+
+| Filename    | Path                                    | Description                                                                                                                                                                          |
+| ----------- | --------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
+| `cars.csv`  | [`data/csv/cars.csv`](./csv/cars.csv)   | Time-series–like dataset containing car identifiers, speed values, and timestamps. Used in window function and time-based query examples (e.g. ordering, window frames).             |
+| `regex.csv` | [`data/csv/regex.csv`](./csv/regex.csv) | Dataset for regular expression examples. Contains input values, regex patterns, replacement strings, and optional flags. Covers ASCII, Unicode, and locale-specific text processing. |

datafusion-examples/data/csv/cars.csv

@@ -0,0 +1,26 @@
+car,speed,time
+red,20.0,1996-04-12T12:05:03.000000000
+red,20.3,1996-04-12T12:05:04.000000000
+red,21.4,1996-04-12T12:05:05.000000000
+red,21.5,1996-04-12T12:05:06.000000000
+red,19.0,1996-04-12T12:05:07.000000000
+red,18.0,1996-04-12T12:05:08.000000000
+red,17.0,1996-04-12T12:05:09.000000000
+red,7.0,1996-04-12T12:05:10.000000000
+red,7.1,1996-04-12T12:05:11.000000000
+red,7.2,1996-04-12T12:05:12.000000000
+red,3.0,1996-04-12T12:05:13.000000000
+red,1.0,1996-04-12T12:05:14.000000000
+red,0.0,1996-04-12T12:05:15.000000000
+green,10.0,1996-04-12T12:05:03.000000000
+green,10.3,1996-04-12T12:05:04.000000000
+green,10.4,1996-04-12T12:05:05.000000000
+green,10.5,1996-04-12T12:05:06.000000000
+green,11.0,1996-04-12T12:05:07.000000000
+green,12.0,1996-04-12T12:05:08.000000000
+green,14.0,1996-04-12T12:05:09.000000000
+green,15.0,1996-04-12T12:05:10.000000000
+green,15.1,1996-04-12T12:05:11.000000000
+green,15.2,1996-04-12T12:05:12.000000000
+green,8.0,1996-04-12T12:05:13.000000000
+green,2.0,1996-04-12T12:05:14.000000000

datafusion-examples/data/csv/regex.csv

@@ -0,0 +1,12 @@
+values,patterns,replacement,flags
+abc,^(a),bb\1bb,i
+ABC,^(A).*,B,i
+aBc,(b|d),e,i
+AbC,(B|D),e,
+aBC,^(b|c),d,
+4000,\b4([1-9]\d\d|\d[1-9]\d|\d\d[1-9])\b,xyz,
+4010,\b4([1-9]\d\d|\d[1-9]\d|\d\d[1-9])\b,xyz,
+Düsseldorf,[\p{Letter}-]+,München,
+Москва,[\p{L}-]+,Moscow,
+Köln,[a-zA-Z]ö[a-zA-Z]{2},Koln,
+اليوم,^\p{Arabic}+$,Today,

datafusion-examples/examples/builtin_functions/regexp.rs

@@ -1,5 +1,4 @@
 // Licensed to the Apache Software Foundation (ASF) under one
-// 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
@@ -18,12 +17,10 @@
 
 //! See `main.rs` for how to run it.
 
-use std::{fs::File, io::Write};
-
 use datafusion::common::{assert_batches_eq, assert_contains};
 use datafusion::error::Result;
 use datafusion::prelude::*;
-use tempfile::tempdir;
+use datafusion_examples::utils::datasets::ExampleDataset;
 
 /// This example demonstrates how to use the regexp_* functions
 ///
@@ -35,29 +32,9 @@ use tempfile::tempdir;
 /// https://docs.rs/regex/latest/regex/#grouping-and-flags
 pub async fn regexp() -> Result<()> {
     let ctx = SessionContext::new();
-    // content from file 'datafusion/physical-expr/tests/data/regex.csv'
-    let csv_data = r#"values,patterns,replacement,flags
-abc,^(a),bb\1bb,i
-ABC,^(A).*,B,i
-aBc,(b|d),e,i
-AbC,(B|D),e,
-aBC,^(b|c),d,
-4000,\b4([1-9]\d\d|\d[1-9]\d|\d\d[1-9])\b,xyz,
-4010,\b4([1-9]\d\d|\d[1-9]\d|\d\d[1-9])\b,xyz,
-Düsseldorf,[\p{Letter}-]+,München,
-Москва,[\p{L}-]+,Moscow,
-Köln,[a-zA-Z]ö[a-zA-Z]{2},Koln,
-اليوم,^\p{Arabic}+$,Today,"#;
-    let dir = tempdir()?;
-    let file_path = dir.path().join("regex.csv");
-    {
-        let mut file = File::create(&file_path)?;
-        // write CSV data
-        file.write_all(csv_data.as_bytes())?;
-    } // scope closes the file
-    let file_path = file_path.to_str().unwrap();
-
-    ctx.register_csv("examples", file_path, CsvReadOptions::new())
+    let dataset = ExampleDataset::Regex;
+
+    ctx.register_csv("examples", dataset.path_str()?, CsvReadOptions::new())
         .await?;
 
     //

datafusion-examples/examples/custom_data_source/csv_json_opener.rs

@@ -31,10 +31,10 @@ use datafusion::{
     },
     error::Result,
     physical_plan::metrics::ExecutionPlanMetricsSet,
-    test_util::aggr_test_schema,
 };
 
 use datafusion::datasource::physical_plan::FileScanConfigBuilder;
+use datafusion_examples::utils::datasets::ExampleDataset;
 use futures::StreamExt;
 use object_store::{ObjectStore, local::LocalFileSystem, memory::InMemory};
 
@@ -50,12 +50,10 @@ pub async fn csv_json_opener() -> Result<()> {
 
 async fn csv_opener() -> Result<()> {
     let object_store = Arc::new(LocalFileSystem::new());
-    let schema = aggr_test_schema();
 
-    let testdata = datafusion::test_util::arrow_test_data();
-    let path = format!("{testdata}/csv/aggregate_test_100.csv");
-
-    let path = std::path::Path::new(&path).canonicalize()?;
+    let dataset = ExampleDataset::Cars;
+    let csv_path = dataset.path();
+    let schema = dataset.schema();
 
     let options = CsvOptions {
         has_header: Some(true),
@@ -71,9 +69,9 @@ async fn csv_opener() -> Result<()> {
 
     let scan_config =
         FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), source)
-            .with_projection_indices(Some(vec![12, 0]))?
+            .with_projection_indices(Some(vec![0, 1]))?
             .with_limit(Some(5))
-            .with_file(PartitionedFile::new(path.display().to_string(), 10))
+            .with_file(PartitionedFile::new(csv_path.display().to_string(), 10))
             .build();
 
     let opener =
@@ -89,15 +87,15 @@ async fn csv_opener() -> Result<()> {
     }
     assert_batches_eq!(
         &[
-            "+--------------------------------+----+",
-            "| c13                            | c1 |",
-            "+--------------------------------+----+",
-            "| 6WfVFBVGJSQb7FhA7E0lBwdvjfZnSW | c  |",
-            "| C2GT5KVyOPZpgKVl110TyZO0NcJ434 | d  |",
-            "| AyYVExXK6AR2qUTxNZ7qRHQOVGMLcz | b  |",
-            "| 0keZ5G8BffGwgF2RwQD59TFzMStxCB | a  |",
-            "| Ig1QcuKsjHXkproePdERo2w0mYzIqd | b  |",
-            "+--------------------------------+----+",
+            "+-----+-------+",
+            "| car | speed |",
+            "+-----+-------+",
+            "| red | 20.0  |",
+            "| red | 20.3  |",
+            "| red | 21.4  |",
+            "| red | 21.5  |",
+            "| red | 19.0  |",
+            "+-----+-------+",
         ],
         &result
     );

datafusion-examples/examples/custom_data_source/csv_sql_streaming.rs

@@ -17,43 +17,44 @@
 
 //! See `main.rs` for how to run it.
 
-use datafusion::common::test_util::datafusion_test_data;
 use datafusion::error::Result;
 use datafusion::prelude::*;
+use datafusion_examples::utils::datasets::ExampleDataset;
 
 /// This example demonstrates executing a simple query against an Arrow data source (CSV) and
 /// fetching results with streaming aggregation and streaming window
 pub async fn csv_sql_streaming() -> Result<()> {
     // create local execution context
     let ctx = SessionContext::new();
 
-    let testdata = datafusion_test_data();
+    let dataset = ExampleDataset::Cars;
+    let csv_path = dataset.path();
 
-    // Register a table source and tell DataFusion the file is ordered by `ts ASC`.
+    // Register a table source and tell DataFusion the file is ordered by `car ASC`.
     // Note it is the responsibility of the user to make sure
     // that file indeed satisfies this condition or else incorrect answers may be produced.
     let asc = true;
     let nulls_first = true;
-    let sort_expr = vec![col("ts").sort(asc, nulls_first)];
+    let sort_expr = vec![col("car").sort(asc, nulls_first)];
     // register csv file with the execution context
     ctx.register_csv(
         "ordered_table",
-        &format!("{testdata}/window_1.csv"),
+        csv_path.to_str().unwrap(),
         CsvReadOptions::new().file_sort_order(vec![sort_expr]),
     )
     .await?;
 
     // execute the query
-    // Following query can be executed with unbounded sources because group by expressions (e.g ts) is
+    // Following query can be executed with unbounded sources because group by expressions (e.g car) is
     // already ordered at the source.
     //
     // Unbounded sources means that if the input came from a "never ending" source (such as a FIFO
     // file on unix) the query could produce results incrementally as data was read.
     let df = ctx
         .sql(
-            "SELECT ts, MIN(inc_col), MAX(inc_col) \
+            "SELECT car, MIN(speed), MAX(speed) \
         FROM ordered_table \
-        GROUP BY ts",
+        GROUP BY car",
         )
         .await?;
 
@@ -64,7 +65,7 @@ pub async fn csv_sql_streaming() -> Result<()> {
     // its result in streaming fashion, because its required ordering is already satisfied at the source.
     let df = ctx
         .sql(
-            "SELECT ts, SUM(inc_col) OVER(ORDER BY ts ASC) \
+            "SELECT car, SUM(speed) OVER(ORDER BY car ASC) \
         FROM ordered_table",
         )
         .await?;

datafusion-examples/examples/data_io/parquet_encrypted.rs

@@ -17,28 +17,30 @@
 
 //! See `main.rs` for how to run it.
 
+use std::sync::Arc;
+
 use datafusion::common::DataFusionError;
 use datafusion::config::{ConfigFileEncryptionProperties, TableParquetOptions};
 use datafusion::dataframe::{DataFrame, DataFrameWriteOptions};
 use datafusion::logical_expr::{col, lit};
 use datafusion::parquet::encryption::decrypt::FileDecryptionProperties;
 use datafusion::parquet::encryption::encrypt::FileEncryptionProperties;
 use datafusion::prelude::{ParquetReadOptions, SessionContext};
-use std::sync::Arc;
+use datafusion_examples::utils::{datasets::ExampleDataset, write_csv_to_parquet};
 use tempfile::TempDir;
 
 /// Read and write encrypted Parquet files using DataFusion
 pub async fn parquet_encrypted() -> datafusion::common::Result<()> {
     // The SessionContext is the main high level API for interacting with DataFusion
     let ctx = SessionContext::new();
 
-    // Find the local path of "alltypes_plain.parquet"
-    let testdata = datafusion::test_util::parquet_test_data();
-    let filename = &format!("{testdata}/alltypes_plain.parquet");
+    // Convert the CSV input into a temporary Parquet directory for querying
+    let dataset = ExampleDataset::Cars;
+    let parquet_temp = write_csv_to_parquet(&ctx, &dataset.path()).await?;
 
     // Read the sample parquet file
     let parquet_df = ctx
-        .read_parquet(filename, ParquetReadOptions::default())
+        .read_parquet(parquet_temp.path_str()?, ParquetReadOptions::default())
         .await?;
 
     // Show information from the dataframe
@@ -52,27 +54,28 @@ pub async fn parquet_encrypted() -> datafusion::common::Result<()> {
     let (encrypt, decrypt) = setup_encryption(&parquet_df)?;
 
     // Create a temporary file location for the encrypted parquet file
-    let tmp_dir = TempDir::new()?;
-    let tempfile = tmp_dir.path().join("alltypes_plain-encrypted.parquet");
-    let tempfile_str = tempfile.into_os_string().into_string().unwrap();
+    let tmp_source = TempDir::new()?;
+    let tempfile = tmp_source.path().join("cars_encrypted");
 
     // Write encrypted parquet
     let mut options = TableParquetOptions::default();
     options.crypto.file_encryption = Some(ConfigFileEncryptionProperties::from(&encrypt));
     parquet_df
         .write_parquet(
-            tempfile_str.as_str(),
+            tempfile.to_str().unwrap(),
             DataFrameWriteOptions::new().with_single_file_output(true),
             Some(options),
         )
         .await?;
 
-    // Read encrypted parquet
+    // Read encrypted parquet back as a DataFrame using matching decryption config
     let ctx: SessionContext = SessionContext::new();
     let read_options =
         ParquetReadOptions::default().file_decryption_properties((&decrypt).into());
 
-    let encrypted_parquet_df = ctx.read_parquet(tempfile_str, read_options).await?;
+    let encrypted_parquet_df = ctx
+        .read_parquet(tempfile.to_str().unwrap(), read_options)
+        .await?;
 
     // Show information from the dataframe
     println!(
@@ -91,11 +94,12 @@ async fn query_dataframe(df: &DataFrame) -> Result<(), DataFusionError> {
     df.clone().describe().await?.show().await?;
 
     // Select three columns and filter the results
-    // so that only rows where id > 1 are returned
+    // so that only rows where speed > 5 are returned
+    // select car, speed, time from t where speed > 5
     println!("\nSelected rows and columns:");
     df.clone()
-        .select_columns(&["id", "bool_col", "timestamp_col"])?
-        .filter(col("id").gt(lit(5)))?
+        .select_columns(&["car", "speed", "time"])?
+        .filter(col("speed").gt(lit(5)))?
         .show()
         .await?;