本文介绍如何使用 Daft 通过 LAS Iceberg REST Catalog 读取和写入 Iceberg 表。

实践目标 #

本文提供一份完整的 Python 示例脚本，封装 LAS Iceberg REST Catalog 鉴权、TOS 文件访问和 Daft 读写 Iceberg 表所需的公共逻辑。用户可以通过 LAS_OPERATION 环境变量选择执行 read、append 或 overwrite。每次运行脚本只执行一种操作，不会依次执行全部操作。
完成本文操作后，可以验证以下链路：

• 使用火山引擎 AK/SK 对 LAS Iceberg REST API 请求进行签名。
• 通过 PyIceberg 从 LAS Iceberg REST Catalog 加载表及当前 Snapshot。
• 通过 Daft 读取存储在 TOS 中的 Iceberg 数据文件。
• 通过 Daft 执行 append 或 overwrite，并通过 LAS Iceberg REST Catalog 提交新的 Snapshot。

您可参考本文的说明，创建本地脚本文件并复制完整代码，进行操作实践。

工作原理 #

LAS Iceberg REST Catalog 提供表元数据查询、并发控制和 Snapshot 提交能力；TOS 用于存储 Parquet 数据文件，以及 metadata、manifest、manifest list 等 Iceberg 元数据文件。
完整读写链路如下：

1. 示例脚本使用 AK/SK 对 LAS Iceberg REST Catalog 请求进行签名。
2. PyIceberg 通过 REST Catalog 加载目标表、Schema、当前 Snapshot 和数据文件位置。
3. Daft 根据 Iceberg 元数据读取 TOS 中的数据文件。
4. 执行写入时，Daft 先将新的数据文件写入 TOS，再通过 PyIceberg 表对象向 LAS Iceberg REST Catalog 提交新的 Snapshot。
5. 脚本重新加载目标表，检查 Snapshot 是否变化，并读取写入结果进行验证。

服务和网络要求 #

开始操作前，请确认：

• 当前地域已支持并开通 LAS Iceberg REST Catalog。该能力的开通方式和地域支持范围请参考：Apache® Gravitino Iceberg REST Catalog 使用说明。
• 运行环境可以访问 LAS Iceberg REST Catalog Endpoint。
• 运行环境可以访问目标表所在的 TOS Endpoint。
• 如果服务设置了 VPC、访问白名单或源 IP 限制，运行环境已满足对应要求。

权限及访问凭证要求 #

请准备具备所需权限的火山引擎 AK/SK。使用 STS 临时安全凭证时，还需准备 Session Token。
AK/SK 所属账号需要同时具备以下权限：

相关权限配置可参考：

• LAS 权限管理：权限管理
• TOS 权限配置：TOS 权限配置

LAS Catalog 权限和 TOS 权限属于两条独立的访问链路。仅具备 Catalog 权限时，可能可以加载表元数据，但无法读取或写入 TOS 中的实际文件。

已验证环境要求 #

本文示例已在以下环境完成 read、append 和 overwrite 端到端验证。建议优先使用相同的 LAS 开发机镜像和依赖版本。

• 示例脚本语法要求 Python 3.10 及以上，但当前仅在 Python 3.11.11 环境完成端到端验证。
• 示例脚本会在当前 Python 进程中适配 Daft 的 TOS Writer。相关逻辑依赖 Daft 的私有文件系统接口 _infer_filesystem 或 _build_filesystem，不同 Daft 版本可能存在兼容性差异，不建议直接使用未经验证的最新版。

准备专用测试表 #

append 和 overwrite 示例使用以下表结构：

请提前在 LAS Catalog 中创建一张具有上述字段的专用测试表。测试表可以为空，也可以包含少量测试数据。
如果目标表使用其他 Schema，需要同时修改示例脚本中的 build_demo_write_dataframe() 和字段校验逻辑，使写入 DataFrame 的字段名称及类型与目标表兼容。

overwrite 示例会覆盖测试表当前可见的全部数据，并在操作前后执行全表行数统计。请仅对数据量较小的专用测试表执行。

检查 Python 环境 #

执行以下命令，确认所需依赖可以正常导入：

python -c 'import daft, pyiceberg, pyarrow, requests, httpx, tosfs; print("dependencies OK")'

