Add remote URL support for copy insert

- Add is_remote_url() and parse_remote_url() helpers to storage.py
- Add copy_from_url() method to StorageBackend for remote-to-managed copies
- Add source_exists(), source_is_directory(), get_source_size() helpers
- Support s3://, gs://, az://, http://, https:// protocols
- Update spec with Remote URL Support section and examples
- Update object.md with "Inserting from Remote URLs" section
- Update insert.md with remote URL examples
- Add TestRemoteURLSupport test class

Author: claude

docs/src/design/tables/object-type-spec.md

@@ -584,37 +584,70 @@ Each insert stores a separate copy of the file, even if identical content was pr
 
 At insert time, the `object` attribute accepts:
 
-1. **File path** (string or `Path`): Path to an existing file (extension extracted)
-2. **Folder path** (string or `Path`): Path to an existing directory
-3. **Tuple of (ext, stream)**: File-like object with explicit extension
+1. **Local file path** (string or `Path`): Path to an existing local file (extension extracted)
+2. **Local folder path** (string or `Path`): Path to an existing local directory
+3. **Remote URL** (string): URL to remote file or folder (`s3://`, `gs://`, `az://`, `http://`, `https://`)
+4. **Tuple of (ext, stream)**: File-like object with explicit extension
 
 ```python
-# From file path - extension (.dat) extracted from source
+# From local file path - extension (.dat) extracted from source
 Recording.insert1({
     "subject_id": 123,
     "session_id": 45,
     "raw_data": "/local/path/to/recording.dat"
 })
 # Stored as: raw_data_Ax7bQ2kM.dat
 
-# From folder path - no extension
+# From local folder path - no extension
 Recording.insert1({
     "subject_id": 123,
     "session_id": 45,
     "raw_data": "/local/path/to/data_folder/"
 })
 # Stored as: raw_data_pL9nR4wE/
 
+# From remote URL - copies from source to managed storage
+Recording.insert1({
+    "subject_id": 123,
+    "session_id": 45,
+    "raw_data": "s3://source-bucket/path/to/data.dat"
+})
+# Stored as: raw_data_kM3nP2qR.dat
+
+# From remote Zarr store (e.g., collaborator data on GCS)
+Recording.insert1({
+    "subject_id": 123,
+    "session_id": 45,
+    "neural_data": "gs://collaborator-bucket/shared/experiment.zarr"
+})
+# Copied to managed storage as: neural_data_pL9nR4wE.zarr
+
 # From stream with explicit extension
 with open("/local/path/data.bin", "rb") as f:
     Recording.insert1({
         "subject_id": 123,
         "session_id": 45,
         "raw_data": (".bin", f)
     })
-# Stored as: raw_data_kM3nP2qR.bin
+# Stored as: raw_data_xY8zW3vN.bin
 ```
 
+### Remote URL Support
+
+Remote URLs are detected by protocol prefix and handled via fsspec:
+
+| Protocol | Example | Notes |
+|----------|---------|-------|
+| `s3://` | `s3://bucket/path/file.dat` | AWS S3, MinIO |
+| `gs://` | `gs://bucket/path/file.dat` | Google Cloud Storage |
+| `az://` | `az://container/path/file.dat` | Azure Blob Storage |
+| `http://` | `http://server/path/file.dat` | HTTP (read-only source) |
+| `https://` | `https://server/path/file.dat` | HTTPS (read-only source) |
+
+**Authentication**: Remote sources may require credentials. fsspec uses standard credential discovery (environment variables, config files, IAM roles). For cross-cloud copies, ensure credentials are configured for both source and destination.
+
+**Performance note**: For large remote-to-remote copies, data flows through the client. This is acceptable for most use cases but may be slow for very large datasets. Future optimizations could include server-side copy for same-provider transfers.
+
 ### Insert Processing Steps
 
 1. Validate input (file/folder exists, stream is readable)

docs/src/design/tables/object.md

@@ -89,7 +89,7 @@ Note: No `@store` suffix needed—storage is determined by pipeline configuratio
 
 ### Inserting Files
 
-Insert a file by providing its path:
+Insert a file by providing its local path:
 
 ```python
 Recording.insert1({
@@ -113,6 +113,37 @@ Recording.insert1({
 })
 ```
 
