More sophisticated cost based optimizer for join ordering
Implement advanced query optimization framework (Tokomak) #440
Implement advanced query optimization framework (Tokomak) #440
Finer optimizations for group by and aggregate functions
Migrate to some sort of arrow2 based implementation (see milestone for more details)
Add DataFusion to h2oai/db-benchmark 147
Improve build time 348
Add DataFusion to h2oai/db-benchmark #147
Improve build time #348
The DataFusion CLI is a command-line interactive SQL utility that allows -queries to be executed against any supported data files. It is a convenient way to +
The DataFusion CLI is a command-line interactive SQL utility for executing +queries against any supported data files. It is a convenient way to try DataFusion out with your own data sources, and test out its SQL support.
Create a CSV file to query.
-$ echo "1,2" > data.csv
+$ echo "a,b" > data.csv
+$ echo "1,2" >> data.csv
-$ datafusion-cli
-DataFusion CLI v12.0.0
-❯ CREATE EXTERNAL TABLE foo STORED AS CSV LOCATION 'data.csv';
-0 rows in set. Query took 0.017 seconds.
-❯ select * from foo;
-+----------+----------+
-| column_1 | column_2 |
-+----------+----------+
-| 1 | 2 |
-+----------+----------+
-1 row in set. Query took 0.012 seconds.
+Query that single file (the CLI also supports parquet, compressed csv, avro, json and more)
+$ datafusion-cli
+DataFusion CLI v17.0.0
+❯ select * from 'data.csv';
++---+---+
+| a | b |
++---+---+
+| 1 | 2 |
++---+---+
+1 row in set. Query took 0.007 seconds.
+
+
+You can also query directories of files with compatible schemas:
+$ ls data_dir/
+data.csv data2.csv
+
+
+$ datafusion-cli
+DataFusion CLI v16.0.0
+❯ select * from 'data_dir';
++---+---+
+| a | b |
++---+---+
+| 3 | 4 |
+| 1 | 2 |
++---+---+
+2 rows in set. Query took 0.007 seconds.
@@ -430,6 +452,7 @@ Run using Docker
Usage¶
+See the current usage using datafusion-cli --help:
Apache Arrow <dev@arrow.apache.org>
Command Line Client for DataFusion query engine.
@@ -446,11 +469,16 @@ Usage
-q, --quiet Reduce printing other than the results and work quietly
-r, --rc <RC>... Run the provided files on startup instead of ~/.datafusionrc
-V, --version Print version information
-
-Type `exit` or `quit` to exit the CLI.
+
+Selecting files directly¶
+Files can be queried directly by enclosing the file or
+directory name in single ' quotes as shown in the example.
+It is also possible to create a table backed by files by explicitly
+via CREATE EXTERNAL TABLE as shown below.
+
Registering Parquet Data Sources¶
Parquet data sources can be registered by executing a CREATE EXTERNAL TABLE SQL statement. It is not necessary to provide schema information for Parquet files.
diff --git a/datafusion/user-guide/configs.html b/datafusion/user-guide/configs.html
index 3ae271b6fb0d..b340d1022082 100644
--- a/datafusion/user-guide/configs.html
+++ b/datafusion/user-guide/configs.html
@@ -321,116 +321,130 @@ Configuration Settings
key
type
default
description
datafusion.catalog.create_default_catalog_and_schema
true
Number of partitions for query execution. Increasing partitions can increase concurrency. Defaults to the number of cpu cores on the system.
datafusion.catalog.default_catalog
datafusion
The default catalog name - this impacts what SQL queries use if not specified
datafusion.catalog.default_schema
public
The default schema name - this impacts what SQL queries use if not specified
datafusion.catalog.information_schema
false
Should DataFusion provide access to information_schema virtual tables for displaying schema information
datafusion.catalog.location
Utf8
NULL
Location scanned to load tables for default schema, defaults to None
Location scanned to load tables for default schema
datafusion.catalog.type
Utf8
datafusion.catalog.format
NULL
Type of TableProvider to use when loading default schema. Defaults to None
Type of TableProvider to use when loading default schema
datafusion.execution.batch_size
UInt64
datafusion.catalog.has_header
false
If the file has a header
datafusion.execution.batch_size
8192
Default batch size while creating new batches, it’s especially useful for buffer-in-memory batches since creating tiny batches would results in too much metadata memory consumption.
Default batch size while creating new batches, it’s especially useful for buffer-in-memory batches since creating tiny batches would results in too much metadata memory consumption
datafusion.execution.coalesce_batches
Boolean
datafusion.execution.coalesce_batches
true
When set to true, record batches will be examined between each operator and small batches will be coalesced into larger batches. This is helpful when there are highly selective filters or joins that could produce tiny output batches. The target batch size is determined by the configuration setting ‘datafusion.execution.coalesce_target_batch_size’.
datafusion.execution.coalesce_target_batch_size
UInt64
4096
Target batch size when coalescing batches. Uses in conjunction with the configuration setting ‘datafusion.execution.coalesce_batches’.
When set to true, record batches will be examined between each operator and small batches will be coalesced into larger batches. This is helpful when there are highly selective filters or joins that could produce tiny output batches. The target batch size is determined by the configuration setting
datafusion.execution.parquet.enable_page_index
Boolean
datafusion.execution.collect_statistics
false
If true, uses parquet data page level metadata (Page Index) statistics to reduce the number of rows decoded.
Should DataFusion collect statistics after listing files
datafusion.execution.parquet.metadata_size_hint
UInt64
NULL
If specified, the parquet reader will try and fetch the last size_hint bytes of the parquet file optimistically. If not specified, two read are required: One read to fetch the 8-byte parquet footer and another to fetch the metadata length encoded in the footer.
datafusion.execution.target_partitions
0
Number of partitions for query execution. Increasing partitions can increase concurrency. Defaults to the number of cpu cores on the system
datafusion.execution.parquet.pruning
Boolean
true
If true, the parquet reader attempts to skip entire row groups based on the predicate in the query and the metadata (min/max values) stored in the parquet file.
datafusion.execution.time_zone
+00:00
The default time zone Some functions, e.g. EXTRACT(HOUR from SOME_TIME), shift the underlying datetime according to this time zone, and then extract the hour
datafusion.execution.parquet.pushdown_filters
Boolean
datafusion.execution.parquet.enable_page_index
false
If true, filter expressions are be applied during the parquet decoding operation to reduce the number of rows decoded.
If true, uses parquet data page level metadata (Page Index) statistics to reduce the number of rows decoded.
datafusion.execution.parquet.reorder_filters
Boolean
false
If true, filter expressions evaluated during the parquet decoding opearation will be reordered heuristically to minimize the cost of evaluation. If false, the filters are applied in the same order as written in the query.
datafusion.execution.parquet.pruning
true
If true, the parquet reader attempts to skip entire row groups based on the predicate in the query and the metadata (min/max values) stored in the parquet file
datafusion.execution.parquet.skip_metadata
Boolean
true
If true, the parquet reader skip the optional embedded metadata that may be in the file Schema. This setting can help avoid schema conflicts when querying multiple parquet files with schemas containing compatible types but different metadata.
datafusion.execution.time_zone
Utf8
+00:00
The session time zone which some function require e.g. EXTRACT(HOUR from SOME_TIME) shift the underline datetime according to the time zone,
If true, the parquet reader skip the optional embedded metadata that may be in the file Schema. This setting can help avoid schema conflicts when querying multiple parquet files with schemas containing compatible types but different metadata
then extract the hour.
datafusion.execution.parquet.metadata_size_hint
NULL
If specified, the parquet reader will try and fetch the last size_hint bytes of the parquet file optimistically. If not specified, two read are required: One read to fetch the 8-byte parquet footer and another to fetch the metadata length encoded in the footer
datafusion.explain.logical_plan_only
Boolean
datafusion.execution.parquet.pushdown_filters
false
When set to true, the explain statement will only print logical plans.
If true, filter expressions are be applied during the parquet decoding operation to reduce the number of rows decoded
datafusion.explain.physical_plan_only
Boolean
datafusion.execution.parquet.reorder_filters
false
When set to true, the explain statement will only print physical plans.
If true, filter expressions evaluated during the parquet decoding operation will be reordered heuristically to minimize the cost of evaluation. If false, the filters are applied in the same order as written in the query
datafusion.optimizer.enable_round_robin_repartition
true
When set to true, the physical plan optimizer will try to add round robin repartition to increase parallelism to leverage more CPU cores
datafusion.optimizer.filter_null_join_keys
Boolean
false
When set to true, the optimizer will insert filters before a join between a nullable and non-nullable column to filter out nulls on the nullable side. This filter can add additional overhead when the file format does not fully support predicate push down.
datafusion.optimizer.hash_join_single_partition_threshold
UInt64
1048576
The maximum estimated size in bytes for one input side of a HashJoin will be collected into a single partition
datafusion.optimizer.repartition_aggregations
true
Should DataFusion repartition data using the aggregate keys to execute aggregates in parallel using the provided target_partitions level”
datafusion.optimizer.max_passes
UInt64
3
Number of times that the optimizer will attempt to optimize the plan
datafusion.optimizer.repartition_joins
true
Should DataFusion repartition data using the join keys to execute joins in parallel using the provided target_partitions level”
datafusion.optimizer.prefer_hash_join
Boolean
datafusion.optimizer.repartition_windows
true
When set to true, the physical plan optimizer will prefer HashJoin over SortMergeJoin. HashJoin can work more efficientlythan SortMergeJoin but consumes more memory. Defaults to true
Should DataFusion repartition data using the partitions keys to execute window functions in parallel using the provided target_partitions level”
datafusion.optimizer.skip_failed_rules
Boolean
true
When set to true, the logical plan optimizer will produce warning messages if any optimization rules produce errors and then proceed to the next rule. When set to false, any rules that produce errors will cause the query to fail.
When set to true, the logical plan optimizer will produce warning messages if any optimization rules produce errors and then proceed to the next rule. When set to false, any rules that produce errors will cause the query to fail
datafusion.optimizer.max_passes
3
Number of times that the optimizer will attempt to optimize the plan
datafusion.optimizer.top_down_join_key_reordering
true
When set to true, the physical plan optimizer will run a top down process to reorder the join keys
datafusion.optimizer.top_down_join_key_reordering
Boolean
datafusion.optimizer.prefer_hash_join
true
When set to true, the physical plan optimizer will run a top down process to reorder the join keys. Defaults to true
When set to true, the physical plan optimizer will prefer HashJoin over SortMergeJoin. HashJoin can work more efficiently than SortMergeJoin but consumes more memory
datafusion.optimizer.hash_join_single_partition_threshold
1048576
The maximum estimated size in bytes for one input side of a HashJoin will be collected into a single partition
datafusion.explain.logical_plan_only
false
When set to true, the explain statement will only print logical plans
datafusion.explain.physical_plan_only
false
When set to true, the explain statement will only print physical plans
DataFrame APIlet ctx = SessionContext::new();
-let df = ctx.read_csv("tests/example.csv", CsvReadOptions::new()).await?;
+let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.filter(col("a").lt_eq(col("b")))?
.aggregate(vec![col("a")], vec![min(col("b"))])?
.limit(0, Some(100))?;
diff --git a/datafusion/user-guide/example-usage.html b/datafusion/user-guide/example-usage.html
index e3b40604abc6..a997783dfbfd 100644
--- a/datafusion/user-guide/example-usage.html
+++ b/datafusion/user-guide/example-usage.html
@@ -364,7 +364,7 @@
-->
Example Usage¶
-In this example some simple processing is performed on the example.csv file.
+In this example some simple processing is performed on the example.csv file.
Update Cargo.toml¶
Add the following to your Cargo.toml file:
@@ -381,7 +381,7 @@ Run a SQL query against data stored in a CSV:async fn main() -> datafusion::error::Result<()> {
// register the table
let ctx = SessionContext::new();
- ctx.register_csv("example", "tests/example.csv", CsvReadOptions::new()).await?;
+ ctx.register_csv("example", "tests/data/example.csv", CsvReadOptions::new()).await?;
// create a plan to run a SQL query
let df = ctx.sql("SELECT a, MIN(b) FROM example GROUP BY a LIMIT 100").await?;
@@ -401,7 +401,7 @@ Use the DataFrame API to process data stored in a CSV:async fn main() -> datafusion::error::Result<()> {
// create the dataframe
let ctx = SessionContext::new();
- let df = ctx.read_csv("tests/example.csv", CsvReadOptions::new()).await?;
+ let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.filter(col("a").lt_eq(col("b")))?
.aggregate(vec![col("a")], vec![min(col("b"))])?
@@ -428,7 +428,7 @@ Output from both examples
Identifiers and Capitalization¶
Please be aware that all identifiers are effectively made lower-case in SQL, so if your csv file has capital letters (ex: Name) you must put your column name in double quotes or the examples won’t work.
-To illustrate this behavior, consider the capitalized_example.csv file:
+To illustrate this behavior, consider the capitalized_example.csv file:
Run a SQL query against data stored in a CSV:¶
use datafusion::prelude::*;
@@ -437,7 +437,7 @@ Run a SQL query against data stored in a CSV:async fn main() -> datafusion::error::Result<()> {
// register the table
let ctx = SessionContext::new();
- ctx.register_csv("example", "tests/capitalized_example.csv", CsvReadOptions::new()).await?;
+ ctx.register_csv("example", "tests/data/capitalized_example.csv", CsvReadOptions::new()).await?;
// create a plan to run a SQL query
let df = ctx.sql("SELECT \"A\", MIN(b) FROM example GROUP BY \"A\" LIMIT 100").await?;
@@ -457,7 +457,7 @@ Use the DataFrame API to process data stored in a CSV:async fn main() -> datafusion::error::Result<()> {
// create the dataframe
let ctx = SessionContext::new();
- let df = ctx.read_csv("tests/capitalized_example.csv", CsvReadOptions::new()).await?;
+ let df = ctx.read_csv("tests/data/capitalized_example.csv", CsvReadOptions::new()).await?;
let df = df.filter(col("A").lt_eq(col("c")))?
.aggregate(vec![col("A")], vec![min(col("b"))])?