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

Code before:
        }
        return DatabricksSqlHook(self.databricks_conn_id, **hook_params)

    def get_db_hook(self) -> DatabricksSqlHook:
        return self._hook

    def _should_run_output_processing(self) -> bool:
        return self.do_xcom_push or bool(self._output_path)

    def _is_gcs_path(self, path: str) -> bool:
        """Check if the path is a GCS URI."""
        return path.startswith("gs://")

    def _parse_gcs_path(self, path: str) -> tuple[str, str]:
        """Parse a GCS URI into bucket and object name."""
        parsed = urlparse(path)
        bucket = parsed.netloc
        object_name = parsed.path.lstrip("/")
        return bucket, object_name

    def _upload_to_gcs(self, local_path: str, gcs_path: str) -> None:
        """Upload a local file to GCS."""
        try:
            from airflow.providers.google.cloud.hooks.gcs import GCSHook
        except ImportError:
            raise AirflowException(
                "The 'apache-airflow-providers-google' package is required for GCS output. "
                "Install it with: pip install apache-airflow-providers-google"
            )

        bucket, object_name = self._parse_gcs_path(gcs_path)
        hook = GCSHook(
            gcp_conn_id=self._gcp_conn_id,
            impersonation_chain=self._gcs_impersonation_chain,
        )
        hook.upload(
            bucket_name=bucket,
            object_name=object_name,
            filename=local_path,
        )
        self.log.info("Uploaded output to %s", gcs_path)

    def _write_parquet(self, file_path: str, field_names: list[str], rows: list[Any]) -> None:
        """Write data to a Parquet file."""
        import pyarrow as pa
        import pyarrow.parquet as pq

        data: dict[str, list] = {name: [] for name in field_names}
        for row in rows:
            row_dict = row._asdict()
            for name in field_names:

Review comment (suggestion):
```suggestion
            raise AirflowOptionalProviderFeatureException(
```

Code after:
            "schema": self.schema,
            "caller": "DatabricksSqlOperator",
            **self.client_parameters,
            **self.hook_params,
        }
        return DatabricksSqlHook(self.databricks_conn_id, **hook_params)

    def get_db_hook(self) -> DatabricksSqlHook:
        return self._hook

    def _should_run_output_processing(self) -> bool:
        return self.do_xcom_push or bool(self._output_path)

    @property
    def _is_gcs_output(self) -> bool:
        """Check if the output path is a GCS URI."""
        return self._output_path.startswith("gs://") if self._output_path else False

    def _parse_gcs_path(self, path: str) -> tuple[str, str]:
        """Parse a GCS URI into bucket and object name."""
        parsed = urlparse(path)
        bucket = parsed.netloc
        object_name = parsed.path.lstrip("/")
        return bucket, object_name

    def _upload_to_gcs(self, local_path: str, gcs_path: str) -> None:
        """Upload a local file to GCS."""
        try:
            from airflow.providers.google.cloud.hooks.gcs import GCSHook
        except ImportError:
            raise AirflowOptionalProviderFeatureException(
                "The 'apache-airflow-providers-google' package is required for GCS output. "
                "Install it with: pip install apache-airflow-providers-google"
            )

        bucket, object_name = self._parse_gcs_path(gcs_path)
        hook = GCSHook(
            gcp_conn_id=self._gcp_conn_id,
            impersonation_chain=self._gcs_impersonation_chain,
        )
        hook.upload(
            bucket_name=bucket,
            object_name=object_name,
            filename=local_path,
        )
        self.log.info("Uploaded output to %s", gcs_path)

    def _write_parquet(self, file_path: str, field_names: list[str], rows: list[Any]) -> None:
        """Write data to a Parquet file."""
        import pyarrow as pa
        import pyarrow.parquet as pq