feat(planner): Allowing setting sort order of parquet files without specifying the schema

datafusion/core/src/datasource/listing_table_factory.rs

@@ -17,6 +17,7 @@
 
 //! Factory for creating ListingTables with default options
 
+use std::collections::HashSet;
 use std::path::Path;
 use std::sync::Arc;
 
@@ -27,7 +28,7 @@ use crate::datasource::listing::{
 use crate::execution::context::SessionState;
 
 use arrow::datatypes::{DataType, SchemaRef};
-use datafusion_common::{arrow_datafusion_err, DataFusionError};
+use datafusion_common::{arrow_datafusion_err, plan_err, DataFusionError, ToDFSchema};
 use datafusion_common::{config_datafusion_err, Result};
 use datafusion_expr::CreateExternalTable;
 
@@ -113,19 +114,39 @@ impl TableProviderFactory for ListingTableFactory {
  .with_collect_stat(state.config().collect_statistics())
  .with_file_extension(file_extension)
  .with_target_partitions(state.config().target_partitions())
- .with_table_partition_cols(table_partition_cols)
- .with_file_sort_order(cmd.order_exprs.clone());
+ .with_table_partition_cols(table_partition_cols);
 
  options
  .validate_partitions(session_state, &table_path)
  .await?;
 
  let resolved_schema = match provided_schema {
- None => options.infer_schema(session_state, &table_path).await?,
+ // We will need to check the table columns against the schema
+ // this is done so that we can do an ORDER BY for external table creation
+ // specifically for parquet file format.
+ // See: https://github.com/apache/datafusion/issues/7317
+ None => {
+ let schema = options.infer_schema(session_state, &table_path).await?;
+ let df_schema = schema.clone().to_dfschema()?;
+ let column_refs: HashSet<_> = cmd
+ .order_exprs
+ .iter()
+ .flat_map(|sort| sort.iter())
+ .flat_map(|s| s.expr.column_refs())
+ .collect();
+
+ for column in &column_refs {
+ if !df_schema.has_column(column) {
+ return plan_err!("Column {column} is not in schema");
+ }
+ }
+
+ schema
+ }
  Some(s) => s,
  };
  let config = ListingTableConfig::new(table_path)
- .with_listing_options(options)
+ .with_listing_options(options.with_file_sort_order(cmd.order_exprs.clone()))
  .with_schema(resolved_schema);
  let provider = ListingTable::try_new(config)?
  .with_cache(state.runtime_env().cache_manager.get_file_statistic_cache());

datafusion/sql/src/statement.rs

@@ -1136,14 +1136,29 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
  schema: &DFSchemaRef,
  planner_context: &mut PlannerContext,
  ) -> Result<Vec<Vec<SortExpr>>> {
- // Ask user to provide a schema if schema is empty.
+ let mut all_results = vec![];
  if !order_exprs.is_empty() && schema.fields().is_empty() {
- return plan_err!(
- "Provide a schema before specifying the order while creating a table."
- );
+ let mut results = vec![];
+ for expr in order_exprs {
+ for ordered_expr in expr {
+ let order_expr = ordered_expr.expr.to_owned();
+ let order_expr = self.sql_expr_to_logical_expr(
+ order_expr,
+ schema,
+ planner_context,
+ )?;
+ let nulls_first = ordered_expr.nulls_first.unwrap_or(true);
+ let asc = ordered_expr.asc.unwrap_or(true);
+ let sort_expr = SortExpr::new(order_expr, asc, nulls_first);
+ results.push(sort_expr);
+ }
+ let sort_results = &results;
+ all_results.push(sort_results.to_owned());
+ }
+
+ return Ok(all_results);
  }
 
- let mut all_results = vec![];
  for expr in order_exprs {
  // Convert each OrderByExpr to a SortExpr:
  let expr_vec =

datafusion/sql/tests/sql_integration.rs

@@ -2002,6 +2002,13 @@ fn create_external_table_parquet_no_schema() {
  quick_test(sql, expected);
 }
 
+#[test]
+fn create_external_table_parquet_no_schema_sort_order() {
+ let sql = "CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION 'foo.parquet' WITH ORDER (id)";
+ let expected = "CreateExternalTable: Bare { table: \"t\" }";
+ quick_test(sql, expected);
+}
+
 #[test]
 fn equijoin_explicit_syntax() {
  let sql = "SELECT id, order_id \

datafusion/sqllogictest/test_files/create_external_table.slt

@@ -228,3 +228,13 @@ OPTIONS (
  format.delimiter '|',
  has_header false,
  compression gzip);
+
+# Create an external parquet table and infer schema to order by
+
+# query should succeed
+statement ok
+CREATE EXTERNAL TABLE t STORED AS parquet LOCATION '../../parquet-testing/data/alltypes_plain.parquet' WITH ORDER (id);
+
+# query should fail with bad column
+statement error
+CREATE EXTERNAL TABLE t STORED AS parquet LOCATION '../../parquet-testing/data/alltypes_plain.parquet' WITH ORDER (foo);

datafusion/sqllogictest/test_files/order.slt

@@ -653,13 +653,6 @@ physical_plan CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/te
 query error DataFusion error: Error during planning: Column a is not in schema
 CREATE EXTERNAL TABLE dt (a_id integer, a_str string, a_bool boolean) STORED AS CSV WITH ORDER (a ASC) LOCATION 'file://path/to/table';
 
-
-# Create external table with DDL ordered columns without schema
-# When schema is missing the query is expected to fail
-query error DataFusion error: Error during planning: Provide a schema before specifying the order while creating a table\.
-CREATE EXTERNAL TABLE dt STORED AS CSV WITH ORDER (a ASC) LOCATION 'file://path/to/table';
-
-
 # Sort with duplicate sort expressions
 # Table is sorted multiple times on the same column name and should not fail
 statement ok