vdk-core: Support for dataframe ingestion (#3181)

This PR amends the send_object_for_ingestion method to be able to accept
a pandas DataFrame object, extract its column_names and rows in list
form, and pass them on to the
send_tabular_data_for_ingestion method.

---------

Signed-off-by: Gabriel Georgiev <gageorgiev@vmware.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: gabrielgeorgiev1

projects/vdk-core/src/vdk/api/job_input.py

@@ -150,12 +150,18 @@ def send_object_for_ingestion(
         The method then returns immediately unless ingester_wait_to_finish_after_every_send flag is set to true.
         The data is then processed and sent asynchronously by multiple background threads in parallel
          (the level of parallelism could be controlled by ingester_number_of_worker_threads configuration).
+        Users should not modify a payload object after it has been passed to a send_object_for_ingestion call,
+        as this can be thread-unsafe and cause inconsistensies in the data.
 
         Arguments:
-            payload: dict
+            payload: Union[dict, "pandas.DataFrame"]
                 The passed object will be translated to a row in destination table.
                 Keys of the object are translated to columns in the table and values will populate a single row.
 
+                The payload argument can be either a dict or a pandas.DataFrame.
+                The pandas library is an optional dependency; ensure it is installed and included in
+                the job's requirements before deployment.
+
                 Note:
                     This method hides technical complexities around @type and @id
                     described in the specification, still you have the freedom to

projects/vdk-core/src/vdk/internal/builtin_plugins/ingestion/ingester_base.py

@@ -10,6 +10,7 @@
 from typing import List
 from typing import Optional
 from typing import Tuple
+from typing import Union
 
 from vdk.api.job_input import IIngester
 from vdk.api.plugin.plugin_input import IIngesterPlugin
@@ -120,7 +121,7 @@ def __init__(
 
     def send_object_for_ingestion(
         self,
-        payload: dict,
+        payload: Union[dict, "pandas.DataFrame"],
         destination_table: Optional[str],
         method: Optional[str],
         target: Optional[str] = None,
@@ -129,6 +130,16 @@ def send_object_for_ingestion(
         """
         See parent doc
         """
+        if self.__object_is_data_frame(payload):
+            return self.send_tabular_data_for_ingestion(
+                rows=payload.itertuples(index=False),
+                column_names=payload.columns.tolist(),
+                destination_table=destination_table,
+                method=method,
+                target=target,
+                collection_id=collection_id,
+            )
+
         if collection_id is None:
             collection_id = "{data_job_name}|{execution_id}".format(
                 data_job_name=self._data_job_name, execution_id=self._op_id
@@ -703,3 +714,16 @@ def __verify_payload_format(self, payload_dict: dict):
                     resolvable_by=ResolvableBy.USER_ERROR,
                 )
             )
+
+    @staticmethod
+    def __object_is_data_frame(obj: Union[dict, "pandas.DataFrame"]) -> bool:
+        log.debug("Checking if object to be ingested is a DataFrame")
+        try:
+            import pandas
+
+            return isinstance(obj, pandas.DataFrame)
+        except ImportError:
+            log.debug(
+                "`pandas` package not found in the current environment, object is (probably) not a dataframe"
+            )
+            return False

projects/vdk-core/tests/functional/ingestion/jobs/ingest-job/1_step.py

@@ -1,5 +1,6 @@
 # Copyright 2023-2024 Broadcom
 # SPDX-License-Identifier: Apache-2.0
+from pandas import DataFrame
 from vdk.api.job_input import IJobInput
 
 
@@ -23,3 +24,9 @@ def run(job_input: IJobInput):
         destination_table="tabular_table",
         method="memory",
     )
+
+    df = DataFrame.from_dict({"A": [1], "B": [2], "C": [3]})
+
+    job_input.send_object_for_ingestion(
+        payload=df, destination_table="dataframe_table", method="memory"
+    )

projects/vdk-core/tests/functional/ingestion/test_run_ingest.py

@@ -52,6 +52,10 @@ def test_run_ingest():
     assert ingest_plugin.payloads[1].payload[0] == expected_rows_object
     assert ingest_plugin.payloads[1].destination_table == "tabular_table"
 
+    expected_df_object = {"A": 1, "B": 2, "C": 3}
+    assert ingest_plugin.payloads[2].payload[0] == expected_df_object
+    assert ingest_plugin.payloads[2].destination_table == "dataframe_table"
+
 
 def test_run_ingest_fails():
     runner = CliEntryBasedTestRunner(FailingIngestIntoMemoryPlugin())