py: raise error if dataframe has no column names (#1826)

abhizer · web-flow · commit 4bac622e7527 · 2024-06-04T00:31:21.000+05:45
Fixes: #1782 Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com>
diff --git a/python/feldera/_helpers.py b/python/feldera/_helpers.py
@@ -1,6 +1,20 @@
 import pandas as pd
 
 
+def ensure_dataframe_has_columns(df: pd.DataFrame):
+    """
+    Ensures that the DataFrame has column names set.
+    """
+
+    if [v for v in range(df.shape[1])] == list(df.columns):
+        raise ValueError(
+            """
+            DataFrame has no column names set.
+            Input DataFrame must have column names set and they must be consistent with the columns in the input table.
+            """
+        )
+
+
 def dataframe_from_response(buffer: list[list[dict]]):
     """
     Converts the response from Feldera to a pandas DataFrame.
diff --git a/python/feldera/sql_context.py b/python/feldera/sql_context.py
@@ -14,6 +14,7 @@
 from feldera.sql_schema import SQLSchema
 from feldera.output_handler import OutputHandler
 from feldera._callback_runner import CallbackRunner, _CallbackRunnerInstruction
+from feldera._helpers import ensure_dataframe_has_columns
 from enum import Enum
 
 
@@ -226,6 +227,8 @@ def connect_source_pandas(self, table_name: str, df: pandas.DataFrame):
         :param df: The pandas DataFrame to be pushed to the pipeline.
         """
 
+        ensure_dataframe_has_columns(df)
+
         tbl = self.tables.get(table_name)
 
         if tbl:
diff --git a/python/tests/test_wireframes.py b/python/tests/test_wireframes.py
@@ -104,6 +104,19 @@ def callback(df: pd.DataFrame, seq_no: int):
 
         sql.run_to_completion()
 
+    def test_df_without_columns(self):
+
+        sql = SQLContext('df_without_columns', TEST_CLIENT).get_or_create()
+        TBL_NAME = "student"
+
+        df = pd.DataFrame([(1, "a"), (2, "b"), (3, "c")])
+
+        sql.register_table(TBL_NAME, SQLSchema({"id": "INT", "name": "STRING"}))
+        sql.register_view("s", f"SELECT * FROM {TBL_NAME}")
+
+        with self.assertRaises(ValueError):
+            sql.connect_source_pandas(TBL_NAME, df)
+
 
 if __name__ == '__main__':
     unittest.main()
