def table(
    self,
    table_name: str,
    *,
    load_ids: Optional[Collection[str]] = None,
    incremental: Optional[Incremental] = None,  # NEW
    **kwargs: Any,
) -> dlt.Relation:

def incremental(self, incremental: Incremental) -> Self:
    """Apply incremental filtering as a WHERE clause on this relation.

    Generates a SQL WHERE condition from the Incremental's cursor state
    (last_value, end_value, range boundaries). Only supports `max` and
    `min` as `last_value_func` — custom functions raise ValueError.

    The cursor_path is interpreted as:
    - Simple name ("created_at") — WHERE on this relation's column
    - Dotted name ("_dlt_loads.inserted_at") — auto-join the table
      (via schema references from PR #3590), then WHERE on the
      joined column

    Args:
        incremental: Incremental instance with cursor configuration.

    Returns:
        New Relation with the WHERE clause applied.
    """

@property
def is_incremental(self) -> bool:
    """Returns True if this relation has an incremental WHERE clause."""

dataset.table("orders", incremental=dlt.sources.incremental("_dlt_loads.inserted_at"))

def incremental(self, inc: Incremental) -> Self:
    table_name, cursor_column = _parse_incremental_cursor_path(inc.cursor_path)

    rel = self
    if table_name:
        # check if already joined
        qualifier_map = _extract_joined_table_aliases(rel.sqlglot_expression)
        if table_name not in qualifier_map:
            # auto-join without projection
            join_params, target_table, target_qualifier = _discover_join_params(rel, table_name)
            rel = _apply_join(
                rel, join_params,
                target_table=target_table,
                target_qualifier=target_qualifier,
                projection_prefix=table_name,
                project=False,  # NEW flag — skip _apply_join_projection
            )
            table_qualifier = target_qualifier
        else:
            table_qualifier = qualifier_map[table_name]

        column_ref = sge.Column(
            this=sge.to_identifier(cursor_column, quoted=True),
            table=sge.to_identifier(table_qualifier, quoted=False),
        )
    else:
        column_ref = sge.Column(this=sge.to_identifier(cursor_column, quoted=True))

    condition = _build_incremental_condition(inc, column_ref)
    condition.meta["dlt_incremental"] = True
    return rel.where(condition)

def _apply_join(
    relation: _TRelation,
    join_params: list[_JoinParams],
    *,
    target_table: str,
    target_qualifier: str,
    projection_prefix: str,
    kind: TJoinType = "inner",
    project: bool = True,  # NEW
) -> _TRelation:
    query = relation.sqlglot_expression.copy()
    # ... build JOIN expressions (unchanged) ...

    if project:
        _apply_join_projection(...)

    rel = relation.__copy__()
    rel._sqlglot_expression = query
    return rel

def _build_incremental_condition(
    inc: Incremental,
    column_ref: sge.Column,
) -> sge.Expression:
    """Build SQLGlot WHERE condition from Incremental state.

    Only supports max and min as last_value_func.
    """
    last_value_func = inc.last_value_func
    if last_value_func not in (min, max):
        raise ValueError(
            "Incremental WHERE generation only supports min/max last_value_func,"
            f" got {last_value_func}"
        )

    start_value = inc.last_value  # reads from state, applies lag
    end_value = inc.end_value

    if last_value_func is max:
        start_op = sge.GTE if inc.range_start == "closed" else sge.GT
        end_op = sge.LT if inc.range_end == "open" else sge.LTE
    else:  # min
        start_op = sge.LTE if inc.range_start == "closed" else sge.LT
        end_op = sge.GT if inc.range_end == "open" else sge.GTE

    conditions = []
    if start_value is not None:
        conditions.append(start_op(this=column_ref, expression=_to_typed_literal(start_value)))
        if end_value is not None:
            conditions.append(end_op(this=column_ref, expression=_to_typed_literal(end_value)))
        if inc.on_cursor_value_missing == "include":
            # OR cursor IS NULL — wraps the existing conditions
            pass  # applied after AND combination below

    if inc.on_cursor_value_missing == "exclude":
        conditions.append(sge.Not(this=sge.Is(this=column_ref, expression=sge.Null())))

    if not conditions:
        raise ValueError("Incremental has no last_value and no on_cursor_value_missing='exclude'")

    result = conditions[0]
    for c in conditions[1:]:
        result = sge.And(this=result, expression=c)

    # wrap with OR IS NULL for "include" mode
    if inc.on_cursor_value_missing == "include" and start_value is not None:
        result = sge.Or(this=result, expression=sge.Is(this=column_ref, expression=sge.Null()))

    return result

def _parse_incremental_cursor_path(cursor_path: str) -> tuple[Optional[str], str]:
    """Parse 'table.column' or 'column' from incremental cursor_path."""
    if "." in cursor_path:
        table_name, column_name = cursor_path.rsplit(".", 1)
        return table_name, column_name
    return None, cursor_path

@property
def is_incremental(self) -> bool:
    for node in self.sqlglot_expression.walk():
        if node.meta.get("dlt_incremental"):
            return True
    return False