+### Inserting from Remote URLs
+
+Insert from cloud storage or HTTP sources—content is copied to managed storage:
+
+```python
+# From S3
+Recording.insert1({
+    "subject_id": 123,
+    "session_id": 45,
+    "raw_data": "s3://source-bucket/path/to/data.dat"
+})
+
+# From Google Cloud Storage (e.g., collaborator data)
+Recording.insert1({
+    "subject_id": 123,
+    "session_id": 45,
+    "neural_data": "gs://collaborator-bucket/shared/experiment.zarr"
+})
+
+# From HTTP/HTTPS
+Recording.insert1({
+    "subject_id": 123,
+    "session_id": 45,
+    "raw_data": "https://example.com/public/data.dat"
+})
+```
+
+Supported protocols: `s3://`, `gs://`, `az://`, `http://`, `https://`
+
+Remote sources may require credentials configured via environment variables or fsspec configuration files.
+
 ### Inserting from Streams
 
 Insert from a file-like object with explicit extension:

docs/src/manipulation/insert.md

@@ -96,24 +96,38 @@ phase_two.Protocol.insert(protocols)
 ## Object attributes
 
 Tables with [`object`](../design/tables/object.md) type attributes can be inserted with
-file paths, folder paths, or streams. The content is automatically copied to object
-storage.
+local file paths, folder paths, remote URLs, or streams. The content is automatically
+copied to object storage.
 
 ```python
-# Insert with file path
+# Insert with local file path
 Recording.insert1({
     "subject_id": 123,
     "session_id": 45,
     "raw_data": "/local/path/to/data.dat"
 })
 
-# Insert with folder path
+# Insert with local folder path
 Recording.insert1({
     "subject_id": 123,
     "session_id": 45,
     "raw_data": "/local/path/to/data_folder/"
 })
 
+# Insert from remote URL (S3, GCS, Azure, HTTP)
+Recording.insert1({
+    "subject_id": 123,
+    "session_id": 45,
+    "raw_data": "s3://source-bucket/path/to/data.dat"
+})
+
+# Insert remote Zarr store (e.g., from collaborator)
+Recording.insert1({
+    "subject_id": 123,
+    "session_id": 45,
+    "neural_data": "gs://collaborator-bucket/shared/experiment.zarr"
+})
+
 # Insert from stream with explicit extension
 with open("/path/to/data.bin", "rb") as f:
     Recording.insert1({
@@ -123,6 +137,8 @@ with open("/path/to/data.bin", "rb") as f:
     })
 ```
 
+Supported remote URL protocols: `s3://`, `gs://`, `az://`, `http://`, `https://`
+
 ### Staged inserts
 
 For large objects like Zarr arrays, use `staged_insert1` to write directly to storage

src/datajoint/storage.py

@@ -22,6 +22,55 @@
 # Characters safe for use in filenames and URLs
 TOKEN_ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"
 
