refactor: replace hand-rolled columnar helpers with mloda's public ones

cli/eval_demo.py

@@ -44,7 +44,7 @@
 
 from mloda.user import Options
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.datasets.text.scifact import ScifactDatasetSource
 from rag_integration.feature_groups.datasets.image.flickr30k import Flickr30kDatasetSource
 from rag_integration.feature_groups.evaluation.metrics import mean_recall_at_k

cli/rag_demo.py

@@ -17,7 +17,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.rag_pipeline import (
     DictDocumentSource,
     FixedSizeChunker,

cli/swap_demo.py

@@ -44,7 +44,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.connectors.generate import ExtractiveResponder, TemplateResponder
 from rag_integration.feature_groups.connectors.orchestrator import HaystackOrchestrator
 from rag_integration.feature_groups.connectors.retrieve import Bm25sRetriever, TfidfRetriever

demo.ipynb

@@ -275,7 +275,7 @@
     "from rag_integration.feature_groups.rag_pipeline.embedding.sentence_transformer import SentenceTransformerEmbedder\n",
     "from rag_integration.feature_groups.rag_pipeline.vector_store.faiss_flat import FaissFlatIndexer\n",
     "from rag_integration.feature_groups.evaluation.faiss_retrieval_evaluator import FaissRetrievalEvaluator\n",
-    "from rag_integration.feature_groups.columnar import columnar_to_rows\n",
+    "from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows\n",
     "\n",
     "feature_name = \"eval_docs__chunked__deduped__embedded__indexed__evaluated\"\n",
     "\n",

rag_integration/feature_groups/connectors/graph_rag/base.py

@@ -41,7 +41,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.connectors.errors import DuplicateDocIdError, InvalidOptionError
 from rag_integration.feature_groups.connectors.mixins import (
     DocCollectionMixin,

rag_integration/feature_groups/datasets/image/base.py

@@ -11,7 +11,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import homogenize_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import homogenize_rows
 
 
 class BaseImageDatasetSource(FeatureGroup):

rag_integration/feature_groups/datasets/text/base.py

@@ -11,7 +11,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import homogenize_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import homogenize_rows
 
 
 class BaseTextDatasetSource(FeatureGroup):

rag_integration/feature_groups/deduplication_base.py

@@ -23,7 +23,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 
 
 class BaseRowDeduplicator(FeatureChainParserMixin, FeatureGroup):
@@ -84,7 +84,7 @@ def _item_size(cls, item: Any) -> int:
         return len(item)
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: FeatureSet) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: FeatureSet) -> List[Dict[str, Any]]:
         """Deduplicate rows: attach duplicate metadata and filter by keep strategy.
 
         Exactly one distinct feature is processed per call. Unlike column-adding feature

rag_integration/feature_groups/evaluation/faiss_retrieval_evaluator.py

@@ -31,7 +31,7 @@
 )
 from mloda.provider import DefaultOptionKeys
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.evaluation.metrics import mean_recall_at_k
 
 
@@ -76,7 +76,7 @@ def compute_framework_rule(cls) -> Optional[Set[Type[ComputeFramework]]]:
         return {PythonDictFramework}
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: FeatureSet) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: FeatureSet) -> List[Dict[str, Any]]:
         """Build FAISS index from corpus embeddings, search with query embeddings, compute Recall@K."""
         try:
             import faiss

rag_integration/feature_groups/evaluation/retrieval_evaluator.py

@@ -15,7 +15,7 @@
 )
 from mloda.provider import DefaultOptionKeys
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.evaluation.metrics import mean_recall_at_k
 
 
@@ -64,7 +64,7 @@ def compute_framework_rule(cls) -> Optional[Set[Type[ComputeFramework]]]:
         return {PythonDictFramework}
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: FeatureSet) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: FeatureSet) -> List[Dict[str, Any]]:
         """Compute Recall@K over the embedded corpus and query rows."""
         try:
             import numpy as np

rag_integration/feature_groups/image_pipeline/embedding/base.py

@@ -13,7 +13,7 @@
 )
 from mloda.provider import DefaultOptionKeys
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 
 
 class BaseImageEmbedder(FeatureChainParserMixin, FeatureGroup):
@@ -143,7 +143,7 @@ def _embed_image(
         ...
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: FeatureSet) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: FeatureSet) -> List[Dict[str, Any]]:
         """Generate embeddings for images, processing row by row for memory efficiency."""
         # mloda 0.9.0 passes columnar data; pivot to rows for row-wise reading.
         rows = columnar_to_rows(data)

rag_integration/feature_groups/image_pipeline/embedding/clip.py

@@ -16,7 +16,7 @@
 from mloda.provider import FeatureSet, property_spec
 from mloda.provider import DefaultOptionKeys
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.image_pipeline.embedding.base import BaseImageEmbedder
 
 # Default local model path: looks two levels above the git repo root (mloda/models/)
@@ -182,7 +182,7 @@ def _embed_text(
         return embedding  # type: ignore[no-any-return]
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: FeatureSet) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: FeatureSet) -> List[Dict[str, Any]]:
         """
         Embed each row using CLIP's vision or text encoder based on row content.
 