可以继续执行以下命令检查实际安装版本：

python --version
python -c 'import importlib.metadata as m; print("pyiceberg", m.version("pyiceberg")); print("pyarrow", m.version("pyarrow")); print("tosfs", m.version("tosfs")); print("httpx", m.version("httpx")); print("requests", m.version("requests"))'

Daft 请优先使用已验证的 VeDaft 镜像版本 0.7.2.post6。

配置环境变量 #

安全录入 LAS AK/SK #

不要将真实 AK/SK 直接写入脚本、文档、命令历史或代码仓库。可以使用终端隐藏输入：

read -rsp "VOLC_ACCESSKEY: " VOLC_ACCESSKEY
echo
read -rsp "VOLC_SECRETKEY: " VOLC_SECRETKEY
echo
export VOLC_ACCESSKEY VOLC_SECRETKEY

使用 STS 临时安全凭证时，继续设置 Session Token：

read -rsp "VOLC_SESSION_TOKEN: " VOLC_SESSION_TOKEN
echo
export VOLC_SESSION_TOKEN

配置 Catalog、TOS 和目标表 #

export LAS_REGION="<region，例如 cn-beijing>"
export LAS_ICEBERG_REST_URI="https://<LAS-REST-ENDPOINT>/iceberg"

# 仅在多网卡机器需要固定访问 Catalog 的源 IP 时设置。
# 不需要时请执行：unset LAS_REST_SOURCE_IP
export LAS_REST_SOURCE_IP="<本机可访问 LAS 的私网 IP>"

export TOS_REGION="<TOS region>"
export TOS_ENDPOINT="https://<TOS-ENDPOINT>"

export ICEBERG_NAMESPACE="<数据库或 namespace>"
export ICEBERG_TABLE="<表名>"
export ICEBERG_READ_LIMIT="10"

请以控制台或服务实际提供的 Endpoint 为准，不要自行拼接额外路径。

可选：为 TOS 配置独立凭证 #

LAS 和 TOS 使用同一套 AK/SK 时，无需配置本节变量，脚本会自动复用 VOLC_ACCESSKEY、VOLC_SECRETKEY 和 VOLC_SESSION_TOKEN。
TOS 使用另一套凭证时，可以通过隐藏输入设置：

read -rsp "TOS_ACCESS_KEY: " TOS_ACCESS_KEY
echo
read -rsp "TOS_SECRET_KEY: " TOS_SECRET_KEY
echo
export TOS_ACCESS_KEY TOS_SECRET_KEY

使用 TOS STS 临时安全凭证时，继续设置：

read -rsp "TOS_SESSION_TOKEN: " TOS_SESSION_TOKEN
echo
export TOS_SESSION_TOKEN

创建示例脚本 #

创建名为 las_daft_iceberg_example.py 的 Python 文件，并将以下完整代码复制到文件中。
该示例脚本封装了 LAS Iceberg REST Catalog 鉴权、TOS 文件访问和 Daft 读写 Iceberg 表所需的完整逻辑。用户可以通过 LAS_OPERATION 环境变量选择执行以下操作：

• read：使用 AK/SK 连接 LAS Iceberg REST Catalog，根据 Catalog 返回的表元数据访问 TOS，并通过 Daft 读取 Iceberg 表。
• append：向目标 Iceberg 表追加数据，并检查写入前后的 Snapshot 是否发生变化。
• overwrite：覆盖目标 Iceberg 表中的现有数据，并检查 Snapshot 和表行数是否符合预期。

每次运行脚本只执行一种操作，不会依次执行 read、append 和 overwrite。脚本默认执行只读操作；只有将 LAS_OPERATION 设置为 append 或 overwrite，并正确配置 LAS_WRITE_CONFIRM 后，才会执行写入。

"""Read and optionally append to a LAS Iceberg table with Daft.

This is a self-contained customer example. The script:

1. signs PyIceberg REST Catalog requests with Volcengine AK/SK;
2. loads an Iceberg table from LAS;
3. reads the table with Daft and TOS native I/O;
4. optionally appends or overwrites with one demo row and commits a new
   Iceberg snapshot.

No endpoint, account, bucket, table, AK, or SK is hard-coded.
Writes are disabled by default and require explicit operation and confirmation
environment variables.
"""

