Incremental table hints and incremental in resource decorator

dlt/common/destination/typing.py

@@ -1,11 +1,17 @@
 from typing import Optional
 
-from dlt.common.schema.typing import _TTableSchemaBase, TWriteDisposition, TTableReferenceParam
+from dlt.common.schema.typing import (
+    _TTableSchemaBase,
+    TWriteDisposition,
+    TTableReferenceParam,
+    IncrementalArgs,
+)
 
 
 class PreparedTableSchema(_TTableSchemaBase, total=False):
     """Table schema with all hints prepared to be loaded"""
 
     write_disposition: TWriteDisposition
     references: Optional[TTableReferenceParam]
+    incremental: Optional[IncrementalArgs]
     _x_prepared: bool  # needed for the type checker

dlt/extract/incremental/typing.py → dlt/common/incremental/typing.py

@@ -2,9 +2,7 @@
 
 from typing import Any, Callable, List, Literal, Optional, Sequence, TypeVar, Union
 
-from dlt.common.schema.typing import TColumnNames
-from dlt.common.typing import TSortOrder
-from dlt.extract.items import TTableHintTemplate
+from dlt.common.typing import TSortOrder, TTableHintTemplate, TColumnNames
 
 TCursorValue = TypeVar("TCursorValue", bound=Any)
 LastValueFunc = Callable[[Sequence[TCursorValue]], Any]
@@ -19,10 +17,12 @@ class IncrementalColumnState(TypedDict):
 
 class IncrementalArgs(TypedDict, total=False):
     cursor_path: str
-    initial_value: Optional[str]
-    last_value_func: Optional[LastValueFunc[str]]
+    initial_value: Optional[Any]
+    last_value_func: Optional[Union[LastValueFunc[str], Literal["min", "max"]]]
+    """Last value callable or name of built in function"""
     primary_key: Optional[TTableHintTemplate[TColumnNames]]
-    end_value: Optional[str]
+    end_value: Optional[Any]
     row_order: Optional[TSortOrder]
     allow_external_schedulers: Optional[bool]
     lag: Optional[Union[float, int]]
+    on_cursor_value_missing: Optional[OnCursorValueMissing]

dlt/common/pipeline.py

@@ -48,15 +48,14 @@
 )
 from dlt.common.schema import Schema
 from dlt.common.schema.typing import (
-    TColumnNames,
     TColumnSchema,
     TWriteDispositionConfig,
     TSchemaContract,
 )
 from dlt.common.storages.load_package import ParsedLoadJobFileName
 from dlt.common.storages.load_storage import LoadPackageInfo
 from dlt.common.time import ensure_pendulum_datetime, precise_time
-from dlt.common.typing import DictStrAny, REPattern, StrAny, SupportsHumanize
+from dlt.common.typing import DictStrAny, REPattern, StrAny, SupportsHumanize, TColumnNames
 from dlt.common.jsonpath import delete_matches, TAnyJsonPath
 from dlt.common.data_writers.writers import TLoaderFileFormat
 from dlt.common.utils import RowCounts, merge_row_counts

dlt/common/schema/typing.py

@@ -20,6 +20,7 @@
 from dlt.common.data_types import TDataType
 from dlt.common.normalizers.typing import TNormalizersConfig
 from dlt.common.typing import TSortOrder, TAnyDateTime, TLoaderFileFormat
+from dlt.common.incremental.typing import IncrementalArgs
 
 try:
     from pydantic import BaseModel as _PydanticBaseModel
@@ -132,8 +133,6 @@ class TColumnPropInfo(NamedTuple):
     "timestamp", "iso_timestamp", "iso_date", "large_integer", "hexbytes_to_text", "wei_to_double"
 ]
 TTypeDetectionFunc = Callable[[Type[Any], Any], Optional[TDataType]]
-TColumnNames = Union[str, Sequence[str]]
-"""A string representing a column name or a list of"""
 
 
 class TColumnType(TypedDict, total=False):
@@ -279,6 +278,7 @@ class TTableSchema(_TTableSchemaBase, total=False):
 
     write_disposition: Optional[TWriteDisposition]
     references: Optional[TTableReferenceParam]
+    incremental: Optional[IncrementalArgs]
 
 
 class TPartialTableSchema(TTableSchema):

dlt/common/typing.py

@@ -29,6 +29,7 @@
     Iterator,
     Generator,
     NamedTuple,
