Repository: apache/airflow
File: providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py (Python)

Code before:
            if output_format == "csv":
                with open(local_path, "w", newline="") as file:
                    if self._csv_params:
                        csv_params = self._csv_params.copy()
                    else:
                        csv_params = {}
                    write_header = csv_params.pop("header", True)
                    writer = csv.DictWriter(file, fieldnames=field_names, **csv_params)
                    if write_header:
                        writer.writeheader()
                    for row in last_results:
                        writer.writerow(row._asdict())
            elif output_format == "json":
                with open(local_path, "w") as file:
                    file.write(json.dumps([row._asdict() for row in last_results]))
            elif output_format == "jsonl":
                with open(local_path, "w") as file:
                    for row in last_results:
                        file.write(json.dumps(row._asdict()))
                        file.write("\n")
            elif output_format == "parquet":
                self._write_parquet(local_path, field_names, last_results)
            elif output_format == "avro":
                self._write_avro(local_path, field_names, last_results)
            else:
                raise AirflowException(f"Unsupported output format: '{self._output_format}'")

            if is_gcs:
                self._upload_to_gcs(local_path, self._output_path)
        finally:
            if is_gcs and os.path.exists(local_path):
                os.unlink(local_path)

        return list(zip(descriptions, results))


COPY_INTO_APPROVED_FORMATS = ["CSV", "JSON", "AVRO", "ORC", "PARQUET", "TEXT", "BINARYFILE"]


class DatabricksCopyIntoOperator(BaseOperator):
    """
    Executes COPY INTO command in a Databricks SQL endpoint or a Databricks cluster.

    COPY INTO command is constructed from individual pieces, that are described in
    `documentation <https://docs.databricks.com/sql/language-manual/delta-copy-into.html>`_.

    .. seealso::
        For more information on how to use this operator, take a look at the guide:
        :ref:`howto/operator:DatabricksSqlCopyIntoOperator`

    :param table_name: Required name of the table. (templated)

Review comment (suggestion):
```suggestion
                raise ValueError(f"Unsupported output format: '{self._output_format}'")
```

Code after:
            local_path = self._output_path

        try:
            output_format = self._output_format.lower()
            if output_format == "csv":
                with open(local_path, "w", newline="") as file:
                    if self._csv_params:
                        csv_params = self._csv_params.copy()
                    else:
                        csv_params = {}
                    write_header = csv_params.pop("header", True)
                    writer = csv.DictWriter(file, fieldnames=field_names, **csv_params)
                    if write_header:
                        writer.writeheader()
                    for row in last_results:
                        writer.writerow(row._asdict())
            elif output_format == "json":
                with open(local_path, "w") as file:
                    file.write(json.dumps([row._asdict() for row in last_results]))
            elif output_format == "jsonl":
                with open(local_path, "w") as file:
                    for row in last_results:
                        file.write(json.dumps(row._asdict()))
                        file.write("\n")
            elif output_format == "parquet":
                self._write_parquet(local_path, field_names, last_results)
            elif output_format == "avro":
                self._write_avro(local_path, field_names, last_results)
            else:
                raise ValueError(f"Unsupported output format: '{self._output_format}'")

            if self._is_gcs_output:
                self._upload_to_gcs(local_path, self._output_path)
        finally:
            if self._is_gcs_output and os.path.exists(local_path):
                os.unlink(local_path)

        return list(zip(descriptions, results))


COPY_INTO_APPROVED_FORMATS = ["CSV", "JSON", "AVRO", "ORC", "PARQUET", "TEXT", "BINARYFILE"]


class DatabricksCopyIntoOperator(BaseOperator):
    """
    Executes COPY INTO command in a Databricks SQL endpoint or a Databricks cluster.

    COPY INTO command is constructed from individual pieces, that are described in
    `documentation <https://docs.databricks.com/sql/language-manual/delta-copy-into.html>`_.

    .. seealso::