# Copyright 2020 The Feast Authors

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# https://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import json

from abc import ABC, abstractmethod

from enum import Enum

from typing import TYPE_CHECKING, Dict, Optional

if TYPE_CHECKING:

from feast.protos.feast.core.DataFormat_pb2 import TableFormat as TableFormatProto

class TableFormatType(Enum):

"""Enum for supported table formats"""

DELTA = "delta"

ICEBERG = "iceberg"

HUDI = "hudi"

class TableFormat(ABC):

"""

Abstract base class for table formats.

Table formats encapsulate metadata and configuration specific to different

table storage formats like Iceberg, Delta Lake, Hudi, etc. They provide

a unified interface for configuring table-specific properties that are

used when reading from or writing to these advanced table formats.

This base class defines the contract that all table format implementations

must follow, including serialization/deserialization capabilities and

property management.

Attributes:

format_type (TableFormatType): The type of table format (iceberg, delta, hudi).

properties (Dict[str, str]): Dictionary of format-specific properties.

Examples:

Table formats are typically used with data sources to specify

advanced table metadata and reading options:

>>> from feast.table_format import IcebergFormat

>>> iceberg_format = IcebergFormat(

... catalog="my_catalog",

... namespace="my_namespace"

... )

>>> iceberg_format.set_property("snapshot-id", "123456789")

"""

def __init__(

self, format_type: TableFormatType, properties: Optional[Dict[str, str]] = None

):

self.format_type = format_type

self.properties = properties or {}

@abstractmethod

def to_dict(self) -> Dict:

"""Convert table format to dictionary representation"""

pass

@classmethod

@abstractmethod

def from_dict(cls, data: Dict) -> "TableFormat":

"""Create table format from dictionary representation"""

pass

def get_property(self, key: str, default: Optional[str] = None) -> Optional[str]:

"""Get a table format property"""

return self.properties.get(key, default)

def set_property(self, key: str, value: str) -> None:

"""Set a table format property"""

self.properties[key] = value

class IcebergFormat(TableFormat):

"""

Apache Iceberg table format configuration.

Iceberg is an open table format for huge analytic datasets. This class provides

configuration for Iceberg-specific properties including catalog configuration,

namespace settings, and table-level properties for reading and writing Iceberg tables.

Args:

catalog (Optional[str]): Name of the Iceberg catalog to use. The catalog manages

table metadata and provides access to tables.

namespace (Optional[str]): Namespace (schema/database) within the catalog where

the table is located.

properties (Optional[Dict[str, str]]): Properties for configuring Iceberg

catalog and table operations (e.g., warehouse location, snapshot-id,

as-of-timestamp, file format, compression, partitioning).

Attributes:

catalog (str): The Iceberg catalog name.

namespace (str): The namespace within the catalog.

properties (Dict[str, str]): Iceberg configuration properties.

Examples:

Basic Iceberg configuration:

>>> iceberg_format = IcebergFormat(

... catalog="my_catalog",

... namespace="my_database"

... )

Advanced configuration with properties:

>>> iceberg_format = IcebergFormat(

... catalog="spark_catalog",

... namespace="lakehouse",

... properties={

... "warehouse": "s3://my-bucket/warehouse",

... "catalog-impl": "org.apache.iceberg.spark.SparkCatalog",

... "format-version": "2",

... "write.parquet.compression-codec": "snappy"

... }

... )

Reading from a specific snapshot:

>>> iceberg_format = IcebergFormat(catalog="my_catalog", namespace="db")

>>> iceberg_format.set_property("snapshot-id", "123456789")

Time travel queries:

>>> iceberg_format.set_property("as-of-timestamp", "1648684800000")

"""

def __init__(

self,

catalog: Optional[str] = None,

namespace: Optional[str] = None,

properties: Optional[Dict[str, str]] = None,

):

super().__init__(TableFormatType.ICEBERG, properties)

self.catalog = catalog

self.namespace = namespace

# Add catalog and namespace to properties if provided

if catalog:

self.properties["iceberg.catalog"] = catalog

if namespace:

self.properties["iceberg.namespace"] = namespace

def to_dict(self) -> Dict:

return {

"format_type": self.format_type.value,

"catalog": self.catalog,

"namespace": self.namespace,

"properties": self.properties,

}

@classmethod

def from_dict(cls, data: Dict) -> "IcebergFormat":

return cls(

catalog=data.get("catalog"),

namespace=data.get("namespace"),

properties=data.get("properties", {}),

)

def to_proto(self) -> "TableFormatProto":

"""Convert to protobuf TableFormat message"""

from feast.protos.feast.core.DataFormat_pb2 import (

TableFormat as TableFormatProto,

)

iceberg_proto = TableFormatProto.IcebergFormat(

catalog=self.catalog or "",

namespace=self.namespace or "",

properties=self.properties,

)

return TableFormatProto(iceberg_format=iceberg_proto)

@classmethod

def from_proto(cls, proto: "TableFormatProto") -> "IcebergFormat":

"""Create from protobuf TableFormat message"""

iceberg_proto = proto.iceberg_format

return cls(

catalog=iceberg_proto.catalog if iceberg_proto.catalog else None,

namespace=iceberg_proto.namespace if iceberg_proto.namespace else None,

properties=dict(iceberg_proto.properties),

)

class DeltaFormat(TableFormat):

"""

Delta Lake table format configuration.

Delta Lake is an open-source storage layer that brings ACID transactions to Apache Spark

and big data workloads. This class provides configuration for Delta-specific properties

including table properties, checkpoint locations, and versioning options.

Args:

checkpoint_location (Optional[str]): Location for storing Delta transaction logs

and checkpoints. Required for streaming operations.

properties (Optional[Dict[str, str]]): Properties for configuring Delta table

behavior (e.g., auto-optimize, vacuum settings, data skipping).

Attributes:

checkpoint_location (str): Path to checkpoint storage location.

properties (Dict[str, str]): Delta table configuration properties.

Examples:

Basic Delta configuration:

>>> delta_format = DeltaFormat()

Configuration with table properties:

>>> delta_format = DeltaFormat(

... properties={

... "delta.autoOptimize.optimizeWrite": "true",

... "delta.autoOptimize.autoCompact": "true",

... "delta.tuneFileSizesForRewrites": "true"

... }

... )

Streaming configuration with checkpoint:

>>> delta_format = DeltaFormat(

... checkpoint_location="s3://my-bucket/checkpoints/my_table"

... )

Time travel - reading specific version:

>>> delta_format = DeltaFormat()

>>> delta_format.set_property("versionAsOf", "5")

Time travel - reading at specific timestamp:

>>> delta_format.set_property("timestampAsOf", "2023-01-01 00:00:00")

"""

def __init__(

self,

checkpoint_location: Optional[str] = None,

properties: Optional[Dict[str, str]] = None,

):

super().__init__(TableFormatType.DELTA, properties)

self.checkpoint_location = checkpoint_location

# Add checkpoint location to properties if provided

if checkpoint_location:

self.properties["delta.checkpointLocation"] = checkpoint_location

def to_dict(self) -> Dict:

return {

"format_type": self.format_type.value,

"checkpoint_location": self.checkpoint_location,

"properties": self.properties,

}

@classmethod

def from_dict(cls, data: Dict) -> "DeltaFormat":

return cls(

checkpoint_location=data.get("checkpoint_location"),

properties=data.get("properties", {}),

)

def to_proto(self) -> "TableFormatProto":

"""Convert to protobuf TableFormat message"""

from feast.protos.feast.core.DataFormat_pb2 import (

TableFormat as TableFormatProto,

)

delta_proto = TableFormatProto.DeltaFormat(

checkpoint_location=self.checkpoint_location or "",

properties=self.properties,

)

return TableFormatProto(delta_format=delta_proto)

@classmethod

def from_proto(cls, proto: "TableFormatProto") -> "DeltaFormat":

"""Create from protobuf TableFormat message"""

delta_proto = proto.delta_format

return cls(

checkpoint_location=delta_proto.checkpoint_location

if delta_proto.checkpoint_location

else None,

properties=dict(delta_proto.properties),

)

class HudiFormat(TableFormat):