+    Sequence,
 )
 
 from typing_extensions import (
@@ -112,6 +113,8 @@ class SecretSentinel:
 
 TSecretStrValue = Annotated[str, SecretSentinel]
 
+TColumnNames = Union[str, Sequence[str]]
+"""A string representing a column name or a list of"""
 TDataItem: TypeAlias = Any
 """A single data item as extracted from data source"""
 TDataItems: TypeAlias = Union[TDataItem, List[TDataItem]]
@@ -126,6 +129,10 @@ class SecretSentinel:
 TLoaderFileFormat = Literal["jsonl", "typed-jsonl", "insert_values", "parquet", "csv", "reference"]
 """known loader file formats"""
 
+TDynHintType = TypeVar("TDynHintType")
+TFunHintTemplate = Callable[[TDataItem], TDynHintType]
+TTableHintTemplate = Union[TDynHintType, TFunHintTemplate[TDynHintType]]
+
 
 class ConfigValueSentinel(NamedTuple):
     """Class to create singleton sentinel for config and secret injected value"""

dlt/destinations/impl/bigquery/bigquery_adapter.py

@@ -4,10 +4,8 @@
 
 from dlt.common.destination import PreparedTableSchema
 from dlt.common.pendulum import timezone
-from dlt.common.schema.typing import (
-    TColumnNames,
-    TTableSchemaColumns,
-)
+from dlt.common.schema.typing import TTableSchemaColumns
+from dlt.common.typing import TColumnNames
 from dlt.destinations.utils import get_resource_for_adapter
 from dlt.extract import DltResource
 from dlt.extract.items import TTableHintTemplate

dlt/destinations/impl/lancedb/lancedb_adapter.py

@@ -1,6 +1,7 @@
 from typing import Any, Dict
 
-from dlt.common.schema.typing import TColumnNames, TTableSchemaColumns
+from dlt.common.schema.typing import TTableSchemaColumns
+from dlt.common.typing import TColumnNames
 from dlt.destinations.utils import get_resource_for_adapter
 from dlt.extract import DltResource
 from dlt.extract.items import TTableHintTemplate

dlt/destinations/impl/qdrant/qdrant_adapter.py

@@ -1,6 +1,7 @@
 from typing import Any
 
-from dlt.common.schema.typing import TColumnNames, TTableSchemaColumns
+from dlt.common.schema.typing import TTableSchemaColumns
+from dlt.common.typing import TColumnNames
 from dlt.extract import DltResource
 from dlt.destinations.utils import get_resource_for_adapter
 

dlt/destinations/impl/weaviate/weaviate_adapter.py

@@ -1,6 +1,7 @@
 from typing import Dict, Any, Literal, Set, get_args
 
-from dlt.common.schema.typing import TColumnNames, TTableSchemaColumns
+from dlt.common.schema.typing import TTableSchemaColumns
+from dlt.common.typing import TColumnNames
 from dlt.extract import DltResource, resource as make_resource
 from dlt.destinations.utils import get_resource_for_adapter
 

dlt/extract/decorators.py

@@ -32,7 +32,6 @@
 from dlt.common.schema.utils import DEFAULT_WRITE_DISPOSITION
 from dlt.common.schema.schema import Schema
 from dlt.common.schema.typing import (
-    TColumnNames,
     TFileFormat,
     TWriteDisposition,
     TWriteDispositionConfig,
@@ -43,7 +42,8 @@
 )
 from dlt.common.storages.exceptions import SchemaNotFoundError
 from dlt.common.storages.schema_storage import SchemaStorage
-from dlt.common.typing import AnyFun, ParamSpec, Concatenate, TDataItem, TDataItems
+from dlt.common.typing import AnyFun, ParamSpec, Concatenate, TDataItem, TDataItems, TColumnNames
+
 from dlt.common.utils import get_callable_name, get_module_name, is_inner_callable
 
 from dlt.extract.hints import make_hints
@@ -70,6 +70,7 @@
     TSourceFunParams,
 )
 from dlt.extract.resource import DltResource, TUnboundDltResource, TDltResourceImpl
+from dlt.extract.incremental import TIncrementalConfig
 
 
 @configspec
@@ -446,6 +447,7 @@ def resource(
     selected: bool = True,
     spec: Type[BaseConfiguration] = None,
     parallelized: bool = False,
+    incremental: Optional[TIncrementalConfig] = None,
     _impl_cls: Type[TDltResourceImpl] = DltResource,  # type: ignore[assignment]
 ) -> TDltResourceImpl: ...
 
@@ -468,6 +470,7 @@ def resource(
     selected: bool = True,
     spec: Type[BaseConfiguration] = None,
     parallelized: bool = False,
+    incremental: Optional[TIncrementalConfig] = None,
     _impl_cls: Type[TDltResourceImpl] = DltResource,  # type: ignore[assignment]
 ) -> Callable[[Callable[TResourceFunParams, Any]], TDltResourceImpl]: ...
 
