LanceDB - Remove Orphaned Chunks

dlt/destinations/impl/lancedb/configuration.py

@@ -59,6 +59,7 @@ class LanceDBClientOptions(BaseConfiguration):
     "sentence-transformers",
     "huggingface",
     "colbert",
+    "ollama",
 ]
 
 
@@ -92,8 +93,6 @@ class LanceDBClientConfiguration(DestinationClientDwhConfiguration):
     Make sure it corresponds with the associated embedding model's dimensionality."""
     vector_field_name: str = "vector"
     """Name of the special field to store the vector embeddings."""
-    id_field_name: str = "id__"
-    """Name of the special field to manage deduplication."""
     sentinel_table_name: str = "dltSentinelTable"
     """Name of the sentinel table that encapsulates datasets. Since LanceDB has no
     concept of schemas, this table serves as a proxy to group related dlt tables together."""

dlt/destinations/impl/lancedb/factory.py

@@ -26,8 +26,8 @@ class lancedb(Destination[LanceDBClientConfiguration, "LanceDBClient"]):
 
     def _raw_capabilities(self) -> DestinationCapabilitiesContext:
         caps = DestinationCapabilitiesContext()
-        caps.preferred_loader_file_format = "jsonl"
-        caps.supported_loader_file_formats = ["jsonl"]
+        caps.preferred_loader_file_format = "parquet"
+        caps.supported_loader_file_formats = ["parquet", "reference"]
         caps.type_mapper = LanceDBTypeMapper
 
         caps.max_identifier_length = 200
@@ -42,6 +42,10 @@ def _raw_capabilities(self) -> DestinationCapabilitiesContext:
         caps.timestamp_precision = 6
         caps.supported_replace_strategies = ["truncate-and-insert"]
 
+        caps.recommended_file_size = 128_000_000
+
+        caps.supported_merge_strategies = ["upsert"]
+
         return caps
 
     @property

dlt/destinations/impl/lancedb/lancedb_adapter.py

@@ -1,16 +1,20 @@
-from typing import Any
+from typing import Any, Dict
 
 from dlt.common.schema.typing import TColumnNames, TTableSchemaColumns
 from dlt.destinations.utils import get_resource_for_adapter
 from dlt.extract import DltResource
+from dlt.extract.items import TTableHintTemplate
 
 
 VECTORIZE_HINT = "x-lancedb-embed"
+NO_REMOVE_ORPHANS_HINT = "x-lancedb-remove-orphans"
 
 
 def lancedb_adapter(
     data: Any,
     embed: TColumnNames = None,
+    merge_key: TColumnNames = None,
+    no_remove_orphans: bool = False,
 ) -> DltResource:
     """Prepares data for the LanceDB destination by specifying which columns should be embedded.
 
@@ -20,6 +24,10 @@ def lancedb_adapter(
             object.
         embed (TColumnNames, optional): Specify columns to generate embeddings for.
             It can be a single column name as a string, or a list of column names.
+        merge_key (TColumnNames, optional): Specify columns to merge on.
+            It can be a single column name as a string, or a list of column names.
+        no_remove_orphans (bool): Specify whether to remove orphaned records in child
+            tables with no parent records after merges to maintain referential integrity.
 
     Returns:
         DltResource: A resource with applied LanceDB-specific hints.
@@ -34,6 +42,7 @@ def lancedb_adapter(
     """
     resource = get_resource_for_adapter(data)
 
+    additional_table_hints: Dict[str, TTableHintTemplate[Any]] = {}
     column_hints: TTableSchemaColumns = {}
 
     if embed:
@@ -50,9 +59,28 @@ def lancedb_adapter(
                 VECTORIZE_HINT: True,  # type: ignore[misc]
             }
 
-    if not column_hints:
-        raise ValueError("A value for 'embed' must be specified.")
+    if merge_key:
+        if isinstance(merge_key, str):
+            merge_key = [merge_key]
+        if not isinstance(merge_key, list):
+            raise ValueError(
+                "'merge_key' must be a list of column names or a single column name as a string."
+            )
+
+        for column_name in merge_key:
+            column_hints[column_name] = {
+                "name": column_name,
+                "merge_key": True,
+            }
+
+    additional_table_hints[NO_REMOVE_ORPHANS_HINT] = no_remove_orphans
+
+    if column_hints or additional_table_hints:
+        resource.apply_hints(columns=column_hints, additional_table_hints=additional_table_hints)
     else:
-        resource.apply_hints(columns=column_hints)
+        raise ValueError(
+            "You must must provide at least either the 'embed' or 'merge_key' or 'remove_orphans'"
+            " argument if using the adapter."
+        )
 
     return resource