from __future__ import annotations

import hashlib
import hmac
import inspect
import os
import posixpath
import re
import secrets
import shlex
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
from urllib.parse import parse_qs, quote, unquote, unquote_plus, urlparse

import httpx
import pyarrow as pa
import pyarrow.fs as pafs
from pyiceberg.catalog.rest import RestCatalog
from pyiceberg.io import fsspec as iceberg_fsspec
from requests.adapters import HTTPAdapter
from requests.auth import AuthBase
from tosfs import TosFileSystem

import daft
import daft.filesystem as daft_filesystem
from daft.io import IOConfig, TosConfig


def require_env(name: str) -> str:
    value = os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value.strip()


def optional_env(name: str) -> str | None:
    value = os.environ.get(name)
    return value.strip() if value and value.strip() else None


REGION = os.environ.get("LAS_REGION", "cn-beijing").strip()
REST_URI = require_env("LAS_ICEBERG_REST_URI").rstrip("/")
REST_SOURCE_IP = optional_env("LAS_REST_SOURCE_IP")

NAMESPACE = require_env("ICEBERG_NAMESPACE")
TABLE_NAME = require_env("ICEBERG_TABLE")
READ_LIMIT = int(os.environ.get("ICEBERG_READ_LIMIT", "10"))

LAS_ACCESS_KEY = require_env("VOLC_ACCESSKEY")
LAS_SECRET_KEY = require_env("VOLC_SECRETKEY")
LAS_SESSION_TOKEN = optional_env("VOLC_SESSION_TOKEN")

TOS_REGION = os.environ.get("TOS_REGION", REGION).strip()
TOS_ENDPOINT = require_env("TOS_ENDPOINT")
TOS_ACCESS_KEY = optional_env("TOS_ACCESS_KEY") or LAS_ACCESS_KEY
TOS_SECRET_KEY = optional_env("TOS_SECRET_KEY") or LAS_SECRET_KEY
TOS_SESSION_TOKEN = optional_env("TOS_SESSION_TOKEN") or LAS_SESSION_TOKEN

OPERATION = os.environ.get("LAS_OPERATION", "read").strip().lower()
if OPERATION not in {"read", "append", "overwrite"}:
    raise RuntimeError(
        "LAS_OPERATION must be 'read', 'append', or 'overwrite'"
    )

ENABLE_WRITE = OPERATION in {"append", "overwrite"}
FULL_TABLE_NAME = f"{NAMESPACE}.{TABLE_NAME}"


@dataclass(frozen=True)
class Credentials:
    access_key: str
    secret_key: str
    session_token: str | None = None


