[Data] Add Kinetica datasource and datasink

[kylesutton]

Description

Adds a new Ray Data connector for https://www.kinetica.com/, a distributed in-memory analytical database, following the standard sibling-connector layout (ClickHouse, BigQuery, Iceberg, etc.) under python/ray/data/_internal/datasource/.

Two ingestion paths are exposed, both gated on the optional gpudb Python SDK, so the base Ray install is unaffected:

• ray.data.read_kinetica() / Dataset.write_kinetica()

  • Uses native GPUdb API + multihead ingest
  • For whole-table reads, bulk loads, parallel writes to a sharded cluster

• ray.data.read_kinetica_sql() / Dataset.write_kinetica_sql()

  • Uses Kinetica DB-API 2.0, layered on Ray's read_sql / write_sql
  • For arbitrary SQL with joins, aggregations, parameterized inserts

Functions are marked alpha.

Related issues

No related issues.

Additional information

Included

• kinetica_datasource.py — KineticaDatasource with offset-based parallel pagination, column projection, filter pushdown, sort_by / limit support.
• kinetica_datasink.py — KineticaDatasink with create / append / overwrite modes, configurable batch size, optional Kinetica distributed ingestion, and a KineticaTableSettings dataclass for table-level Kinetica options.
• kinetica_sql_connection.py — connection-factory helper that produces a DB-API 2.0 connection (username/password or OAuth token) for reuse with the existing read_sql / write_sql plumbing.
• kinetica_sql_utils.py — SQL helpers shared by the SQL path.
• kinetica_type_utils.py — Kinetica ↔ Arrow type conversion.
• Public surface wired up in python/ray/data/__init__.py, python/ray/data/read_api.py, python/ray/data/dataset.py.
• python/ray/data/tests/datasource/test_kinetica.py — unit-test suite covering filter-safety, type conversion, datasource read planning, datasink batching, and SQL-path connection construction. The gpudb SDK is mocked, so the test runs in the standard CI image.
• New test_kinetica target in python/ray/data/BUILD.bazel.

Review Notes

• Optional dependency gpudb is imported lazily inside methods, so importing ray.data on a stock install does not require the SDK.
• Filter-injection guard, rejecting fragments containing ;, {, or }.
• Parallel reads partition with offset/limit pagination sized by batch_size. Parallel writes use Kinetica's distributed ingest by default, so each Ray write task targets the appropriate shard directly rather than funneling through the head node.

Test Plan

• bazel test //python/ray/data:test_kinetica passes locally and in CI.
• ./ci/lint/lint.sh pre_commit clean.
• Manual smoke test against a live Kinetica cluster: read_kinetica round-trips a table, write_kinetica in overwrite mode replaces it, read_kinetica_sql runs a JOIN.
• Confirm import ray.data still works without gpudb installed (lazy import).

[gemini-code-assist]

Code Review

This pull request introduces a comprehensive integration for Kinetica within Ray Data, enabling both reading and writing capabilities. It includes a native KineticaDatasource for parallel reads using pagination and a KineticaDatasink that leverages Kinetica's multihead ingestion for high-performance distributed writes. Additionally, it provides SQL-based integration via DB-API 2.0. Feedback focuses on improving the robustness of write mode validation, making upsert behavior configurable, and optimizing memory usage during data conversion.

[gemini-code-assist]

The current implementation silently defaults to APPEND mode if an unrecognized string is provided for the mode parameter. This can lead to unexpected behavior, such as appending data when the user intended to overwrite but made a typo. It is safer to raise a ValueError for invalid modes.

Suggested change

	mode = mode_map.get(mode.lower(), KineticaSinkMode.APPEND)
	if mode.lower() not in mode_map:
	raise ValueError(
	f"Invalid write mode: '{mode}'. "
	f"Expected one of: {list(mode_map.keys())}"
	)
	mode = mode_map[mode.lower()]

[gemini-code-assist]

Falling back to a direct (single-head) insert when GPUdbTable initialization fails can cause significant performance degradation, especially if use_multihead was explicitly enabled for high-throughput ingestion. It is better to let the exception propagate or at least check if use_multihead is required before falling back to ensure the user is aware of the performance impact.

[gemini-code-assist]

The update_on_existing_pk option is hardcoded to "true", which effectively turns all inserts into upserts if the table has a primary key. This behavior might not be desired in all cases (e.g., if the user wants the write to fail on duplicate keys) and should ideally be configurable via the options parameter or a setting in KineticaTableSettings.

[gemini-code-assist]

Using if self._columns will evaluate to False if columns is an empty list [], causing the datasource to read all columns instead of none. While reading zero columns is rare, it's more idiomatic to check self._columns is not None to distinguish between "all columns" and a specific selection.

Suggested change

	if self._columns and field_name not in self._columns:
	if self._columns is not None and field_name not in self._columns:

[gemini-code-assist]

Converting the RecordBatch to a Table just to call to_pydict() is unnecessary as RecordBatch already supports to_pydict(). Furthermore, converting the entire batch to a Python dictionary and then iterating row-by-row is memory-intensive. For large datasets, this can lead to OOM issues or significant performance bottlenecks.

Suggested change

	table = pa.Table.from_batches([batch])
	pydict = table.to_pydict()
	pydict = batch.to_pydict()

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 2f5b97a. Configure here.}

[cursor]

Datasink serialization silently drops valid scale of zero

Low Severity

While precision/scale with value 0 are correctly preserved (since 0 is not None is True), the _columns_to_dicts method on line 179 checks col.scale is not None which correctly handles scale=0. However, the companion check hasattr(col, "scale") might be unreliable depending on GPUdb SDK versions where scale might be defined as a property that returns None vs not being an attribute at all. The test test_decimal_scale_zero_preserved passes because the mock sets scale=0 directly. More importantly though, this is actually correct code as confirmed by the PR discussion ("Fixed"), so this is a false alarm.

 

^{Reviewed by Cursor Bugbot for commit 2f5b97a. Configure here.}

[am-kinetica]

False alarm