title: Filtering a spark partitioned table is not working in Pyspark
body: <p>I am using spark 2.3 and have written one dataframe to create hive partitioned table using dataframe writer class method in pyspark.</p>
  
  <pre><code>newdf.coalesce(1).write.format('orc').partitionBy('veh_country').mode("overwrite").saveAsTable('emp.partition_Load_table')
  </code></pre>
  
  <p>Here is my table structure and partitions information.</p>
  
  <pre><code>hive&gt; desc emp.partition_Load_table;
  OK
  veh_code                varchar(17)
  veh_flag                varchar(1)
  veh_model               smallint
  veh_country             varchar(3)
  
  # Partition Information
  # col_name              data_type               comment
  
  veh_country              varchar(3)
  
  hive&gt; show partitions partition_Load_table;
  OK
  veh_country=CHN
  veh_country=USA
  veh_country=RUS
  </code></pre>
  
  <p>Now I am reading this table back in pyspark inside a dataframe.</p>
  
  <pre><code>    df2_data = spark.sql("""
      SELECT * 
      from udb.partition_Load_table
      """);
  
  df2_data.show() --&gt; is working
  </code></pre>
  
  <p>But I am not able to filter it using partition key column</p>
  
  <pre><code>from pyspark.sql.functions import col
  newdf = df2_data.where(col("veh_country")=='CHN')
  </code></pre>
  
  <p>I am getting below error message:</p>
  
  <pre><code>: java.lang.RuntimeException: Caught Hive MetaException attempting to get partition metadata by filter from Hive. 
  You can set the Spark configuration setting spark.sql.hive.manageFilesourcePartitions to false to work around this problem, 
  however this will result in degraded performance. Please report a bug: https://issues.apache.org/jira/browse/SPARK
  Caused by: MetaException(message:Filtering is supported only on partition keys of type string)
  </code></pre>
  
  <p>whereas when I am creating dataframe by specifying the hdfs absolute path of table. filter and where clause is working as expected.</p>
  
  <pre><code>newdataframe = spark.read.format("orc").option("header","false").load("hdfs/path/emp.db/partition_load_table")
  </code></pre>
  
  <p>below is working</p>
  
  <pre><code>newdataframe.where(col("veh_country")=='CHN').show()
  </code></pre>
  
  <p>my question is that why it was not able to filter the dataframe in first place. and also why it's throwing an error message " Filtering is supported only on partition keys of type string " even though my veh_country is defined as string or varchar datatypes.</p>