Add tests with adding columns + updating rows, with and without partitions, for both snapshot and CDF queries

Author: PatrickJin-db

client/src/test/scala/io/delta/sharing/client/DeltaSharingRestClientSuite.scala

@@ -183,6 +183,8 @@ class DeltaSharingRestClientSuite extends DeltaSharingIntegrationTest {
         Table(name = "deletion_vectors_with_dvs_dv_property_on", schema = "default", share = "share8"),
         Table(name = "dv_and_cm_table", schema = "default", share = "share8"),
         Table(name = "timestampntz_cdf_table", schema = "default", share = "share8")
+        Table(name = "add_columns_non_partitioned_cdf", schema = "default", share = "share8"),
+        Table(name = "add_columns_partitioned_cdf", schema = "default", share = "share8")
       )
       assert(expected == client.listAllTables().toSet)
     } finally {

python/delta_sharing/reader.py

@@ -382,17 +382,24 @@ def table_changes_to_pandas(self, cdfOptions: CdfOptions) -> pd.DataFrame:
         response = self._rest_client.list_table_changes(self._table, cdfOptions)
 
         schema_json = loads(response.metadata.schema_string)
+        converters = to_converters(schema_json)
+        schema_with_cdf = self._add_special_cdf_schema(schema_json)
 
         if len(response.actions) == 0:
-            return get_empty_table(self._add_special_cdf_schema(schema_json))
+            return get_empty_table(schema_with_cdf)
 
-        converters = to_converters(schema_json)
         pdfs = []
         for action in response.actions:
             pdf = DeltaSharingReader._to_pandas(action, converters, True, None)
             pdfs.append(pdf)
 
-        return pd.concat(pdfs, axis=0, ignore_index=True, copy=False)
+        merged = pd.concat(pdfs, axis=0, ignore_index=True, copy=False)
+
+        col_map = {}
+        for col in merged.columns:
+            col_map[col.lower()] = col
+
+        return merged[[col_map[field["name"].lower()] for field in schema_with_cdf["fields"]]]
 
     def _copy(
         self,