### 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)

"""Execute DBAPI query."""
    @abc.abstractmethod
    def field_to_bigquery(self, field) -> dict[str, str]:
        """Convert a DBAPI field to BigQuery schema format."""
    @abc.abstractmethod
    def convert_type(self, value, schema_type, **kwargs):
        """Convert a value from DBAPI to output-friendly formats."""
    def _get_col_type_dict(self):
        """Return a dict of column name and column type based on self.schema if not None."""
        schema = []
        if isinstance(self.schema, str):
            schema = json.loads(self.schema)
        elif isinstance(self.schema, list):
            schema = self.schema
        elif self.schema is not None:
            self.log.warning("Using default schema due to unexpected type. Should be a string or list.")
        col_type_dict = {}
        try:
            col_type_dict = {col["name"]: col["type"] for col in schema}
        except KeyError:
            self.log.warning(
                "Using default schema due to missing name or type. Please "
                "refer to: https://cloud.google.com/bigquery/docs/schemas"
                "#specifying_a_json_schema_file"
            )
        return col_type_dict
    def _write_local_schema_file(self, cursor):