class VolcOpenApiAuthProvider:
    """Generate Volcengine OpenAPI HMAC-SHA256 request signatures."""

    def __init__(
        self,
        service: str,
        region: str,
        access_key: str,
        secret_key: str,
        session_token: str | None = None,
    ) -> None:
        if not access_key or not access_key.strip():
            raise ValueError("access_key must not be empty")
        if not secret_key or not secret_key.strip():
            raise ValueError("secret_key must not be empty")
        if not service or not service.strip():
            raise ValueError("service must not be empty")
        if not region or not region.strip():
            raise ValueError("region must not be empty")

        self.service = service
        self.region = region
        self.credentials = Credentials(access_key, secret_key, session_token)

    def authentication(self, request: httpx.Request) -> httpx.Request:
        request_url = str(request.url)
        parsed_url = urlparse(request_url)
        headers = request.headers

        request_date = self._reset_and_get_date_header(headers)
        scope_date = request_date[:8]
        payload = request.content if request.content is not None else b""
        payload_hash = hashlib.sha256(payload).hexdigest()
        headers["x-content-sha256"] = payload_hash

        if self.credentials.session_token:
            headers["x-security-token"] = self.credentials.session_token

        if "host" in headers:
            # LAS validates the hostname, excluding the port, in the Host header.
            headers["host"] = parsed_url.netloc.split(":")[0]

        canonical_headers, signed_headers = self._extract_canonical_headers(headers)
        canonical_request = self._canonical_request(
            request_url,
            request.method,
            canonical_headers,
            signed_headers,
            payload_hash,
        )

        scope = f"{scope_date}/{self.region}/{self.service}/request"
        string_to_sign = "\n".join(
            [
                "HMAC-SHA256",
                request_date,
                scope,
                hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(),
            ]
        ).encode("utf-8")
        signing_key = self._generate_signing_key(
            self.credentials.secret_key,
            scope_date,
            self.region,
            self.service,
        )
        signature = hmac.new(signing_key, string_to_sign, hashlib.sha256).hexdigest()

        headers["Authorization"] = (
            "HMAC-SHA256 "
            f"Credential={self.credentials.access_key}/{scope}, "
            f"SignedHeaders={signed_headers}, "
            f"Signature={signature}"
        )
        return request

    @staticmethod
    def _reset_and_get_date_header(headers: httpx.Headers) -> str:
        date_value = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        for header in ("x-date", "date"):
            if header in headers:
                date_value = headers[header]
                break
        for header in ("x-date", "date"):
            if header in headers:
                del headers[header]
        headers["x-date"] = date_value
        return date_value

    @staticmethod
    def _normalize_whitespace(value: str) -> str:
        return " ".join(shlex.split(value, posix=False))

    @classmethod
    def _extract_canonical_headers(
        cls,
        headers: httpx.Headers,
    ) -> tuple[str, str]:
        selected: dict[str, list[str]] = {}
        for name, value in headers.items():
            name = name.strip().lower()
            if name not in {"host", "content-type", "date"} and not (
                name.startswith("x-") and name != "x-client-context"
            ):
                continue
            selected.setdefault(name, []).append(
                cls._normalize_whitespace(value).strip()
            )

        canonical_headers = ""
        signed_header_names: list[str] = []
        for name in sorted(selected):
            canonical_headers += f"{name}:{','.join(sorted(selected[name]))}\n"
            signed_header_names.append(name)
        return canonical_headers, ";".join(signed_header_names)

    @staticmethod
    def _canonical_path(path: str, service: str) -> str:
        fixed_path = posixpath.normpath(path)
        fixed_path = re.sub("/+", "/", fixed_path)
        if path.endswith("/") and not fixed_path.endswith("/"):
            fixed_path += "/"
        if service in {"s3", "host"}:
            fixed_path = unquote(fixed_path)
        return quote(fixed_path, safe="/~")

    @staticmethod
    def _canonical_querystring(query: str) -> str:
        query = quote(unquote_plus(query), safe="&=")
        items: dict[str, list[str]] = {}
        for name, values in parse_qs(query, keep_blank_values=True).items():
            encoded_name = quote(name, safe="-_.~")
            items[encoded_name] = [
                quote(value, safe="-_.~") for value in values
            ]

        pairs: list[str] = []
        for name in sorted(items):
            for value in items[name]:
                pairs.append(f"{name}={value}")
        return "&".join(pairs).replace("+", "%20")

    def _canonical_request(
        self,
        url: str,
        method: str,
        canonical_headers: str,
        signed_headers: str,
        payload_hash: str,
    ) -> str:
        parsed = urlparse(url)
        canonical_path = self._canonical_path(parsed.path, self.service)
        query = url.split("?", 1)[1] if "?" in url else ""
        canonical_query = self._canonical_querystring(query)
        return (
            f"{method.upper()}\n{canonical_path}\n{canonical_query}\n"
            f"{canonical_headers}\n{signed_headers}\n{payload_hash}"
        )

    @classmethod
    def _generate_signing_key(
        cls,
        secret_key: str,
        date: str,
        region: str,
        service: str,
    ) -> bytes:
        def sign(key: bytes, message: str) -> bytes:
            return hmac.new(key, message.encode("utf-8"), hashlib.sha256).digest()

        date_key = sign(secret_key.encode("utf-8"), date)
        region_key = sign(date_key, region)
        service_key = sign(region_key, service)
        return sign(service_key, "request")


