feat(mlop-2760): Create metadata module (feature, feature_set, reader, writer and pipeline) (#418)

# Why

This change introduces a robust metadata generation system for
Butterfree components. The primary motivation is to create a
standardized and automated way to:

- Document the schema of feature set pipelines and their components.
- Facilitate serialization of pipeline metadata for various use cases,
such as cataloging, lineage tracking, and integration with other tools.

**This will improve the maintainability, understandability, and
interoperability of feature set pipelines within the Butterfree
ecosystem.**

# What

## New `metadata` module:

- Introduced a new `butterfree/metadata` directory to house all Pydantic
models for metadata.

## Dependencies

     - `Pydantic` added to requirements

## Tests

    - Unit tests for EVERY behavior of new class
- Integration tests for pipeline with `FeatureSet` and with
`AggregatedFeatureSet`

# How

The solution refactors and centralizes metadata definition using
Pydantic models within a dedicated `butterfree/metadata` directory.

1.  **Hierarchical Metadata Construction:**
* Individual components (`Feature`, `KeyFeature`, `TimestampFeature`,
`Window`, `Reader` subclasses, `Writer`) are responsible for building
their own metadata Pydantic models (`FeatureMetadata`, `ReaderMetadata`
variants, `WriterMetadata`).
   
* Container classes (`FeatureSet`, `AggregatedFeatureSet`,
`FeatureSetPipeline`) aggregate metadata from their constituent parts.
For instance, `FeatureSet.build_metadata()` calls `build_metadata()` on
its keys, timestamp, and then constructs metadata for its transformed
features before packaging it all into a `FeatureSetMetadata` object.
  
2. **Pydantic Model Usage:** Each metadata type (e.g., for a feature, a
reader, a feature set) has a corresponding Pydantic model ensuring
structure and validation.

Author: albjoaov

.gitignore

@@ -121,3 +121,5 @@ init/
 
 # integration tests artifacts
 metastore_db/
+
+.DS_Store

butterfree/extract/readers/file_reader.py

@@ -7,6 +7,7 @@
 
 from butterfree.clients import SparkClient
 from butterfree.extract.readers.reader import Reader
+from butterfree.metadata.reader_metadata import FileReaderMetadata
 
 
 class FileReader(Reader):
@@ -117,3 +118,11 @@ def consume(self, client: SparkClient) -> DataFrame:
             path=self.path,
             **self.options,
         )
+
+    def build_metadata(self) -> FileReaderMetadata:
+        """Build the metadata for the file reader."""
+        return FileReaderMetadata(
+            path=self.path,
+            format=self.format,
+            incremental_strategy=self.incremental_strategy is not None,
+        )

butterfree/extract/readers/kafka_reader.py

@@ -9,6 +9,7 @@
 from butterfree.configs import environment
 from butterfree.extract.pre_processing import explode_json_column
 from butterfree.extract.readers.reader import Reader
+from butterfree.metadata.reader_metadata import KafkaReaderMetadata
 
 
 class KafkaReader(Reader):
@@ -182,3 +183,9 @@ def consume(self, client: SparkClient) -> DataFrame:
 
         # apply schema defined in self.value_schema
         return self._struct_df(raw_df)
+
+    def build_metadata(self) -> KafkaReaderMetadata:
+        """Build the metadata for the kafka reader."""
+        return KafkaReaderMetadata(
+            topic=self.topic,
+        )

butterfree/extract/readers/reader.py

@@ -2,20 +2,25 @@
 
 from abc import ABC, abstractmethod
 from functools import reduce
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional, Union
 
 from pyspark.sql import DataFrame
 
 from butterfree.clients import SparkClient
 from butterfree.dataframe_service import IncrementalStrategy
 from butterfree.hooks import HookableComponent
+from butterfree.metadata.reader_metadata import (
+    FileReaderMetadata,
+    KafkaReaderMetadata,
+    TableReaderMetadata,
+)
 
 
 class Reader(ABC, HookableComponent):
     """Abstract base class for Readers.
 
     Attributes:
-        id: unique string id for register the reader as a view on the metastore.
+        id: unique string id for register the reader as a view.
         transformations: list os methods that will be applied over the dataframe
             after the raw data is extracted.
 
@@ -139,3 +144,10 @@ def _apply_transformations(self, df: DataFrame) -> DataFrame:
             self.transformations,
             df,
         )
+
+    @abstractmethod
+    def build_metadata(
+        self,
+    ) -> Union[FileReaderMetadata, KafkaReaderMetadata, TableReaderMetadata]:
+        """Abstract method to build the metadata for reader type."""
+        pass

butterfree/extract/readers/table_reader.py

@@ -6,6 +6,7 @@
 
 from butterfree.clients import SparkClient
 from butterfree.extract.readers.reader import Reader
+from butterfree.metadata.reader_metadata import TableReaderMetadata
 
 
 class TableReader(Reader):
@@ -66,3 +67,11 @@ def consume(self, client: SparkClient) -> DataFrame:
 
         """
         return client.read_table(self.table, self.database)
