def METHOD_NAME(self, task_instance):
    """
    Collect the input, output, job and run facets for append operator
    """
    from astro.lineage import (
        BaseFacet,
        DataQualityMetricsInputDatasetFacet,
        DataSourceDatasetFacet,
        OpenlineageDataset,
        OperatorLineage,
        OutputStatisticsOutputDatasetFacet,
        SchemaDatasetFacet,
        SchemaField,
        SqlJobFacet,
    )
    from astro.lineage.facets import TableDatasetFacet
    # TODO: remove pushing to XCom once we update the airflow version.
    append_query = task_instance.xcom_pull(task_ids=task_instance.task_id, key="append_query")
    source_table_rows = self.source_table.row_count
    input_dataset: list[OpenlineageDataset] = []
    output_dataset: list[OpenlineageDataset] = []
    if self.source_table.openlineage_emit_temp_table_event():
        input_dataset = [
            OpenlineageDataset(
                namespace=self.source_table.openlineage_dataset_namespace(),
                name=self.source_table.openlineage_dataset_name(),
                facets={
                    "input_table_facet": TableDatasetFacet(
                        table_name=self.source_table.name,
                        source_table_rows=source_table_rows,
                        columns=self.columns,
                        metadata=self.source_table.metadata,
                    ),
                    "schema": SchemaDatasetFacet(
                        fields=[
                            SchemaField(
                                name=self.source_table.metadata.schema,
                                type=self.source_table.metadata.database,
                            )
                        ]
                    ),
                    "dataSource": DataSourceDatasetFacet(
                        name=self.source_table.name, uri=self.source_table.openlineage_dataset_uri()
                    ),
                    "dataQualityMetrics": DataQualityMetricsInputDatasetFacet(
                        rowCount=self.source_table.row_count, columnMetrics={}
                    ),
                },
            )
        ]
    if self.source_table.openlineage_emit_temp_table_event():
        output_dataset = [
            OpenlineageDataset(
                namespace=self.target_table.openlineage_dataset_namespace(),
                name=self.target_table.openlineage_dataset_name(),
                facets={
                    "output_table_facet": TableDatasetFacet(
                        table_name=self.target_table.name,
                        columns=self.columns,
                        source_table_rows=source_table_rows,
                        metadata=self.target_table.metadata,
                    ),
                    "outputStatistics": OutputStatisticsOutputDatasetFacet(
                        rowCount=self.target_table.row_count
                    ),
                    "dataSource": DataSourceDatasetFacet(
                        name=self.target_table.name, uri=self.target_table.openlineage_dataset_uri()
                    ),
                    "dataQualityMetrics": DataQualityMetricsInputDatasetFacet(
                        rowCount=self.target_table.row_count, columnMetrics={}
                    ),
                },
            )
        ]
    run_facets: dict[str, BaseFacet] = {}
    job_facets: dict[str, BaseFacet] = {"sql": SqlJobFacet(query=str(append_query))}
    return OperatorLineage(
        inputs=input_dataset, outputs=output_dataset, run_facets=run_facets, job_facets=job_facets
    )