+# Supported remote URL protocols for copy insert
+REMOTE_PROTOCOLS = ("s3://", "gs://", "gcs://", "az://", "abfs://", "http://", "https://")
+
+
+def is_remote_url(path: str) -> bool:
+    """
+    Check if a path is a remote URL.
+
+    Args:
+        path: Path string to check
+
+    Returns:
+        True if path is a remote URL
+    """
+    if not isinstance(path, str):
+        return False
+    return path.lower().startswith(REMOTE_PROTOCOLS)
+
+
+def parse_remote_url(url: str) -> tuple[str, str]:
+    """
+    Parse a remote URL into protocol and path.
+
+    Args:
+        url: Remote URL (e.g., 's3://bucket/path/file.dat')
+
+    Returns:
+        Tuple of (protocol, path) where protocol is fsspec-compatible
+    """
+    url_lower = url.lower()
+
+    # Map URL schemes to fsspec protocols
+    protocol_map = {
+        "s3://": "s3",
+        "gs://": "gcs",
+        "gcs://": "gcs",
+        "az://": "abfs",
+        "abfs://": "abfs",
+        "http://": "http",
+        "https://": "https",
+    }
+
+    for prefix, protocol in protocol_map.items():
+        if url_lower.startswith(prefix):
+            path = url[len(prefix) :]
+            return protocol, path
+
+    raise errors.DataJointError(f"Unsupported remote URL protocol: {url}")
+
 
 def generate_token(length: int = 8) -> str:
     """
@@ -494,6 +543,155 @@ def get_fsmap(self, remote_path: str | PurePosixPath) -> fsspec.FSMap:
         full_path = self._full_path(remote_path)
         return fsspec.FSMap(full_path, self.fs)
 
+    def copy_from_url(self, source_url: str, dest_path: str | PurePosixPath) -> int:
+        """
+        Copy a file from a remote URL to managed storage.
+
+        Args:
+            source_url: Remote URL (s3://, gs://, http://, etc.)
+            dest_path: Destination path in managed storage
+
+        Returns:
+            Size of copied file in bytes
+        """
+        protocol, source_path = parse_remote_url(source_url)
+        full_dest = self._full_path(dest_path)
+
+        logger.debug(f"copy_from_url: {protocol}://{source_path} -> {self.protocol}:{full_dest}")
+
+        # Get source filesystem
+        source_fs = fsspec.filesystem(protocol)
+
+        # Check if source is a directory
+        if source_fs.isdir(source_path):
+            return self._copy_folder_from_url(source_fs, source_path, dest_path)
+
+        # Copy single file
+        if self.protocol == "file":
+            # Download to local destination
+            Path(full_dest).parent.mkdir(parents=True, exist_ok=True)
+            source_fs.get_file(source_path, full_dest)
+            return Path(full_dest).stat().st_size
+        else:
+            # Remote-to-remote copy via streaming
+            with source_fs.open(source_path, "rb") as src:
+                content = src.read()
+            self.fs.pipe_file(full_dest, content)
+            return len(content)
+
+    def _copy_folder_from_url(
+        self, source_fs: fsspec.AbstractFileSystem, source_path: str, dest_path: str | PurePosixPath
+    ) -> dict:
+        """
+        Copy a folder from a remote URL to managed storage.
+
+        Args:
+            source_fs: Source filesystem
+            source_path: Path in source filesystem
+            dest_path: Destination path in managed storage
+
+        Returns:
+            Manifest dict with file list, total_size, and item_count
+        """
+        full_dest = self._full_path(dest_path)
+        logger.debug(f"copy_folder_from_url: {source_path} -> {self.protocol}:{full_dest}")
+
+        # Collect file info for manifest
+        files = []
+        total_size = 0
+
+        # Walk source directory
+        for root, dirs, filenames in source_fs.walk(source_path):
+            for filename in filenames:
+                src_file = f"{root}/{filename}" if root != source_path else f"{source_path}/{filename}"
+                rel_path = src_file[len(source_path) :].lstrip("/")
+                file_size = source_fs.size(src_file)
+                files.append({"path": rel_path, "size": file_size})
+                total_size += file_size
+
+                # Copy file
+                dest_file = f"{full_dest}/{rel_path}"
+                if self.protocol == "file":
+                    Path(dest_file).parent.mkdir(parents=True, exist_ok=True)
+                    source_fs.get_file(src_file, dest_file)
+                else:
+                    with source_fs.open(src_file, "rb") as src:
+                        content = src.read()
+                    self.fs.pipe_file(dest_file, content)
+
+        # Build manifest
+        manifest = {
+            "files": files,
+            "total_size": total_size,
+            "item_count": len(files),
+            "created": datetime.now(timezone.utc).isoformat(),
+        }
+
+        # Write manifest alongside folder
+        manifest_path = f"{dest_path}.manifest.json"
+        self.put_buffer(json.dumps(manifest, indent=2).encode(), manifest_path)
+
+        return manifest
+
+    def source_is_directory(self, source: str) -> bool:
+        """
+        Check if a source path (local or remote URL) is a directory.
+
+        Args:
+            source: Local path or remote URL
+
+        Returns:
+            True if source is a directory
+        """
+        if is_remote_url(source):
+            protocol, path = parse_remote_url(source)
+            source_fs = fsspec.filesystem(protocol)
+            return source_fs.isdir(path)
+        else:
+            return Path(source).is_dir()
+
+    def source_exists(self, source: str) -> bool:
+        """
+        Check if a source path (local or remote URL) exists.
+
+        Args:
+            source: Local path or remote URL
+
+        Returns:
+            True if source exists
+        """
+        if is_remote_url(source):
+            protocol, path = parse_remote_url(source)
+            source_fs = fsspec.filesystem(protocol)
+            return source_fs.exists(path)
+        else:
+            return Path(source).exists()
+
+    def get_source_size(self, source: str) -> int | None:
+        """
+        Get the size of a source file (local or remote URL).
+
+        Args:
+            source: Local path or remote URL
+
+        Returns:
+            Size in bytes, or None if directory or cannot determine
+        """
+        try:
+            if is_remote_url(source):
+                protocol, path = parse_remote_url(source)
+                source_fs = fsspec.filesystem(protocol)
+                if source_fs.isdir(path):
+                    return None
+                return source_fs.size(path)
+            else:
+                p = Path(source)
+                if p.is_dir():
+                    return None
+                return p.stat().st_size
+        except Exception:
+            return None
+
 
 STORE_METADATA_FILENAME = "datajoint_store.json"