### Description

Add the ability to partition parquet files by columns. Right now you can partition files only by size.

### Use case/motivation

_No response_

### Related issues

_No response_

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

"""Configure a csv writer with the file_handle and write schema
        as headers for the new file.
        """
        csv_writer = csv.writer(file_handle, encoding="utf-8", delimiter=self.field_delimiter)
        csv_writer.writerow(schema)
        return csv_writer
    def _configure_parquet_file(self, file_handle, parquet_schema):
        parquet_writer = pq.ParquetWriter(file_handle.name, parquet_schema)
        return parquet_writer
    def _convert_parquet_schema(self, cursor):
        type_map = {
            "INTEGER": pa.int64(),
            "FLOAT": pa.float64(),
            "NUMERIC": pa.float64(),
            "BIGNUMERIC": pa.float64(),
            "BOOL": pa.bool_(),
            "STRING": pa.string(),
            "BYTES": pa.binary(),
            "DATE": pa.date32(),
            "DATETIME": pa.date64(),
            "TIMESTAMP": pa.timestamp("s"),
        }
        columns = [field[0] for field in cursor.description]
        bq_fields = [self.field_to_bigquery(field) for field in cursor.description]
        bq_types = [bq_field.get("type") if bq_field is not None else None for bq_field in bq_fields]
        pq_types = [type_map.get(bq_type, pa.string()) for bq_type in bq_types]
        parquet_schema = pa.schema(zip(columns, pq_types))
        return parquet_schema
    @abc.abstractmethod
    def query(self):