"""

Apache Hudi table format configuration.

Apache Hudi is a data management framework used to simplify incremental data processing

and data pipeline development. This class provides configuration for Hudi-specific

properties including table type, record keys, and write operations.

Args:

table_type (Optional[str]): Type of Hudi table. Options are:

- "COPY_ON_WRITE": Stores data in columnar format (Parquet) and rewrites entire files

- "MERGE_ON_READ": Stores data using combination of columnar and row-based formats

record_key (Optional[str]): Field(s) that uniquely identify a record. Can be a single

field or comma-separated list for composite keys.

precombine_field (Optional[str]): Field used to determine the latest version of a record

when multiple updates exist (usually a timestamp or version field).

properties (Optional[Dict[str, str]]): Additional Hudi table properties for

configuring compaction, indexing, and other Hudi features.

Attributes:

table_type (str): The Hudi table type (COPY_ON_WRITE or MERGE_ON_READ).

record_key (str): The record key field(s).

precombine_field (str): The field used for record deduplication.

properties (Dict[str, str]): Additional Hudi configuration properties.

Examples:

Basic Hudi configuration:

>>> hudi_format = HudiFormat(

... table_type="COPY_ON_WRITE",

... record_key="user_id",

... precombine_field="timestamp"

... )

Configuration with composite record key:

>>> hudi_format = HudiFormat(

... table_type="MERGE_ON_READ",

... record_key="user_id,event_type",

... precombine_field="event_timestamp"

... )

Advanced configuration with table properties:

>>> hudi_format = HudiFormat(

... table_type="COPY_ON_WRITE",

... record_key="id",

... precombine_field="updated_at",

... properties={

... "hoodie.compaction.strategy": "org.apache.hudi.table.action.compact.strategy.LogFileSizeBasedCompactionStrategy",

... "hoodie.index.type": "BLOOM",

... "hoodie.bloom.index.parallelism": "100"

... }

... )

Reading incremental data:

>>> hudi_format = HudiFormat(table_type="COPY_ON_WRITE")

>>> hudi_format.set_property("hoodie.datasource.query.type", "incremental")

>>> hudi_format.set_property("hoodie.datasource.read.begin.instanttime", "20230101000000")

"""

def __init__(

self,

table_type: Optional[str] = None, # COPY_ON_WRITE or MERGE_ON_READ

record_key: Optional[str] = None,

precombine_field: Optional[str] = None,

properties: Optional[Dict[str, str]] = None,

):

super().__init__(TableFormatType.HUDI, properties)

self.table_type = table_type

self.record_key = record_key

self.precombine_field = precombine_field

# Add Hudi-specific properties if provided

if table_type:

self.properties["hoodie.datasource.write.table.type"] = table_type

if record_key:

self.properties["hoodie.datasource.write.recordkey.field"] = record_key

if precombine_field:

self.properties["hoodie.datasource.write.precombine.field"] = (

precombine_field

)

def to_dict(self) -> Dict:

return {

"format_type": self.format_type.value,

"table_type": self.table_type,

"record_key": self.record_key,

"precombine_field": self.precombine_field,

"properties": self.properties,

}

@classmethod

def from_dict(cls, data: Dict) -> "HudiFormat":

return cls(

table_type=data.get("table_type"),

record_key=data.get("record_key"),

precombine_field=data.get("precombine_field"),

properties=data.get("properties", {}),

)

def to_proto(self) -> "TableFormatProto":

"""Convert to protobuf TableFormat message"""

from feast.protos.feast.core.DataFormat_pb2 import (

TableFormat as TableFormatProto,

)

hudi_proto = TableFormatProto.HudiFormat(

table_type=self.table_type or "",

record_key=self.record_key or "",

precombine_field=self.precombine_field or "",

properties=self.properties,

)

return TableFormatProto(hudi_format=hudi_proto)

@classmethod

def from_proto(cls, proto: "TableFormatProto") -> "HudiFormat":

"""Create from protobuf TableFormat message"""

hudi_proto = proto.hudi_format

return cls(

table_type=hudi_proto.table_type if hudi_proto.table_type else None,

record_key=hudi_proto.record_key if hudi_proto.record_key else None,

precombine_field=hudi_proto.precombine_field

if hudi_proto.precombine_field

else None,

properties=dict(hudi_proto.properties),

)