class SourceAddressAdapter(HTTPAdapter):
    """Make requests/urllib3 originate from a selected local IP."""

    def __init__(self, source_ip: str, *args: Any, **kwargs: Any) -> None:
        self.source_ip = source_ip
        super().__init__(*args, **kwargs)

    def init_poolmanager(
        self,
        connections: int,
        maxsize: int,
        block: bool = False,
        **pool_kwargs: Any,
    ) -> None:
        pool_kwargs["source_address"] = (self.source_ip, 0)
        super().init_poolmanager(
            connections,
            maxsize,
            block=block,
            **pool_kwargs,
        )


class VolcRequestsAuth(AuthBase):
    """Adapt the standalone httpx signer to requests used by PyIceberg."""

    def __init__(
        self,
        access_key: str,
        secret_key: str,
        session_token: str | None,
        region: str,
    ) -> None:
        self.region = region
        self.auth_provider = VolcOpenApiAuthProvider(
            service="las",
            region=region,
            access_key=access_key,
            secret_key=secret_key,
            session_token=session_token,
        )

    def __call__(self, request: Any) -> Any:
        body = request.body
        if body is None:
            body = b""
        elif not isinstance(body, (bytes, str)):
            raise TypeError(f"Unsupported request body type: {type(body)!r}")

        parsed_url = urlparse(request.url)
        headers = dict(request.headers)
        headers["Host"] = parsed_url.hostname
        headers["Region"] = self.region

        httpx_request = httpx.Request(
            method=request.method,
            url=request.url,
            headers=headers,
            content=body,
        )
        signed_request = self.auth_provider.authentication(httpx_request)

        for header_name in (
            "Host",
            "Authorization",
            "X-Date",
            "X-Content-Sha256",
            "X-Security-Token",
        ):
            if header_name in signed_request.headers:
                request.headers[header_name] = signed_request.headers[header_name]

        request.headers["Region"] = self.region
        return request


class VolcRestCatalog(RestCatalog):
    """PyIceberg REST Catalog using Volcengine signing and optional source IP."""

    def __init__(
        self,
        name: str,
        *,
        access_key: str,
        secret_key: str,
        session_token: str | None,
        region: str,
        source_ip: str | None,
        **properties: Any,
    ) -> None:
        self._volc_auth = VolcRequestsAuth(
            access_key=access_key,
            secret_key=secret_key,
            session_token=session_token,
            region=region,
        )
        self._region = region
        self._source_ip = source_ip
        super().__init__(name=name, **properties)

    def _create_session(self) -> Any:
        session = super()._create_session()
        session.trust_env = False
        session.auth = self._volc_auth
        session.headers.update({"Region": self._region})

        if self._source_ip:
            adapter = SourceAddressAdapter(self._source_ip)
            session.mount("http://", adapter)
            session.mount("https://", adapter)

        return session


def create_tos_filesystem(properties: dict[str, str]) -> TosFileSystem:
    """Build TosFS for PyIceberg metadata and Daft's Python writer."""
    kwargs: dict[str, Any] = {
        "endpoint": properties["tos.endpoint"],
        "region": properties["tos.region"],
        "key": properties["tos.access-key"],
        "secret": properties["tos.secret-key"],
        "max_retry_num": 1,
        "connection_timeout": 5,
        "socket_timeout": 30,
    }

    token = properties.get("tos.security-token")
    if token:
        parameters = inspect.signature(TosFileSystem.__init__).parameters
        token_parameter = next(
            (
                name
                for name in ("security_token", "session_token", "token")
                if name in parameters
            ),
            None,
        )
        if token_parameter is None:
            raise RuntimeError(
                "The installed tosfs does not expose a Session Token parameter. "
                "Use long-term TOS credentials or upgrade tosfs."
            )
        kwargs[token_parameter] = token

    return TosFileSystem(**kwargs)


TOS_PROPERTIES = {
    "py-io-impl": "pyiceberg.io.fsspec.FsspecFileIO",
    "tos.endpoint": TOS_ENDPOINT,
    "tos.region": TOS_REGION,
    "tos.access-key": TOS_ACCESS_KEY,
    "tos.secret-key": TOS_SECRET_KEY,
}
if TOS_SESSION_TOKEN:
    TOS_PROPERTIES["tos.security-token"] = TOS_SESSION_TOKEN


