amend ListingTableConfig

  if file_schema is provided, then schema_policy is automatically Fixed

Author: kosiew

datafusion/core/src/datasource/listing/table.rs

@@ -150,13 +150,15 @@ impl ListingTableConfig {
     ///
     /// If the schema is provided, it must contain only the fields in the file
     /// without the table partitioning columns.
+    ///
+    /// Note: Providing a schema automatically sets the schema policy to `SchemaPolicy::Fixed`.
     pub fn with_schema(self, schema: SchemaRef) -> Self {
         Self {
             table_paths: self.table_paths,
-            file_schema: Some(schema),
+            file_schema: Some(schema.clone()),
             options: self.options,
             schema_adapter_factory: self.schema_adapter_factory,
-            schema_policy: Some(SchemaPolicy::Fixed(schema.clone())),
+            schema_policy: Some(SchemaPolicy::Fixed(schema)),
         }
     }
 
@@ -165,37 +167,99 @@ impl ListingTableConfig {
     /// If not provided, format and other options are inferred via
     /// [`Self::infer_options`].
     pub fn with_listing_options(self, listing_options: ListingOptions) -> Self {
-        Self {
+        let config = Self {
             table_paths: self.table_paths,
             file_schema: self.file_schema,
             options: Some(listing_options),
             schema_adapter_factory: self.schema_adapter_factory,
             schema_policy: self.schema_policy,
-        }
+        };
+        // Ensure schema policy is consistent with schema
+        config.ensure_schema_policy()
     }
 
     /// Add `schema_adapter_factory` to [`ListingTableConfig`]
     pub fn with_schema_adapter_factory(
         self,
         schema_adapter_factory: Arc<dyn SchemaAdapterFactory>,
     ) -> Self {
-        Self {
+        let config = Self {
             table_paths: self.table_paths,
             file_schema: self.file_schema,
             options: self.options,
             schema_adapter_factory: Some(schema_adapter_factory),
             schema_policy: self.schema_policy,
-        }
+        };
+        // Ensure schema policy is consistent with schema
+        config.ensure_schema_policy()
     }
 
     /// Set the schema policy for the [`ListingTable`]
     pub fn with_schema_policy(self, schema_policy: SchemaPolicy) -> Self {
-        Self {
-            table_paths: self.table_paths,
-            file_schema: self.file_schema,
-            options: self.options,
-            schema_adapter_factory: self.schema_adapter_factory,
-            schema_policy: Some(schema_policy),
+        // If we have a file_schema, enforce Fixed policy with that schema
+        if let Some(schema) = &self.file_schema {
+            match schema_policy {
+                SchemaPolicy::Fixed(_) => {
+                    // Use provided Fixed policy with our schema
+                    Self {
+                        table_paths: self.table_paths,
+                        file_schema: self.file_schema,
+                        options: self.options,
+                        schema_adapter_factory: self.schema_adapter_factory,
+                        schema_policy: Some(SchemaPolicy::Fixed(schema.clone())),
+                    }
+                }
+                _ => {
+                    // When file_schema is present, always use Fixed policy regardless of requested policy
+                    Self {
+                        table_paths: self.table_paths,
+                        file_schema: self.file_schema,
+                        options: self.options,
+                        schema_adapter_factory: self.schema_adapter_factory,
+                        schema_policy: Some(SchemaPolicy::Fixed(schema.clone())),
+                    }
+                }
+            }
+        } else {
+            // No file_schema, so accept the requested policy
+            Self {
+                table_paths: self.table_paths,
+                file_schema: self.file_schema,
+                options: self.options,
+                schema_adapter_factory: self.schema_adapter_factory,
+                schema_policy: Some(schema_policy),
+            }
+        }
+    }
+
+    /// Ensures that if a file schema is provided, the schema policy is set to Fixed
+    fn ensure_schema_policy(self) -> Self {
+        if let Some(schema) = &self.file_schema {
+            match self.schema_policy {
+                Some(SchemaPolicy::Fixed(_)) => {
+                    // Already have a Fixed policy, but make sure it's using our schema
+                    Self {
+                        table_paths: self.table_paths,
+                        file_schema: self.file_schema,
+                        options: self.options,
+                        schema_adapter_factory: self.schema_adapter_factory,
+                        schema_policy: Some(SchemaPolicy::Fixed(schema.clone())),
+                    }
+                }
+                _ => {
+                    // If schema is present but policy isn't Fixed, make it Fixed
+                    Self {
+                        table_paths: self.table_paths,
+                        file_schema: self.file_schema,
+                        options: self.options,
+                        schema_adapter_factory: self.schema_adapter_factory,
+                        schema_policy: Some(SchemaPolicy::Fixed(schema.clone())),
+                    }
+                }
+            }
+        } else {
+            // No schema, no change needed
+            self
         }
     }
 
@@ -270,13 +334,16 @@ impl ListingTableConfig {
             .with_file_extension(listing_file_extension)
             .with_target_partitions(state.config().target_partitions());
 
-        Ok(Self {
+        let config = Self {
             table_paths: self.table_paths,
             file_schema: self.file_schema,
             options: Some(listing_options),
             schema_adapter_factory: self.schema_adapter_factory,
             schema_policy: self.schema_policy,
-        })
+        };
+
+        // Ensure schema policy is consistent with schema
+        Ok(config.ensure_schema_policy())
     }
 
     /// Infer the [`SchemaRef`] based on `table_path`s.
@@ -289,6 +356,17 @@ impl ListingTableConfig {
     pub async fn infer_schema(self, state: &dyn Session) -> Result<Self> {
         match self.options {
             Some(options) => {
+                // If a file schema is already provided, use it and ensure policy is Fixed
+                if let Some(schema) = self.file_schema {
+                    return Ok(Self {
+                        table_paths: self.table_paths,
+                        file_schema: Some(schema.clone()),
+                        options: Some(options),
+                        schema_adapter_factory: self.schema_adapter_factory,
+                        schema_policy: Some(SchemaPolicy::Fixed(schema)),
+                    });
+                }
+
                 let schema = match self.schema_policy.unwrap_or(SchemaPolicy::First) {
                     // For Fixed policy, we already have a schema
                     SchemaPolicy::Fixed(schema) => schema,
@@ -312,6 +390,7 @@ impl ListingTableConfig {
                     }
                 };
 
+                // When we infer a schema, keep the current policy
                 Ok(Self {
                     table_paths: self.table_paths,
                     file_schema: Some(schema),
@@ -354,13 +433,16 @@ impl ListingTableConfig {
                     })
                     .collect::<Vec<_>>();
                 let options = options.with_table_partition_cols(partitions);
-                Ok(Self {
+                let config = Self {
                     table_paths: self.table_paths,
                     file_schema: self.file_schema,
                     options: Some(options),
                     schema_adapter_factory: self.schema_adapter_factory,
                     schema_policy: self.schema_policy,
-                })
+                };
+
+                // Ensure schema policy is consistent with schema
+                Ok(config.ensure_schema_policy())
             }
             None => config_err!("No `ListingOptions` set for inferring schema"),
         }
@@ -855,11 +937,19 @@ impl ListingTable {
     ///
     /// See documentation and example on [`ListingTable`] and [`ListingTableConfig`]
     pub fn try_new(config: ListingTableConfig) -> Result<Self> {
-        let file_schema = match config.schema_policy {
-            Some(SchemaPolicy::Fixed(schema)) => schema,
-            _ => config
-                .file_schema
-                .ok_or_else(|| DataFusionError::Internal("No schema provided.".into()))?,
+        // If file_schema is provided but schema_policy is not Fixed, apply the rule
+        let file_schema = match (config.file_schema, &config.schema_policy) {
+            // If we have an explicit Fixed policy, use it
+            (_, Some(SchemaPolicy::Fixed(schema))) => schema.clone(),
+
+            // If we have a file_schema but no fixed policy, use the file_schema
+            // Note: We don't change schema_policy here because that's done during config creation
+            (Some(schema), _) => schema.clone(),
+
+            // No schema provided - error condition
+            (None, _) => {
+                return Err(DataFusionError::Internal("No schema provided.".into()));
+            }
         };
 
         let options = config.options.ok_or_else(|| {