rm downcast (apache#15575)

Author: jayzhan211

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

@@ -875,15 +875,13 @@ impl TableProvider for ListingTable {
  filters.iter().cloned().partition(|filter| {
  can_be_evaluted_for_partition_pruning(&table_partition_col_names, filter)
  });
- // TODO (https://github.com/apache/datafusion/issues/11600) remove downcast_ref from here?
- let session_state = state.as_any().downcast_ref::<SessionState>().unwrap();
 
  // We should not limit the number of partitioned files to scan if there are filters and limit
  // at the same time. This is because the limit should be applied after the filters are applied.
  let statistic_file_limit = if filters.is_empty() { limit } else { None };
 
  let (mut partitioned_file_lists, statistics) = self
- .list_files_for_scan(session_state, &partition_filters, statistic_file_limit)
+ .list_files_for_scan(state, &partition_filters, statistic_file_limit)
  .await?;
 
  // if no files need to be read, return an `EmptyExec`
@@ -942,7 +940,7 @@ impl TableProvider for ListingTable {
  self.options
  .format
  .create_physical_plan(
- session_state,
+ state,
  FileScanConfigBuilder::new(
  object_store_url,
  Arc::clone(&self.file_schema),
@@ -1022,10 +1020,8 @@ impl TableProvider for ListingTable {
  // Get the object store for the table path.
  let store = state.runtime_env().object_store(table_path)?;
 
- // TODO (https://github.com/apache/datafusion/issues/11600) remove downcast_ref from here?
- let session_state = state.as_any().downcast_ref::<SessionState>().unwrap();
  let file_list_stream = pruned_partition_list(
- session_state,
+ state,
  store.as_ref(),
  table_path,
  &[],
@@ -1073,7 +1069,7 @@ impl TableProvider for ListingTable {
 
  self.options()
  .format
- .create_writer_physical_plan(input, session_state, config, order_requirements)
+ .create_writer_physical_plan(input, state, config, order_requirements)
  .await
  }