Fix Kinetica datasource/datasink bugs

- Remove redundant value is not None checks in convert_arrow_batch_to_records
  (the None case is already handled earlier in the function)
- Recompute records_per_task after parallelism cap in get_read_tasks to ensure
  even distribution of work across tasks
- Pass PyArrow schema instead of Ray Data Schema to KineticaDatasink in
  write_kinetica to fix type mismatch

Signed-off-by: anindyam1969 <amukherjee@kinetica.com>

Author: am-kinetica

python/ray/data/_internal/datasource/kinetica_datasource.py

@@ -623,6 +623,10 @@ def get_read_tasks(
         # This handles the case where parallelism > total_count
         effective_parallelism = min(effective_parallelism, self._total_count)
 
+        # Recompute records_per_task after capping parallelism to ensure
+        # even distribution of work across tasks
+        records_per_task = max(1, self._total_count // effective_parallelism)
+
         read_tasks = []
         offset = 0
 

python/ray/data/_internal/datasource/kinetica_type_utils.py

@@ -604,19 +604,22 @@ def convert_arrow_batch_to_records(
                 elif isinstance(value, date):
                     record[col_name] = value.isoformat()
                 else:
-                    record[col_name] = str(value) if value is not None else None
+                    # value is not None here (handled by if block at line 569)
+                    record[col_name] = str(value)
             elif col_type == "time":
                 # Convert time to ISO format string (HH:MM:SS.ffffff)
                 if isinstance(value, time):
                     record[col_name] = value.isoformat()
                 else:
-                    record[col_name] = str(value) if value is not None else None
+                    # value is not None here (handled by if block at line 569)
+                    record[col_name] = str(value)
             elif col_type in ("datetime", "timestamp"):
                 # Convert datetime to ISO format string (YYYY-MM-DDTHH:MM:SS.ffffff)
                 if isinstance(value, datetime):
                     record[col_name] = value.isoformat()
                 else:
-                    record[col_name] = str(value) if value is not None else None
+                    # value is not None here (handled by if block at line 569)
+                    record[col_name] = str(value)
             else:
                 # Handle any remaining date/time types that weren't detected
                 # by column properties

python/ray/data/dataset.py

@@ -5273,13 +5273,20 @@ def write_kinetica(
         """
         from ray.data._internal.datasource.kinetica_datasink import KineticaDatasink
 
+        # Extract the underlying PyArrow schema from Ray Data Schema.
+        # KineticaDatasink expects pa.Schema, not ray.data.Schema.
+        ray_schema = self.schema()
+        pa_schema = (
+            ray_schema.base_schema if hasattr(ray_schema, "base_schema") else ray_schema
+        )
+
         datasink = KineticaDatasink(
             url=url,
             table_name=table_name,
             username=username,
             password=password,
             mode=mode,
-            schema=self.schema(),
+            schema=pa_schema,
             table_settings=table_settings,
             batch_size=batch_size,
             use_multihead=use_multihead,