+
+    def build_metadata(self) -> TableReaderMetadata:
+        """Build the metadata for the table reader."""
+        return TableReaderMetadata(
+            table=self.table,
+            database=self.database,
+            incremental_strategy=self.incremental_strategy is not None,
+        )

butterfree/load/writers/writer.py

@@ -9,6 +9,7 @@
 from butterfree.clients import SparkClient
 from butterfree.configs.db import AbstractWriteConfig
 from butterfree.hooks import HookableComponent
+from butterfree.metadata.writer_metadata import WriterMetadata
 from butterfree.transform import FeatureSet
 
 
@@ -122,3 +123,22 @@ def validate(
             AssertionError: if validation fails.
 
         """
+
+    def build_metadata(self) -> WriterMetadata:
+        """Get the writer's metadata as a Pydantic model.
+
+        This method creates a standardized representation of writer metadata
+        that can be used for documentation, validation, and serialization purposes.
+
+        Returns:
+            A BaseWriterMetadata model containing the writer's metadata
+        """
+
+        writer_metadata = WriterMetadata(
+            type=self.__class__.__name__,
+            interval_mode=self.interval_mode,
+            write_to_entity=self.write_to_entity,
+            db_config=self.db_config.__class__.__name__,
+        )
+
+        return writer_metadata

butterfree/metadata/__init__.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class FeatureMetadata(BaseModel):
+    """Metadata model for a column in a feature set.
+
+    This model represents the metadata of a single column in a feature set,
+    including its name, data type, and whether it's a primary key.
+    """
+
+    name: str = Field(..., description="The name of the column")
+    data_type: str = Field(
+        ...,
+        description=(
+            "The data type of the column (e.g., StringType, IntegerType) represented by pyspark.sql.types"  # noqa: E501
+        ),
+    )
+    description: str = Field(
+        ...,
+        description="The description of the column",
+    )
+    primary_key: bool = Field(
+        ...,
+        description="Whether the column is a primary (or partition if it's a Cassandra table) key",  # noqa: E501
+    )

butterfree/metadata/feature_metadata.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from typing import List, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+from butterfree.metadata.feature_metadata import FeatureMetadata
+
+
+class FeatureSetMetadata(BaseModel):
+    """Metadata model for a feature set catalog.
+
+    This model represents the catalog information of a feature set,
+    including its name, description, and column definitions.
+    """
+
+    entity: str = Field(
+        ..., description="The entity type associated with the feature set"
+    )
+    name: str = Field(..., description="The name of the Feature Set")
+    type: Literal["FeatureSet", "AggregatedFeatureSet"] = Field(
+        ..., description="The type of feature set"
+    )
+    description: str = Field(..., description="The description of the Feature Set")
+    windows_definition: Optional[List[str]] = Field(
+        None, description="The definition of the windows for the feature set"
+    )
+    features: List[FeatureMetadata] = Field(
+        ..., description="A list of column definitions"
+    )

butterfree/metadata/feature_set_metadata.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from typing import List, Union
+
+from pydantic import BaseModel, Field
+from typing_extensions import Annotated
+
+from butterfree.load.writers.writer import WriterMetadata
+from butterfree.metadata.feature_set_metadata import FeatureSetMetadata
+from butterfree.metadata.reader_metadata import (
+    FileReaderMetadata,
+    KafkaReaderMetadata,
+    TableReaderMetadata,
+)
+
+
+class FeatureSetPipelineMetadata(BaseModel):
+    """Metadata model for a feature set pipeline.
+
+    This model represents the complete metadata of a feature set pipeline,
+    including its configuration, data sources, output schema, and processing details.
+    """
+
+    feature_set: FeatureSetMetadata = Field(
+        ..., description="Metadata about the feature set's output"
+    )
+
+    # Required for correct serialization using Union
+    readers: List[
+        Annotated[
+            Union[FileReaderMetadata, KafkaReaderMetadata, TableReaderMetadata],
+            Field(discriminator="type"),
+        ]
+    ] = Field(
+        ...,
+        description="A list of data sources required to generate the feature set",
+    )
+
+    writers: List[WriterMetadata] = Field(
+        ..., description="The writers to be used for the feature set"
+    )