chore: add experimental series.str.to_blob function (#1224)

* chore: add experimental series.str.to_blob function

* extract dtype

* update docs

* fix connection

Author: GarrettWu

bigframes/core/compile/scalar_op_compiler.py

@@ -1735,6 +1735,12 @@ def binary_remote_function_op_impl(
     return x_transformed
 
 
+# Blob Ops
+@scalar_op_compiler.register_binary_op(ops.obj_make_ref_op)
+def obj_make_ref_op(x: ibis_types.Value, y: ibis_types.Value):
+    return obj_make_ref(uri=x, authorizer=y)
+
+
 # Ternary Operations
 @scalar_op_compiler.register_ternary_op(ops.where_op)
 def where_op(
@@ -1906,3 +1912,8 @@ def vector_distance(vector1, vector2, type: str) -> ibis_dtypes.Float64:  # type
 @ibis_udf.scalar.builtin(name="OBJ.FETCH_METADATA")
 def obj_fetch_metadata(obj_ref: _OBJ_REF_IBIS_DTYPE) -> _OBJ_REF_IBIS_DTYPE:  # type: ignore
     """Fetch metadata from ObjectRef Struct."""
+
+
+@ibis_udf.scalar.builtin(name="OBJ.MAKE_REF")
+def obj_make_ref(uri: str, authorizer: str) -> _OBJ_REF_IBIS_DTYPE:  # type: ignore
+    """Make ObjectRef Struct from uri and connection."""

bigframes/dtypes.py

@@ -57,6 +57,28 @@
 GEO_DTYPE = gpd.array.GeometryDtype()
 # JSON
 JSON_DTYPE = pd.ArrowDtype(pa.large_string())
+OBJ_REF_DTYPE = pd.ArrowDtype(
+    pa.struct(
+        (
+            pa.field(
+                "uri",
+                pa.string(),
+            ),
+            pa.field(
+                "version",
+                pa.string(),
+            ),
+            pa.field(
+                "authorizer",
+                pa.string(),
+            ),
+            pa.field(
+                "details",
+                pa.large_string(),  # JSON
+            ),
+        )
+    )
+)
 
 # Used when storing Null expressions
 DEFAULT_DTYPE = FLOAT_DTYPE

bigframes/operations/__init__.py

@@ -893,6 +893,21 @@ def output_type(self, *input_types):
         return left_type
 
 
+## Blob Ops
+@dataclasses.dataclass(frozen=True)
+class ObjMakeRef(BinaryOp):
+    name: typing.ClassVar[str] = "obj.make_ref"
+
+    def output_type(self, *input_types):
+        if not all(map(dtypes.is_string_like, input_types)):
+            raise TypeError("obj.make_ref requires string-like arguments")
+
+        return dtypes.OBJ_REF_DTYPE
+
+
+obj_make_ref_op = ObjMakeRef()
+
+
 # Ternary Ops
 @dataclasses.dataclass(frozen=True)
 class WhereOp(TernaryOp):

bigframes/operations/blob.py

@@ -27,6 +27,13 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
     def metadata(self):
+        """Retrive the metadata of the Blob.
+
+        .. note::
+            BigFrames Blob is still under experiments. It may not work and subject to change in the future.
+
+        Returns:
+            JSON: metadata of the Blob. Contains fields: content_type, md5_hash, size and updated(time)."""
         details_json = self._apply_unary_op(ops.obj_fetch_metadata_op).struct.field(
             "details"
         )

bigframes/operations/strings.py

@@ -20,6 +20,7 @@
 import bigframes_vendored.constants as constants
 import bigframes_vendored.pandas.core.strings.accessor as vendorstr
 
+from bigframes import clients
 from bigframes.core import log_adapter
 import bigframes.dataframe as df
 import bigframes.operations as ops
@@ -284,6 +285,35 @@ def cat(
     ) -> series.Series:
         return self._apply_binary_op(others, ops.strconcat_op, alignment=join)
 
+    def to_blob(self, connection: Optional[str] = None) -> series.Series:
+        """Create a BigFrames Blob series from a series of URIs.
+
+        .. note::
+            BigFrames Blob is still under experiments. It may not work and subject to change in the future.
+
+
+        Args:
+            connection (str or None, default None):
+                Connection to connect with remote service. str of the format <PROJECT_NUMBER/PROJECT_ID>.<LOCATION>.<CONNECTION_ID>.
+                If None, use default connection in session context. BigQuery DataFrame will try to create the connection and attach
+                permission if the connection isn't fully set up.
+
+        Returns:
+            bigframes.series.Series: Blob Series.
+
+        """
+        if not bigframes.options.experiments.blob:
+            raise NotImplementedError()
+
+        session = self._block.session
+        connection = connection or session._bq_connection
+        connection = clients.resolve_full_bq_connection_name(
+            connection,
+            default_project=session._project,
+            default_location=session._location,
+        )
+        return self._apply_binary_op(connection, ops.obj_make_ref_op)
+
 
 def _parse_flags(flags: int) -> Optional[str]:
     re2flags = []