@@ -76,36 +76,11 @@ def to_df(
            validation_reference (optional): The validation to apply against the retrieved dataframe.
            timeout (optional): The query timeout if applicable.
        """
-        features_df = self._to_df_internal(timeout=timeout)
-
-        if self.on_demand_feature_views:
-            # TODO(adchia): Fix requirement to specify dependent feature views in feature_refs
-            for odfv in self.on_demand_feature_views:
-                if odfv.mode not in {"pandas", "substrait"}:
-                    raise Exception(
-                        f'OnDemandFeatureView mode "{odfv.mode}" not supported for offline processing.'
-                    )
-                features_df = features_df.join(
-                    odfv.get_transformed_features_df(
-                        features_df,
-                        self.full_feature_names,
-                    )
-                )
-
-        if validation_reference:
-            if not flags_helper.is_test():
-                warnings.warn(
-                    "Dataset validation is an experimental feature. "
-                    "This API is unstable and it could and most probably will be changed in the future. "
-                    "We do not guarantee that future changes will maintain backward compatibility.",
-                    RuntimeWarning,
-                )
-
-            validation_result = validation_reference.profile.validate(features_df)
-            if not validation_result.is_success:
-                raise ValidationFailed(validation_result)
-
-        return features_df

    def to_arrow(
        self,
@@ -122,23 +97,20 @@ def to_arrow(
            validation_reference (optional): The validation to apply against the retrieved dataframe.
            timeout (optional): The query timeout if applicable.
        """
-        if not self.on_demand_feature_views and not validation_reference:
-            return self._to_arrow_internal(timeout=timeout)
-
-        features_df = self._to_df_internal(timeout=timeout)
        if self.on_demand_feature_views:
            for odfv in self.on_demand_feature_views:
-                if odfv.mode not in {"pandas", "substrait"}:
-                    raise Exception(
-                        f'OnDemandFeatureView mode "{odfv.mode}" not supported for offline processing.'
-                    )
-                features_df = features_df.join(
-                    odfv.get_transformed_features_df(
-                        features_df,
-                        self.full_feature_names,
-                    )
                )

        if validation_reference:
            if not flags_helper.is_test():
                warnings.warn(
@@ -148,11 +120,13 @@ def to_arrow(
                    RuntimeWarning,
                )

-            validation_result = validation_reference.profile.validate(features_df)
            if not validation_result.is_success:
                raise ValidationFailed(validation_result)

-        return pyarrow.Table.from_pandas(features_df)

    def to_sql(self) -> str:
        """

@@ -3,6 +3,7 @@

import dill
import pandas as pd

from feast.field import Field, from_value_type
from feast.protos.feast.core.Transformation_pb2 import (
@@ -26,6 +27,19 @@ def __init__(self, udf: FunctionType, udf_string: str = ""):
        self.udf = udf
        self.udf_string = udf_string

    def transform(self, input_df: pd.DataFrame) -> pd.DataFrame:
        if not isinstance(input_df, pd.DataFrame):
            raise TypeError(

@@ -2,6 +2,7 @@
from typing import Any, Dict, List

import dill

from feast.field import Field, from_value_type
from feast.protos.feast.core.Transformation_pb2 import (
@@ -24,6 +25,11 @@ def __init__(self, udf: FunctionType, udf_string: str = ""):
        self.udf = udf
        self.udf_string = udf_string

    def transform(self, input_dict: Dict) -> Dict:
        if not isinstance(input_dict, Dict):
            raise TypeError(

@@ -34,6 +34,15 @@ def table_provider(names, schema: pyarrow.Schema):
        ).read_all()
        return table.to_pandas()

    def infer_features(self, random_input: Dict[str, List[Any]]) -> List[Field]:
        df = pd.DataFrame.from_dict(random_input)
        output_df: pd.DataFrame = self.transform(df)

@@ -216,7 +216,7 @@ def test_to_sql():

@pytest.mark.parametrize("timeout", (None, 30))
def test_to_df_timeout(retrieval_job, timeout: Optional[int]):
-    with patch.object(retrieval_job, "_to_df_internal") as mock_to_df_internal:
        retrieval_job.to_df(timeout=timeout)
        mock_to_df_internal.assert_called_once_with(timeout=timeout)