| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -604,7 +604,7 @@ def _upload_local_data(self, local_table: local_data.ManagedArrowTable): | |||
| 604 | 604 | # Might be better as a queue and a worker thread | |
| 605 | 605 | with self._upload_lock: | |
| 606 | 606 | if local_table not in self.cache._uploaded_local_data: | |
| 607 | - uploaded = self.loader.load_data( | ||
| 607 | + uploaded = self.loader.load_data_or_write_data( | ||
| 608 | 608 | local_table, bigframes.core.guid.generate_guid() | |
| 609 | 609 | ) | |
| 610 | 610 | self.cache.cache_remote_replacement(local_table, uploaded) | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -350,16 +350,38 @@ def read_managed_data( | |||
| 350 | 350 | session=self._session, | |
| 351 | 351 | ) | |
| 352 | 352 | ||
| 353 | + def load_data_or_write_data( | ||
| 354 | + self, | ||
| 355 | + data: local_data.ManagedArrowTable, | ||
| 356 | + offsets_col: str, | ||
| 357 | + ) -> bq_data.BigqueryDataSource: | ||
| 358 | + """Write local data into BigQuery using the local API if possible, | ||
| 359 | + otherwise use the write API.""" | ||
| 360 | + can_load = all( | ||
| 361 | + _is_dtype_can_load(item.column, item.dtype) for item in data.schema.items | ||
| 362 | + ) | ||
| 363 | + if can_load: | ||
| 364 | + return self.load_data(data, offsets_col=offsets_col) | ||
| 365 | + else: | ||
| 366 | + return self.write_data(data, offsets_col=offsets_col) | ||
| 367 | + | ||
| 353 | 368 | def load_data( | |
| 354 | 369 | self, | |
| 355 | 370 | data: local_data.ManagedArrowTable, | |
| 356 | 371 | offsets_col: str, | |
| 357 | 372 | ) -> bq_data.BigqueryDataSource: | |
| 358 | 373 | """Load managed data into bigquery""" | |
| 359 | - | ||
| 360 | - # JSON support incomplete | ||
| 361 | - for item in data.schema.items: | ||
| 362 | - _validate_dtype_can_load(item.column, item.dtype) | ||
| 374 | + cannot_load_columns = { | ||
| 375 | + item.column: item.dtype | ||
| 376 | + for item in data.schema.items | ||
| 377 | + if not _is_dtype_can_load(item.column, item.dtype) | ||
| 378 | + } | ||
| 379 | + | ||
| 380 | + if cannot_load_columns: | ||
| 381 | + raise NotImplementedError( | ||
| 382 | + f"Nested JSON types are currently unsupported for BigQuery Load API. " | ||
| 383 | + f"Unsupported columns: {cannot_load_columns}. {constants.FEEDBACK_LINK}" | ||
| 384 | + ) | ||
| 363 | 385 | ||
| 364 | 386 | schema_w_offsets = data.schema.append( | |
| 365 | 387 | schemata.SchemaItem(offsets_col, bigframes.dtypes.INT_DTYPE) | |
@@ -1474,31 +1496,27 @@ def _transform_read_gbq_configuration(configuration: Optional[dict]) -> dict: | |||
| 1474 | 1496 | return configuration | |
| 1475 | 1497 | ||
| 1476 | 1498 | ||
| 1477 | - def _validate_dtype_can_load(name: str, column_type: bigframes.dtypes.Dtype): | ||
| 1499 | + def _is_dtype_can_load(name: str, column_type: bigframes.dtypes.Dtype) -> bool: | ||
| 1478 | 1500 | """ | |
| 1479 | 1501 | Determines whether a datatype is supported by bq load jobs. | |
| 1480 | 1502 | ||
| 1481 | 1503 | Due to a BigQuery IO limitation with loading JSON from Parquet files (b/374784249), | |
| 1482 | 1504 | we're using a workaround: storing JSON as strings and then parsing them into JSON | |
| 1483 | 1505 | objects. | |
| 1484 | 1506 | TODO(b/395912450): Remove workaround solution once b/374784249 got resolved. | |
| 1485 | - | ||
| 1486 | - Raises: | ||
| 1487 | - NotImplementedError: Type is not yet supported by load jobs. | ||
| 1488 | 1507 | """ | |
| 1489 | 1508 | # we can handle top-level json, but not nested yet through string conversion | |
| 1490 | 1509 | if column_type == bigframes.dtypes.JSON_DTYPE: | |
| 1491 | - return | ||
| 1510 | + return True | ||
| 1492 | 1511 | ||
| 1493 | 1512 | if isinstance( | |
| 1494 | 1513 | column_type, pandas.ArrowDtype | |
| 1495 | 1514 | ) and bigframes.dtypes.contains_db_dtypes_json_arrow_type( | |
| 1496 | 1515 | column_type.pyarrow_dtype | |
| 1497 | 1516 | ): | |
| 1498 | - raise NotImplementedError( | ||
| 1499 | - f"Nested JSON types, found in column `{name}`: `{column_type}`', " | ||
| 1500 | - f"are currently unsupported for upload. {constants.FEEDBACK_LINK}" | ||
| 1501 | - ) | ||
| 1517 | + return False | ||
| 1518 | + | ||
| 1519 | + return True | ||
| 1502 | 1520 | ||
| 1503 | 1521 | ||
| 1504 | 1522 | # itertools.batched not available in python <3.12, so we use this instead | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -1092,7 +1092,9 @@ def test_read_pandas_w_nested_json_fails(session, write_engine): | |||
| 1092 | 1092 | pa.list_(pa.struct([("json_field", bigframes.dtypes.JSON_ARROW_TYPE)])) | |
| 1093 | 1093 | ), | |
| 1094 | 1094 | ) | |
| 1095 | - with pytest.raises(NotImplementedError, match="Nested JSON types, found in column"): | ||
| 1095 | + with pytest.raises( | ||
| 1096 | + NotImplementedError, match="Nested JSON types are currently unsupported" | ||
| 1097 | + ): | ||
| 1096 | 1098 | session.read_pandas(pd_s, write_engine=write_engine) | |
| 1097 | 1099 | ||
| 1098 | 1100 | ||
@@ -1178,7 +1180,9 @@ def test_read_pandas_w_nested_json_index_fails(session, write_engine): | |||
| 1178 | 1180 | pa.list_(pa.struct([("json_field", bigframes.dtypes.JSON_ARROW_TYPE)])) | |
| 1179 | 1181 | ), | |
| 1180 | 1182 | ) | |
| 1181 | - with pytest.raises(NotImplementedError, match="Nested JSON types, found in"): | ||
| 1183 | + with pytest.raises( | ||
| 1184 | + NotImplementedError, match="Nested JSON types are currently unsupported" | ||
| 1185 | + ): | ||
| 1182 | 1186 | session.read_pandas(pd_idx, write_engine=write_engine) | |
| 1183 | 1187 | ||
| 1184 | 1188 | ||
| Back | FazBrowse Home | New Git URL |
0 commit comments