def create_table_format(format_type: TableFormatType, **kwargs) -> TableFormat:

"""

Factory function to create appropriate TableFormat instance based on type.

This is a convenience function that creates the correct TableFormat subclass

based on the provided format type, passing through any additional keyword arguments

to the constructor.

Args:

format_type (TableFormatType): The type of table format to create.

**kwargs: Additional keyword arguments passed to the format constructor.

Returns:

TableFormat: An instance of the appropriate TableFormat subclass.

Raises:

ValueError: If an unsupported format_type is provided.

Examples:

Create an Iceberg format:

>>> iceberg_format = create_table_format(

... TableFormatType.ICEBERG,

... catalog="my_catalog",

... namespace="my_db"

... )

Create a Delta format:

>>> delta_format = create_table_format(

... TableFormatType.DELTA,

... checkpoint_location="s3://bucket/checkpoints"

... )

"""

if format_type == TableFormatType.ICEBERG:

return IcebergFormat(**kwargs)

elif format_type == TableFormatType.DELTA:

return DeltaFormat(**kwargs)

elif format_type == TableFormatType.HUDI:

return HudiFormat(**kwargs)

else:

raise ValueError(f"Unknown table format type: {format_type}")

def table_format_from_dict(data: Dict) -> TableFormat:

"""

Create TableFormat instance from dictionary representation.

This function deserializes a dictionary (typically from JSON or protobuf)

back into the appropriate TableFormat instance. The dictionary must contain

a 'format_type' field that indicates which format class to instantiate.

Args:

data (Dict): Dictionary containing table format configuration. Must include

'format_type' field with value 'iceberg', 'delta', or 'hudi'.

Returns:

TableFormat: An instance of the appropriate TableFormat subclass.

Raises:

ValueError: If format_type is not recognized.

KeyError: If format_type field is missing from data.

Examples:

Deserialize an Iceberg format:

>>> data = {

... "format_type": "iceberg",

... "catalog": "my_catalog",

... "namespace": "my_db"

... }

>>> iceberg_format = table_format_from_dict(data)

"""

if "format_type" not in data:

raise KeyError("Missing 'format_type' field in data")

format_type = data["format_type"]

if format_type == TableFormatType.ICEBERG.value:

return IcebergFormat.from_dict(data)

elif format_type == TableFormatType.DELTA.value:

return DeltaFormat.from_dict(data)

elif format_type == TableFormatType.HUDI.value:

return HudiFormat.from_dict(data)

else:

raise ValueError(f"Unknown table format type: {format_type}")

def table_format_from_json(json_str: str) -> TableFormat:

"""

Create TableFormat instance from JSON string.

This is a convenience function that parses a JSON string and creates

the appropriate TableFormat instance. Useful for loading table format

configurations from files or network requests.

Args:

json_str (str): JSON string containing table format configuration.

Returns:

TableFormat: An instance of the appropriate TableFormat subclass.

Raises:

json.JSONDecodeError: If the JSON string is invalid.

ValueError: If format_type is not recognized.

KeyError: If format_type field is missing.

Examples:

Load from JSON string:

>>> json_config = '{"format_type": "delta", "checkpoint_location": "s3://bucket/checkpoints"}'

>>> delta_format = table_format_from_json(json_config)

"""

data = json.loads(json_str)

return table_format_from_dict(data)

def table_format_from_proto(proto: "TableFormatProto") -> TableFormat:

"""

Create TableFormat instance from protobuf TableFormat message.

Args:

proto: TableFormat protobuf message

Returns:

TableFormat: An instance of the appropriate TableFormat subclass.

Raises:

ValueError: If the proto doesn't contain a recognized format.

"""

which_format = proto.WhichOneof("format")

if which_format == "iceberg_format":

return IcebergFormat.from_proto(proto)

elif which_format == "delta_format":

return DeltaFormat.from_proto(proto)

elif which_format == "hudi_format":

return HudiFormat.from_proto(proto)

else:

raise ValueError(f"Unknown table format in proto: {which_format}")