rag_integration/feature_groups/image_pipeline/image_source/base.py

@@ -11,7 +11,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import homogenize_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import homogenize_rows
 
 
 class BaseImageSource(FeatureGroup):

rag_integration/feature_groups/image_pipeline/pii_redaction/base.py

@@ -13,7 +13,7 @@
 )
 from mloda.provider import DefaultOptionKeys
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 
 
 class BaseImagePIIRedactor(FeatureChainParserMixin, FeatureGroup):
@@ -159,7 +159,7 @@ def _redact_region_for_feature(
         return cls._redact_region(image_data, image_format, regions)
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: FeatureSet) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: FeatureSet) -> List[Dict[str, Any]]:
         """Perform PII redaction on images, processing row by row for memory efficiency."""
         # mloda 0.9.0 passes columnar data; pivot to rows for row-wise reading.
         rows = columnar_to_rows(data)

rag_integration/feature_groups/image_pipeline/preprocessing/base.py

@@ -13,7 +13,7 @@
 )
 from mloda.provider import DefaultOptionKeys
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 
 
 class BaseImagePreprocessor(FeatureChainParserMixin, FeatureGroup):
@@ -130,7 +130,7 @@ def _preprocess_image(
         ...
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: FeatureSet) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: FeatureSet) -> List[Dict[str, Any]]:
         """Preprocess images row by row for memory efficiency."""
         # mloda 0.9.0 passes columnar data; pivot to rows for row-wise reading.
         rows = columnar_to_rows(data)

rag_integration/feature_groups/rag_pipeline/chunking/base.py

@@ -13,7 +13,7 @@
 )
 from mloda.provider import DefaultOptionKeys
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 
 
 class BaseChunker(FeatureChainParserMixin, FeatureGroup):
@@ -132,7 +132,7 @@ def _chunk_text_for_feature(cls, text: str, feature: Feature) -> List[str]:
         return cls._chunk_text(text, cls._get_chunk_size(feature), cls._get_chunk_overlap(feature))
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: FeatureSet) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: FeatureSet) -> List[Dict[str, Any]]:
         """Perform chunking on the source feature."""
         result = []
 

rag_integration/feature_groups/rag_pipeline/document_source/base.py

@@ -11,7 +11,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import homogenize_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import homogenize_rows
 
 
 class BaseDocumentSource(FeatureGroup):

rag_integration/feature_groups/rag_pipeline/embedding/base.py

@@ -13,7 +13,7 @@
 )
 from mloda.provider import DefaultOptionKeys
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 
 
 class BaseEmbedder(FeatureChainParserMixin, FeatureGroup):
@@ -129,7 +129,7 @@ def _embed_texts(
         ...
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: FeatureSet) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: FeatureSet) -> List[Dict[str, Any]]:
         """Generate embeddings for the source feature, with optional artifact support."""
         artifact_cls = cls.artifact()
 

rag_integration/feature_groups/rag_pipeline/pii_redaction/base.py

@@ -13,7 +13,7 @@
 )
 from mloda.provider import DefaultOptionKeys
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 
 
 class BasePIIRedactor(FeatureChainParserMixin, FeatureGroup):
@@ -174,7 +174,7 @@ def _redact_texts_for_feature(cls, texts: List[str], feature: Feature) -> List[s
         )
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: FeatureSet) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: FeatureSet) -> List[Dict[str, Any]]:
         """Perform PII redaction on the source feature."""
         # mloda 0.9.0 delivers columnar data; read it row-wise.
         rows = columnar_to_rows(data)