def create_catalog() -> VolcRestCatalog:
    # PyIceberg does not recognize tos:// by default.
    iceberg_fsspec.SCHEME_TO_FS["tos"] = create_tos_filesystem

    return VolcRestCatalog(
        name="las",
        uri=REST_URI,
        access_key=LAS_ACCESS_KEY,
        secret_key=LAS_SECRET_KEY,
        session_token=LAS_SESSION_TOKEN,
        region=REGION,
        source_ip=REST_SOURCE_IP,
        **TOS_PROPERTIES,
    )


def create_daft_io_config() -> IOConfig:
    return IOConfig(
        tos=TosConfig(
            region=TOS_REGION,
            endpoint=TOS_ENDPOINT,
            access_key=TOS_ACCESS_KEY,
            secret_key=TOS_SECRET_KEY,
            security_token=TOS_SESSION_TOKEN,
            max_retries=1,
            connect_timeout_ms=5_000,
            read_timeout_ms=30_000,
        )
    )


def enable_daft_tos_writer() -> None:
    """Teach Daft's Python Iceberg writer how to create files for tos:// paths.

    Older LAS-Daft builds use ``_infer_filesystem``. Newer builds use
    ``_build_filesystem``. This process-local adapter supports both layouts.
    """
    if getattr(daft_filesystem, "_las_tos_writer_enabled", False):
        return

    tos_fsspec = create_tos_filesystem(TOS_PROPERTIES)
    tos_pyarrow_filesystem = pafs.PyFileSystem(pafs.FSSpecHandler(tos_fsspec))

    if hasattr(daft_filesystem, "_infer_filesystem"):
        original_infer_filesystem = daft_filesystem._infer_filesystem

        def infer_filesystem_with_tos(path: Any, io_config: IOConfig) -> Any:
            parsed = urlparse(str(path))
            if parsed.scheme.lower() == "tos":
                resolved_path = (parsed.netloc + parsed.path).lstrip("/")
                resolved_path = tos_pyarrow_filesystem.normalize_path(resolved_path)
                return resolved_path, tos_pyarrow_filesystem, None
            return original_infer_filesystem(path, io_config)

        daft_filesystem._infer_filesystem = infer_filesystem_with_tos
    elif hasattr(daft_filesystem, "_build_filesystem"):
        original_build_filesystem = daft_filesystem._build_filesystem

        def build_filesystem_with_tos(protocol: str, io_config: IOConfig) -> Any:
            if protocol.lower() == "tos":
                return tos_pyarrow_filesystem, None
            return original_build_filesystem(protocol, io_config)

        daft_filesystem._build_filesystem = build_filesystem_with_tos
    else:
        raise RuntimeError(
            "Unsupported Daft version: cannot locate its Python filesystem factory"
        )

    daft_filesystem._las_tos_writer_enabled = True


def build_demo_write_dataframe() -> tuple[daft.DataFrame, int]:
    """Build one row for the demo schema: uuid int, name string, price double."""
    test_uuid = secrets.randbelow(1_000_000_000)
    arrow_table = pa.table(
        {
            "uuid": pa.array([test_uuid], type=pa.int32()),
            "name": pa.array([f"written_by_daft_{test_uuid}"], type=pa.string()),
            "price": pa.array([99.9], type=pa.float64()),
        }
    )
    return daft.from_arrow(arrow_table), test_uuid


