[DSv2] Create UC client

[TimothyW553]

🥞 Stacked PR

Use this link to review incremental changes.

• catalogtableutils-ccv2 [Files changed]
  • stack/ccv2-catalog-config [Files changed]
    • stack/ccv2-catalog-wireup [Files changed]

---

Which Delta project/connector is this regarding?

• Spark
• Standalone
• Flink
• Kernel
• Other (fill in here)

Description

Adds Unity Catalog (UC) integration for catalog-managed commits (CCv2) to DSv2 connector with Kernel.

Key changes:

• DeltaSnapshotManagerFactory: Factory that automatically selects CatalogManagedSnapshotManager for UC tables or PathBasedSnapshotManager for regular tables
• UnityCatalogManagedCommitClient: UC-specific implementation of ManagedCommitClient that wraps UCCatalogManagedClient
• CatalogManagedSnapshotManager: Generic snapshot manager that delegates to catalog-managed commit clients for loading snapshots/versions
• CatalogTableUtils: Utilities for detecting UC-managed tables and extracting table IDs from catalog metadata

Architecture:

SparkTable
    └─ DeltaSnapshotManagerFactory
         └─ CatalogManagedSnapshotManager
              └─ UnityCatalogManagedCommitClient
                   └─ UCCatalogManagedClient (kernel/unitycatalog)

This enables the DSv2 connector to read UC catalog-managed tables via Kernel, with automatic fallback to path-based access for regular tables.

How was this patch tested?

• Unit tests for factory, snapshot manager, UC client, and utils
• Integration test (DeltaCatalogSuite) validates end-to-end UC table reads via DSv2
• Existing Delta tests continue to pass with fallback to path-based managers

Does this PR introduce any user-facing changes?

No.

[huan233usc]

Maybe not called it Client？This abstraction is 1 level above to encapsulate specific Catalog implementation, maybe call it ManagedCatalogAdapter?

[TimothyW553]

Ah, yes. Renamed to ManagedCatalogAdapter to better reflect that this is an adapter abstraction above catalog-specific implementations. Also renamed the UC implementation to UnityCatalogAdapter...

[huan233usc]

Try not introduce Spark dependency to this class -- maybe add an pojo

public class UnityCatalogConnectionInfo {
    private final String tableId;
    private final String endpoint;
    private final String token;
    private final String catalogName;

    public UnityCatalogConnectionInfo(String tableId, String endpoint, String token, String catalogName) {
        this.tableId = tableId;
        this.endpoint = endpoint;
        this.token = token;
        this.catalogName = catalogName;
    }

and

public class SparkUnityCatalogUtils {
    
    public static UnityCatalogConnectionInfo extractConnectionInfo(CatalogTable table, SparkSession spark) {}

[TimothyW553]

Ah yep that's cleaner and makes it more extensible. Refactored to add:

1. UnityCatalogConnectionInfo - Pure data POJO (tableId, tablePath, endpoint, token)
2. SparkUnityCatalogUtils - Extracts UC config from Spark metadata
3. UnityCatalogAdapter - Two factory methods:
   • fromCatalog(catalogTable, spark) - convenient wrapper
   • fromConnectionInfo(info) - no Spark dependency

---

Just for personal and future reference.

The original design tightly coupled UnityCatalogAdapter to Spark - method signature was like f(CatalogTable, Spark). Now the adapter depends only on the POJO, which allows:

• Creating adapters from config files, env vars, REST APIs (not just Spark)
• Testing UC logic without mocking Spark
• Serializing/passing connection info across boundaries

Example with pojo

// From config file
Properties props = loadConfig("uc.properties");
UnityCatalogConnectionInfo info = new UnityCatalogConnectionInfo(...);
UnityCatalogAdapter.fromConnectionInfo(info);

[huan233usc]

loadLatestSnapshot() will call commitClient.getRatifiedCommits as well. We need to avoid calling it twice.

[TimothyW553]

Ah yes, I see the codepath now for getting commits and the latest snapshot. Fixed by extracting catalog commits directly from the snapshot's log segment instead of calling UC twice:

List<ParsedCatalogCommitData> catalogCommits = latestSnapshot.getLogSegment().getAllCatalogCommits();

[huan233usc]

Plz add comments if the param passed in is not obvious. (Optional.empty()/param_name/)

[TimothyW553]

Updated. I did it this way

catalogAdapter.loadSnapshot(kernelEngine, /* versionOpt = */ Optional.empty(), /* timestampOpt = */ Optional.empty())

(a previous PR was done like this)

[huan233usc]

not necessary for logging/telemetry I guess

[huan233usc]

Why not having two method: fromPath(String tablePath) fromCatalogTable(CatalogTable catalogTable) to simplify the code?

[TimothyW553]

Sounds good, I've split the factory into two methods:

• fromPath(tablePath, hadoopConf) - for filesystem-based tables
• fromCatalogTable(catalogTable, spark, hadoopConf) - for catalog tables with automatic UC detection

Makes the API cleaner and avoids requiring SparkSession for simple path-based access.

[huan233usc]

maybe just import vs using fully names

[huan233usc]

Can you try using the sample catalog managed table used by kernelUnityCatalog?