fix: to_gbq may swap data columns when replace table (#2532)

Fixes#2502

Author: chelsea-lin

bigframes/session/bq_caching_executor.py

@@ -247,7 +247,7 @@ def _export_gbq(
  )
 sql=compiled.sql
 
-if (existing_tableisnotNone) and_if_schema_match(
+if (existing_tableisnotNone) and_is_schema_match(
 existing_table.schema, array_value.schema
  ):
 # b/409086472: Uses DML for table appends and replacements to avoid
@@ -690,16 +690,16 @@ def _result_schema(
  )
 
 
-def_if_schema_match(
-table_schema: Tuple[bigquery.SchemaField, ...], schema: schemata.ArraySchema
+def_is_schema_match(
+table_schema: Tuple[bigquery.SchemaField, ...],
+schema: schemata.ArraySchema,
 ) ->bool:
 iflen(table_schema) !=len(schema.items):
 returnFalse
-forfieldintable_schema:
-iffield.namenotinschema.names:
+forfield, schema_iteminzip(table_schema, schema.items):
+iffield.name!=schema_item.column:
 returnFalse
-ifbigframes.dtypes.convert_schema_field(field)[1] !=schema.get_type(
-field.name
- ):
+_, field_dtype=bigframes.dtypes.convert_schema_field(field)
+iffield_dtype!=schema_item.dtype:
 returnFalse
 returnTrue

tests/system/small/test_dataframe_io.py

@@ -631,6 +631,13 @@ def test_to_gbq_if_exists_is_replace(scalars_dfs, dataset_id):
 assertlen(gcs_df) ==len(scalars_pandas_df)
 pd.testing.assert_index_equal(gcs_df.columns, scalars_pandas_df.columns)
 
+# When replacing a table with same schema but different column order
+reordered_df=scalars_df[scalars_df.columns[::-1]]
+reordered_df.to_gbq(destination_table, if_exists="replace")
+gcs_df=pandas_gbq.read_gbq(destination_table, index_col="rowindex")
+assertlen(gcs_df) ==len(scalars_pandas_df)
+pd.testing.assert_index_equal(gcs_df.columns, reordered_df.columns)
+
 # When replacing a table with different schema
 partitial_scalars_df=scalars_df.drop(columns=["string_col"])
 partitial_scalars_df.to_gbq(destination_table, if_exists="replace")