def main() -> None:
    print("Target table:", FULL_TABLE_NAME)
    print("Operation:", OPERATION)
    print("REST URI:", REST_URI)
    print("Region:", REGION)
    print("REST source IP:", REST_SOURCE_IP or "default route")
    print("TOS endpoint:", TOS_ENDPOINT)
    print("Credentials loaded; secrets will not be printed.")

    catalog = create_catalog()
    table = catalog.load_table((NAMESPACE, TABLE_NAME))
    io_config = create_daft_io_config()

    print("\nIceberg table loaded")
    print("Location:", table.location())
    print("Schema:")
    print(table.schema())

    print(f"\nReading the first {READ_LIMIT} rows with Daft...")
    daft.read_iceberg(table, io_config=io_config).limit(READ_LIMIT).show()

    if not ENABLE_WRITE:
        print("\nRead verification succeeded. Write is disabled.")
        print(
            "Set LAS_OPERATION=append or overwrite and provide the matching "
            "LAS_WRITE_CONFIRM value to enable a demo write."
        )
        return

    write_confirmation = os.environ.get("LAS_WRITE_CONFIRM")
    required_confirmation = (
        f"OVERWRITE:{FULL_TABLE_NAME}"
        if OPERATION == "overwrite"
        else FULL_TABLE_NAME
    )
    if write_confirmation != required_confirmation:
        raise RuntimeError(
            "Write confirmation does not match the target table. "
            f"Set LAS_WRITE_CONFIRM={required_confirmation!r} only for a "
            "dedicated test table."
        )

    actual_fields = {
        field.name: str(field.field_type).lower()
        for field in table.schema().fields
    }
    expected_fields = {
        "uuid": "int",
        "name": "string",
        "price": "double",
    }
    if set(actual_fields) != set(expected_fields):
        raise RuntimeError(
            "The bundled write payload only supports the demo schema "
            "(uuid int, name string, price double). Change "
            "build_demo_write_dataframe() to match the target table schema. "
            f"Actual fields: {actual_fields}"
        )
    incompatible_fields = {
        name: {"expected": expected_fields[name], "actual": actual_fields[name]}
        for name in expected_fields
        if actual_fields[name] != expected_fields[name]
    }
    if incompatible_fields:
        raise RuntimeError(
            "The target table field types do not match the bundled demo payload. "
            f"Incompatible fields: {incompatible_fields}"
        )

    enable_daft_tos_writer()
    write_df, test_uuid = build_demo_write_dataframe()

    snapshot_before = table.current_snapshot()
    rows_before = (
        daft.read_iceberg(table, io_config=io_config).count_rows()
        if OPERATION == "overwrite"
        else None
    )
    print(
        "\nSnapshot before:",
        snapshot_before.snapshot_id if snapshot_before else "none",
    )
    if rows_before is not None:
        print("Rows before overwrite:", rows_before)
    print(f"Writing one demo row with Daft in {OPERATION!r} mode...")
    write_df.show()

    write_result = write_df.write_iceberg(
        table,
        mode=OPERATION,
        io_config=io_config,
    )
    print("\nWrite result:")
    write_result.show()
    write_operations = set(write_result.to_pydict()["operation"])
    if "ADD" not in write_operations:
        raise RuntimeError("The Iceberg write did not report an ADD operation")
    if OPERATION == "overwrite" and rows_before and "DELETE" not in write_operations:
        raise RuntimeError(
            "Overwrite verification failed: existing rows were present, but "
            "the write did not report a DELETE operation"
        )

    refreshed_table = catalog.load_table((NAMESPACE, TABLE_NAME))
    snapshot_after = refreshed_table.current_snapshot()
    if snapshot_after is None:
        raise RuntimeError(f"No Iceberg snapshot exists after the {OPERATION}")
    if snapshot_before and snapshot_after.snapshot_id == snapshot_before.snapshot_id:
        raise RuntimeError(
            f"The Iceberg snapshot ID did not change after the {OPERATION}"
        )

    print("Snapshot after:", snapshot_after.snapshot_id)
    print("Reading back the inserted row...")
    result_df = daft.read_iceberg(refreshed_table, io_config=io_config)
    inserted_df = result_df.where(result_df["uuid"] == test_uuid)
    inserted_df.show()

    if inserted_df.count_rows() != 1:
        raise RuntimeError("The newly written row was not found exactly once")

    if OPERATION == "overwrite":
        rows_after = result_df.count_rows()
        print("Rows after overwrite:", rows_after)
        if rows_after != 1:
            raise RuntimeError(
                "Overwrite verification failed: expected the test table to "
                f"contain exactly 1 row, found {rows_after}"
            )

    print(f"{OPERATION.capitalize()} and read-back succeeded. Test uuid:", test_uuid)


if __name__ == "__main__":
    main()

读取 Iceberg 表 #

首次运行建议先执行只读验证：

export LAS_OPERATION="read"
unset LAS_WRITE_CONFIRM

python las_daft_iceberg_example.py

脚本会执行以下操作：

1. 对 LAS Iceberg REST Catalog 请求进行 AK/SK 签名。
2. 通过 PyIceberg 加载目标表和当前 Snapshot。
3. 通过 Daft 读取目标表前 ICEBERG_READ_LIMIT 行数据。
4. 不执行任何写入操作。