rag_integration/feature_groups/rag_pipeline/pii_redaction/pattern.py

@@ -79,7 +79,7 @@ def _get_patterns(cls, feature: Feature) -> Dict[str, Pattern[str]]:
         return patterns
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: Any) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: Any) -> List[Dict[str, Any]]:
         """Extract custom patterns from feature options before redacting."""
         for feature in features.features:
             cls._active_patterns = cls._get_patterns(feature)

rag_integration/feature_groups/rag_pipeline/vector_store/base.py

@@ -13,7 +13,7 @@
 )
 from mloda.provider import DefaultOptionKeys
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.rag_pipeline.vector_store.vector_store_artifact import VectorStoreArtifact
 
 
@@ -91,7 +91,7 @@ def _index_type_name(cls) -> str:
         ...
 
     @classmethod
-    def calculate_feature(cls, data: List[Dict[str, Any]], features: FeatureSet) -> List[Dict[str, Any]]:
+    def calculate_feature(cls, data: Any, features: FeatureSet) -> List[Dict[str, Any]]:
         """Build FAISS index from embeddings, save via artifact, attach row metadata."""
         artifact_cls = cls.artifact()
 

tests/connectors/generate/generate_contract.py

@@ -14,7 +14,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.connectors.generate.base import BaseGenerateConnector
 
 

tests/connectors/graph_rag/graph_rag_contract.py

@@ -18,7 +18,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.connectors.graph_rag.base import BaseGraphRagConnector
 
 

tests/connectors/graph_rag/test_graph_source_chaining.py

@@ -19,7 +19,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.connectors.graph_rag.adjacency_graph_rag import AdjacencyGraphRag
 from rag_integration.feature_groups.connectors.graph_rag.base import BaseGraphRagConnector
 from rag_integration.feature_groups.connectors.graph_rag.kg_source import TriplesKnowledgeGraph
@@ -145,21 +145,21 @@ def test_chained_top_k_applies() -> None:
 def test_graph_source_with_inline_nodes_raises() -> None:
     options = _chained_options(extra={AdjacencyGraphRag.NODES: [{"doc_id": "n", "text": "t"}]})
     with pytest.raises(ValueError, match="one graph only"):
-        AdjacencyGraphRag.calculate_feature([], _feature_set(options))
+        AdjacencyGraphRag.calculate_feature({}, _feature_set(options))
 
 
 def test_graph_source_with_inline_edges_raises() -> None:
     options = _chained_options(extra={AdjacencyGraphRag.EDGES: [["a", "b"]]})
     with pytest.raises(ValueError, match="one graph only"):
-        AdjacencyGraphRag.calculate_feature([], _feature_set(options))
+        AdjacencyGraphRag.calculate_feature({}, _feature_set(options))
 
 
 def test_graph_source_without_upstream_row_raises() -> None:
     with pytest.raises(ValueError, match="produced no row"):
-        AdjacencyGraphRag.calculate_feature([], _feature_set(_chained_options()))
+        AdjacencyGraphRag.calculate_feature({}, _feature_set(_chained_options()))
 
 
 def test_graph_source_with_malformed_payload_raises() -> None:
-    data = [{TriplesKnowledgeGraph.ROOT_FEATURE_NAME: ["not", "a", "dict"]}]
+    data = {TriplesKnowledgeGraph.ROOT_FEATURE_NAME: [["not", "a", "dict"]]}
     with pytest.raises(ValueError, match="nodes"):
         AdjacencyGraphRag.calculate_feature(data, _feature_set(_chained_options()))

tests/connectors/graph_rag/test_kg_source.py

@@ -16,7 +16,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.connectors.graph_rag.kg_source import (
     BaseKnowledgeGraphSource,
     TriplesKnowledgeGraph,

tests/connectors/orchestrator/orchestrator_contract.py

@@ -15,7 +15,7 @@
     PythonDictFramework,
 )
 
-from rag_integration.feature_groups.columnar import columnar_to_rows
+from mloda_plugins.compute_framework.base_implementations.python_dict.python_dict_utils import columnar_to_rows
 from rag_integration.feature_groups.connectors.orchestrator.base import BaseOrchestratorConnector