[#12846] feat(spark-connector): Add governed Apache Doris batch read and write support

[jiangxt2]

What changes were proposed in this pull request?

This pull request adds an opt-in governed Apache Doris batch read and write adapter for jdbc-doris catalogs in the Spark 3.5 connector on Scala 2.12.

The adapter preserves Gravitino authorization and credential vending, validates the Gravitino logical schema against Doris FE and JDBC physical metadata, and selects either the official Doris tablet reader or Spark JDBC V2 according to the query semantics. Predicates over normalized Doris columns remain Spark residuals with Spark-visible String semantics, and unsupported schema or planner combinations fail closed.

Governed writes support batch append and a separately enabled full-table truncate overwrite. They require MODIFY_TABLE authorization before physical Doris access, validate the exact writable schema, delegate to the official Doris Stream Load writer, force 2PC and strict safety options, and reject streaming, predicate overwrite, dynamic overwrite, row-level operations, and Spark catalog DDL.

The change also adds the Doris Server Catalog property contract, strict vended-credential handling for specialized mode, a Spark version gate, a documented Doris release boundary, compile-only external connector dependencies, documentation, unit tests, and real Doris integration tests.

The Spark 3.5 module owns the Doris Connector compile/test dependency and positive classpath coverage, while spark-common has no Doris version-specific dependency. Physical-schema validation preserves internal FE/JDBC drift details, while external JDBC failures expose only sanitized SQLState and vendor-code diagnostics without retaining raw, potentially credential-bearing causes.

Why are the changes needed?

The existing generic JDBC fallback does not provide a Doris-specific contract for tablet reads, JDBC-lossy Doris types, physical-schema drift, mixed native/JDBC pushdown semantics, or governed Stream Load writes. This change provides that contract without changing the default jdbc-doris behavior or the behavior of MySQL, PostgreSQL, and other JDBC providers.

Fix: #12846

Does this PR introduce any user-facing change?

Yes.

• spark.sql.gravitino.enableDorisSupport=true opts into the specialized adapter; the default remains false, which preserves generic JDBC behavior.
• Specialized mode supports Spark 3.5.3 or later in the 3.5 line with Scala 2.12. The certified matrix is documented for Spark 3.5.3/Doris 3.0.6.2 and Spark 3.5.9/Doris 4.0.6, plus Spark 3.5.9/Doris 3.0.6.2 standalone deployment coverage. Other Doris releases are not certified and do not automatically fall back; users must leave the flag disabled to select generic JDBC for them.
• Specialized mode requires catalog-managed jdbc-url, jdbc-driver, doris-fenodes, and doris-query-port, plus a vended JdbcCredential; configured jdbc-user and jdbc-password are not accepted as specialized Spark connection credentials.
• doris-write-mode=batch enables governed append, and doris-write-overwrite-mode=truncate separately enables non-atomic full-table truncate overwrite. Both remain disabled by default.
• The Server Catalog registers the optional SQL-lane properties doris-jdbc-partition-column, doris-jdbc-lower-bound, doris-jdbc-upper-bound, doris-jdbc-num-partitions, and doris-jdbc-fetch-size and validates the positive-integer fields. Specialized Spark catalog initialization additionally requires all four partition options to be present together and revalidates positive-integer values.
• The official Doris Spark Connector 26.0.0 and MySQL Connector/J remain external runtime dependencies and must be available to the Spark driver and executors.

How was this patch tested?

Current-head checks:

• ./gradlew :spark-connector:spark-common:test --tests org.apache.gravitino.spark.connector.plugin.TestGravitinoDriverPlugin -PskipITs — passed 11 tests without a Doris Connector dependency in spark-common.
• ./gradlew :spark-connector:spark-3.5:test --tests org.apache.gravitino.spark.connector.plugin.TestDorisSparkVersionGate35 --tests org.apache.gravitino.spark.connector.jdbc.doris.TestDorisTableCatalog35 --tests org.apache.gravitino.spark.connector.plugin.TestGravitinoDriverPlugin -PskipITs — passed 4 Spark 3.5-specific tests; common plugin coverage is reported separately above.
• ./gradlew :spark-connector:spark-common:compileTestJava :spark-connector:spark-3.5:compileJava -PscalaVersion=2.13 -PsparkVersionOverride=3.5.9 -PskipITs — passed, validating the common dependency boundary and Doris source exclusion.
• ./gradlew spotlessCheck rat — passed.
• git diff --check origin/main...HEAD — passed.

Additional post-rebase checks performed before the latest review-hardening commit:

• ./gradlew :catalogs:catalog-jdbc-doris:test --tests org.apache.gravitino.catalog.doris.TestDorisCatalogPropertiesMetadata -PskipITs — passed; that Server-side code was unchanged by the latest commit.
• The broader Spark 3.5 Doris unit suite covering capability, schema, scan, write, authorization, version, and catalog behavior passed; every path touched by the latest commit was rerun by the current-head checks above.

Real Doris evidence collected before the final rebase and the latest review-hardening commit:

• Spark 3.5.3 + Doris 3.0.6.2 embedded: SparkJdbcDorisCatalogIT35 23 tests and SparkJdbcDorisLoadPrivilegeIT35 3 tests passed, with two environment-mode skips and no failures or errors.
• Spark 3.5.9 + Doris 4.0.6 embedded: SparkJdbcDorisCatalogIT35 23 tests and SparkJdbcDorisLoadPrivilegeIT35 3 tests passed, with two environment-mode skips and no failures or errors.
• Spark 3.5.3 + Doris 3.0.6.2 and Spark 3.5.9 + Doris 4.0.6: SparkJdbcDorisAuthorizationIT35 passed once on each combination with no failures, errors, or skips.
• Spark 3.5.9 + Doris 3.0.6.2 standalone: SparkJdbcDorisCatalogIT35 passed 23 tests against two active one-core workers with the external Gravitino, Doris, and MySQL JARs, with one deploy-only skip and no failures or errors.

The final rebase integrated upstream changes in shared Spark test configuration and documentation without changing the Doris production behavior. The latest review-hardening commit only removes a version-specific test dependency from spark-common, consolidates equivalent registration branches, and improves sanitized error diagnostics; it does not change the Doris read/write data plane, schema policy, authorization ordering, or forced connector options. The current-head checks above were rerun after that commit; the real Doris matrix was not rerun after the final rebase or the review-hardening commit.

[jiangxt2]

Supplemental functional Docker evidence

I ran additional functional characterization against the current PR head (d2b7c692248333dd3aed6786116dcf4bd4c6192e) using Spark 3.5.9 / Scala 2.12 and the repository-managed Doris 4.0.6 Docker fixture. These checks intentionally compare behavior rather than timing or throughput.

Area	Existing generic JDBC path	Governed Doris adapter	Verified conclusion
Default selection	jdbc-doris uses the generic JDBC catalog when specialized support is disabled	Selected only when spark.sql.gravitino.enableDorisSupport=true	The existing default and other JDBC providers remain unchanged
Eligible detail-read plan	JDBCScan	Official DorisScanV2 tablet reader	The hybrid router selects the intended physical reader
Detail-read parallelism	One JDBC partition without generic JDBC partition properties	Eight Doris tablet partitions with doris.request.tablet.size=1 in the focused check; the earlier standalone run used both executors	Native tablet parallelism is functional rather than only declared
Catalog-controlled SQL parallelism	The Doris SQL-lane properties are not mapped by the generic converter	Four configured JDBC ranges, with one test row observed in every range	The specialized SQL lane uses all configured partitions when bounds match the data domain
Aggregates, Top-N, limit, and offset	Executed through Spark JDBC	Routed through the governed JDBC semantics lane	Results matched; this lane is a correctness and controlled-parallelism feature, not a universal acceleration claim
Doris LARGEINT and JSON	Their Gravitino ExternalType representation causes the generic Spark type converter to throw UnsupportedOperationException	Accepted through explicit SQL projection and exposed as Spark String	This is String normalization, not native non-String type support
Doris DATETIME(p)	Generic conversion does not establish the specialized precision-specific contract	Exposed through the documented precision-specific String representation and validated on writes	This is a Doris semantic contract, not recovery from a generic UnsupportedOperationException
Standard non-String types	Converter-level characterization succeeds for Boolean, integral, floating, Decimal, Date, Binary, no-time-zone Timestamp, and List	The same converter-level Spark types are produced	There is no newly supported non-String UnsupportedOperationException case
Small multi-partition append	Four-partition Spark JDBC append completed with the expected Doris count and checksum	Four-partition append through the official Doris writer completed with the same count and checksum	Both paths can write in parallel; the specialized value is the Doris-native writer and governance contract
Governed write protocol	Spark JDBC writer semantics	Stream Load mode with forced 2PC, strict mode, zero filter tolerance, schemaless mode off, and automatic redirect off	The adapter adds a protected Doris-specific write contract rather than merely exposing more write operations
Authorization and credentials	Existing generic JDBC behavior	Vended JDBC credentials are required, and read/write authorization completes before specialized physical access	The stricter behavior is intentionally scoped to the opt-in adapter
Capability surface	Generic JDBC capabilities	BATCH_READ by default; batch append and optional truncate are explicitly enabled, while streaming, row-level operations, predicate/dynamic overwrite, and Spark catalog DDL remain rejected	The adapter follows a minimum-capability model

The exact answer to the UnsupportedOperationException question is: no non-String case was found. Generic JDBC throws for Gravitino ExternalType; every such case accepted by the specialized converter is represented as Spark String. Standard non-String types already convert successfully through generic JDBC, so the PR does not claim them as new exception compatibility.