成功时会输出：

Read verification succeeded. Write is disabled.

Append 写入并提交 Snapshot #

本操作会真实向 TOS 写入数据文件，并向 LAS Iceberg REST Catalog 提交新的 Snapshot。请确认当前目标为专用测试表。

执行 append #

export ICEBERG_NAMESPACE="<专用测试数据库或 namespace>"
export ICEBERG_TABLE="<专用测试表>"

export LAS_OPERATION="append"
export LAS_WRITE_CONFIRM="${ICEBERG_NAMESPACE}.${ICEBERG_TABLE}"

python las_daft_iceberg_example.py

LAS_WRITE_CONFIRM 必须与完整表名完全一致，否则脚本会拒绝执行写入。

写入和验证逻辑 #

脚本会生成一条符合演示 Schema 的测试记录，并执行：

write_result = write_df.write_iceberg(
    table,
    mode="append",
    io_config=io_config,
)

写入完成后，脚本会：

1. 检查写入结果中是否包含 ADD 操作。
2. 重新从 LAS Iceberg REST Catalog 加载目标表。
3. 确认当前 Snapshot ID 已发生变化。
4. 根据本次生成的 uuid 读取新记录，并确认该记录只存在一条。

成功时会输出类似内容：

Append and read-back succeeded. Test uuid: 123456789

Overwrite 整表覆盖验证 #

高风险操作：overwrite 会使用当前 DataFrame 的内容替换整张表当前可见的数据，不是更新某一条记录。请仅对专用测试表执行。

本示例会在 overwrite 前后统计整张表的行数，因此会扫描全表，仅适用于数据量较小的测试表。

执行 overwrite #

为了降低误操作风险，overwrite 的确认值必须在完整表名前增加 OVERWRITE:：

export ICEBERG_NAMESPACE="<专用测试数据库或 namespace>"
export ICEBERG_TABLE="<专用测试表>"

export LAS_OPERATION="overwrite"
export LAS_WRITE_CONFIRM="OVERWRITE:${ICEBERG_NAMESPACE}.${ICEBERG_TABLE}"

python las_daft_iceberg_example.py

覆盖和验证逻辑 #

脚本会自动完成以下校验：

1. 记录覆盖前的 Snapshot ID 和表总行数。
2. 使用 write_iceberg(mode="overwrite") 写入一条新记录。
3. 检查写入结果中是否包含 ADD；覆盖前存在数据时，还必须包含 DELETE。
4. 重新加载目标表，确认 Snapshot ID 已发生变化。
5. 确认新记录可以读回。
6. 确认测试表当前仅包含本次写入的一条记录。

成功时会输出类似内容：

Rows after overwrite: 1
Overwrite and read-back succeeded. Test uuid: 123456789

overwrite 会让旧数据文件不再属于表的当前 Snapshot，但通常不会立即从 TOS 物理删除。旧 Snapshot 在过期前仍可能引用这些文件，存储空间需要通过 Iceberg Snapshot 过期和孤儿文件清理机制释放。

验证完成后，可以删除本次创建的专用测试表。
如测试过程中产生了多个 Snapshot，请根据实际的数据治理策略执行 Snapshot 过期和孤儿文件清理。Iceberg 写入通常会先将数据文件写入对象存储，再提交 Snapshot；如果任务在数据文件写入成功后、Snapshot 提交前失败，可能产生未被任何 Snapshot 引用的孤儿文件。
完成操作后，建议清理终端中的访问凭证和写入确认变量：

unset VOLC_ACCESSKEY
unset VOLC_SECRETKEY
unset VOLC_SESSION_TOKEN
unset TOS_ACCESS_KEY
unset TOS_SECRET_KEY
unset TOS_SESSION_TOKEN
unset LAS_WRITE_CONFIRM

如不再继续操作，也可以清理其他环境变量：

unset LAS_REGION
unset LAS_ICEBERG_REST_URI
unset LAS_REST_SOURCE_IP
unset TOS_REGION
unset TOS_ENDPOINT
unset ICEBERG_NAMESPACE
unset ICEBERG_TABLE
unset ICEBERG_READ_LIMIT
unset LAS_OPERATION