@@ -490,6 +493,7 @@ def resource(
     selected: bool = True,
     spec: Type[BaseConfiguration] = None,
     parallelized: bool = False,
+    incremental: Optional[TIncrementalConfig] = None,
     _impl_cls: Type[TDltResourceImpl] = DltResource,  # type: ignore[assignment]
     standalone: Literal[True] = True,
 ) -> Callable[
@@ -515,6 +519,7 @@ def resource(
     selected: bool = True,
     spec: Type[BaseConfiguration] = None,
     parallelized: bool = False,
+    incremental: Optional[TIncrementalConfig] = None,
     _impl_cls: Type[TDltResourceImpl] = DltResource,  # type: ignore[assignment]
 ) -> TDltResourceImpl: ...
 
@@ -536,6 +541,7 @@ def resource(
     selected: bool = True,
     spec: Type[BaseConfiguration] = None,
     parallelized: bool = False,
+    incremental: Optional[TIncrementalConfig] = None,
     _impl_cls: Type[TDltResourceImpl] = DltResource,  # type: ignore[assignment]
     standalone: bool = False,
     data_from: TUnboundDltResource = None,
@@ -642,7 +648,9 @@ def make_resource(_name: str, _section: str, _data: Any) -> TDltResourceImpl:
             selected,
             cast(DltResource, data_from),
             True,
+            incremental=incremental,
         )
+
         # If custom nesting level was specified then
         # we need to add it to table hints so that
         # later in normalizer dlt/common/normalizers/json/relational.py

dlt/extract/extract.py

@@ -2,7 +2,7 @@
 from collections.abc import Sequence as C_Sequence
 from copy import copy
 import itertools
-from typing import Iterator, List, Dict, Any, Optional
+from typing import Iterator, List, Dict, Any, Optional, Mapping
 import yaml
 
 from dlt.common.configuration.container import Container
@@ -17,13 +17,12 @@
     WithStepInfo,
     reset_resource_state,
 )
-from dlt.common.typing import DictStrAny
+from dlt.common.typing import DictStrAny, TColumnNames
 from dlt.common.runtime import signals
 from dlt.common.runtime.collector import Collector, NULL_COLLECTOR
 from dlt.common.schema import Schema, utils
 from dlt.common.schema.typing import (
     TAnySchemaColumns,
-    TColumnNames,
     TSchemaContract,
     TTableFormat,
     TWriteDispositionConfig,
@@ -39,7 +38,7 @@
 
 from dlt.extract.decorators import SourceInjectableContext, SourceSchemaInjectableContext
 from dlt.extract.exceptions import DataItemRequiredForDynamicTableHints
-from dlt.extract.incremental import IncrementalResourceWrapper
+from dlt.extract.incremental import IncrementalResourceWrapper, Incremental
 from dlt.extract.pipe_iterator import PipeIterator
 from dlt.extract.source import DltSource
 from dlt.extract.resource import DltResource
@@ -238,7 +237,10 @@ def _compute_metrics(self, load_id: str, source: DltSource) -> ExtractMetrics:
                         hint = hint.incremental
                     # sometimes internal incremental is not bound
                     if hint:
-                        hints[name] = dict(hint)  # type: ignore[call-overload]
+                        if isinstance(hint, Incremental):
+                            hints[name] = hint.to_table_hint()
+                        else:
+                            hints[name] = dict(hint)  # type: ignore[call-overload]
                     continue
                 if name == "original_columns":
                     # this is original type of the columns ie. Pydantic model

dlt/extract/hints.py

@@ -4,7 +4,6 @@
 from dlt.common import logger
 from dlt.common.schema.typing import (
     C_DLT_ID,
-    TColumnNames,
     TColumnProp,
     TFileFormat,
     TPartialTableSchema,
@@ -28,7 +27,7 @@
     new_column,
     new_table,
 )
-from dlt.common.typing import TDataItem
+from dlt.common.typing import TDataItem, TColumnNames
 from dlt.common.time import ensure_pendulum_datetime
 from dlt.common.utils import clone_dict_nested
 from dlt.common.normalizers.json.relational import DataItemNormalizer
@@ -204,6 +203,10 @@ def compute_table_schema(self, item: TDataItem = None, meta: Any = None) -> TTab
             for k, v in table_template.items()
             if k not in NATURAL_CALLABLES
         }  # type: ignore
+        if "incremental" in table_template:
+            incremental = table_template["incremental"]
+            if isinstance(incremental, Incremental) and incremental is not Incremental.EMPTY:
+                resolved_template["incremental"] = incremental
         table_schema = self._create_table_schema(resolved_template, self.name)
         migrate_complex_types(table_schema, warn=True)
         validate_dict_ignoring_xkeys(
@@ -518,6 +521,9 @@ def _create_table_schema(resource_hints: TResourceHints, resource_name: str) ->
                     "disposition": resource_hints["write_disposition"]
                 }  # wrap in dict
             DltResourceHints._merge_write_disposition_dict(resource_hints)  # type: ignore[arg-type]
+        if "incremental" in resource_hints:
+            if isinstance(resource_hints["incremental"], Incremental):
+                resource_hints["incremental"] = resource_hints["incremental"].to_table_hint()  # type: ignore[typeddict-item]
         dict_ = cast(TTableSchema, resource_hints)
         dict_["resource"] = resource_name
         return dict_