jiangxt2 opened a new issue, #12846: URL: https://github.com/apache/gravitino/issues/12846
## Describe the feature Add opt-in Apache Doris batch read and write support to the Gravitino Spark connector for catalogs backed by the existing `jdbc-doris` Server Catalog. When Doris-specific Spark support is disabled, `jdbc-doris` continues to use the existing generic JDBC adapter. When it is enabled on a supported build, the adapter preserves Gravitino authorization and credential vending, validates the logical table against the physical Doris schema, uses the official Doris tablet reader for eligible detail scans, uses Spark JDBC V2 for semantics-sensitive SQL reads, and delegates governed batch writes to the official Doris Stream Load writer. The initial specialized target is Spark 3.5 with Scala 2.12 and the officially published `org.apache.doris:spark-doris-connector-spark-3.5:26.0.0` artifact. Spark 3.5.3 is the minimum supported patch because the current Spark 3.5 implementation uses the write-aware catalog API introduced in that release. The required Spark, Scala, and Doris validation combinations are listed explicitly below; evidence from one combination will not be generalized to untested combinations. The specialized adapter is selected with `spark.sql.gravitino.enableDorisSupport=true`, whose default is `false`. Governed batch append is enabled separately with the Server Catalog property `doris-write-mode=batch`, whose default is `disabled`. Full-table truncate overwrite additionally requires `doris-write-overwrite-mode=truncate`, whose default is `reject`. The `jdbc-doris` Server Catalog adds `doris-fenodes` for comma-separated Doris FE HTTP endpoints and `doris-query-port` for the FE MySQL-protocol query and Catalog metadata port. These properties remain optional for Server-side metadata and DDL through generic JDBC, but both are required when the specialized Spark adapter is enabled. Specialized mode also requires a vended `JdbcCredential`, normally configured with `credential-providers=jdbc-user-password`; if no usable JDBC credential is vended, initialization fails closed instead of accepting a Spark-provided user or password. The same Doris Catalog property contract defines `doris-write-mode=disabled|batch` and `doris-write-overwrite-mode=reject|truncate`. Invalid enum values fail during Server Catalog property transformation or specialized initialization, and `truncate` is invalid unless write mode is `batch`. The catalog-scoped JDBC SQL-lane properties map as follows: `doris-jdbc-partition-column` to Spark JDBC `partitionColumn`, `doris-jdbc-lower-bound` to `lowerBound`, `doris-jdbc-upper-bound` to `upperBound`, `doris-jdbc-num-partitions` to `numPartitions`, and `doris-jdbc-fetch-size` to `fetchsize`. The four partition properties must be present together, and `numPartitions` and `fetchsize` must be positive integers; invalid values fail during specialized catalog initialization. ## Motivation Gravitino already provides the `jdbc-doris` Server Catalog for Doris metadata, while the Spark connector currently treats this provider through the generic JDBC path. Generic JDBC remains a useful fallback, but it does not define a complete Doris-specific contract for tablet reads, JDBC-lossy Doris types, physical-schema drift, Stream Load writes, or the different pushdown semantics of the native and JDBC planners. [Discussion #12473](https://github.com/apache/gravitino/discussions/12473) explicitly scoped the original architecture discussion to governed Doris batch reads. The maintainer response confirmed that Doris support is valuable and invited follow-up issues and implementation work, but it did not explicitly review or approve a write contract. The Discussion established that the independent connector is a reference implementation rather than an implementation to merge as-is, and that authorization ordering, credential handling, schema and type behavior, planner semantics, dependency visibility, and real Doris-backed tests must be explicit. This Feature Issue retains that governed read direction and separately proposes opt-in batch append and full-table truncate overwrite for community review. [Issue #3567](https://github.com/apache/gravitino/issues/3567) and its linked design document previously explored a broader JDBC Catalog design covering MySQL, PostgreSQL, and Doris DDL and DML. This proposal adopts its Doris data-plane direction—use the official Doris Spark Connector and HTTP Stream Load instead of limiting Doris to generic JDBC—but does not adopt its broader Spark DDL, DELETE, generic row-level operation, or UPSERT scope. This Issue does not close or supersede #3567. ## Describe the solution The proposed adapter will: * preserve the existing generic `jdbc-doris` behavior by default and select the specialized adapter only when the provider is exactly `jdbc-doris` and the explicit Spark feature flag is enabled, without changing MySQL, OceanBase, PostgreSQL, or other JDBC providers; * complete `SELECT_TABLE` authorization before loading the Doris physical schema or constructing a read delegate; * map a supported write of the target table to Gravitino `MODIFY_TABLE` and complete that authorization before constructing an official physical write delegate or performing Doris write I/O; in `INSERT INTO target SELECT ... FROM source`, the target requires `MODIFY_TABLE` while each source table continues through the normal `SELECT_TABLE` read path; * require a denied `MODIFY_TABLE` request to stop the Spark adapter before it loads a Doris physical schema or read/write delegate, performs Stream Load, or mutates data; the separately authorized Server existence-probe behavior is tracked by [Issue #12785](https://github.com/apache/gravitino/issues/12785) and [PR #12786](https://github.com/apache/gravitino/pull/12786), but that independent Server change does not block this Doris feature; * require catalog-managed `jdbc-url`, `jdbc-driver`, `doris-fenodes`, and `doris-query-port` together with a vended JDBC user and password; `url`, `driver`, `dbtable`, `query`, user/password, `doris.fenodes`, `doris.query.port`, `doris.table.identifier`, and the governed write policy cannot be replaced by Spark read or catalog options; * reject every query-time/DataFrame per-read option, while allowing the reviewed Server Catalog-scoped JDBC SQL-lane options `doris-jdbc-partition-column`, `doris-jdbc-lower-bound`, `doris-jdbc-upper-bound`, `doris-jdbc-num-partitions`, and `doris-jdbc-fetch-size`; the first four partition options must be configured together, and unknown or protected catalog and `spark.bypass.*` options fail closed without exposing their values; * allow only the reviewed Doris Connector catalog-initialization options `doris.request.retries`, `doris.request.connect.timeout.ms`, `doris.request.read.timeout.ms`, `doris.request.query.timeout.s`, `doris.request.tablet.size`, `doris.batch.size`, `doris.exec.mem.limit`, `doris.filter.query.in.max.count`, and `doris.thrift.max.message.size`; * validate the Gravitino logical schema against one post-authorization physical-schema context that combines the official Doris FE schema with JDBC `information_schema.columns` metadata for column count, order, names, canonical type families, exact type signatures, and nullability, and fail closed if the two physical sources disagree; * use the official Doris tablet reader only for detail scans whose required columns do not need normalization and whose operators remain eligible for the native lane, while routing aggregates, Top-N, global limit, offset, configured partitioned SQL reads, and Doris-specific normalized projections through Spark JDBC V2; * keep normalized predicates as Spark residuals so the documented Spark-visible String and base64 semantics are preserved; * expose only `BATCH_READ` by default; accept Spark `INSERT` only when batch mode is enabled; use the coarse `INSERT + DELETE` privilege set only as a pre-load gate when both batch mode and truncate overwrite are enabled; and let the governed `WriteBuilder`, which implements `SupportsTruncate` but not predicate or dynamic overwrite interfaces, allow only the explicit full-table `truncate()` callback while rejecting `overwrite(filters)`, dynamic overwrite, and partition overwrite; * delegate append to the official Stream Load writer while forcing `doris.sink.mode=stream_load`, `doris.sink.enable-2pc=true`, `doris.sink.properties.strict_mode=true`, `doris.max.filter.ratio=0`, `doris.write.schemaless=false`, and `doris.sink.auto-redirect=false`; the Stream Load `partial_columns` property is not supplied, and catalog, bypass, or Doris-namespaced per-write paths that could place it in connector configuration are rejected, while an absent `doris.write.fields` is derived by Doris Connector 26.0.0 from the exact validated Spark input schema; * validate `LogicalWriteInfo.options()` before constructing the official writer and reject every `doris.*`, `spark.bypass.*`, connection, table-identifier, or governed-policy key; unrelated application metadata may remain present, but Doris Connector 26.0.0 constructs its writer from the catalog configuration and validated schema rather than using those per-write options; * require an exact writable schema and keep normalized Doris type families read-only, except for a validated precision-specific Doris `DATETIME(p)` String contract; and * keep the official Doris Connector and MySQL Connector/J external to the Gravitino Spark runtime and document their complete classpath and network requirements: the Gravitino Server needs MySQL Connector/J and JDBC access to Doris, while the Spark driver and every executor need the Gravitino runtime, Doris Connector, and MySQL Connector/J and must reach the Doris FE/BE endpoints required by their JDBC, tablet-read, metadata, and Stream Load paths. Gravitino authorization does not replace Doris-native privileges. ### Doris-native privilege matrix | Operation | Required Doris privileges | | --- | --- | | Governed read | `SELECT_PRIV` | | Governed append | `SELECT_PRIV` for physical-schema inspection plus `LOAD_PRIV` for Stream Load | | Governed full-table truncate overwrite | `SELECT_PRIV + LOAD_PRIV + DROP_PRIV`; `DROP_PRIV` is required because the official connector executes SQL `TRUNCATE TABLE`, while `ALTER_PRIV` is not required | Full-table truncate overwrite is deliberately a separate opt-in and is not atomic with the subsequent load. If the load fails after truncation, the table may remain empty or partially populated. Connector 2PC coordinates individual writer transactions during normal Spark job success or failure, but it is not a cross-partition single-transaction or job-wide atomic commit guarantee. The specialized adapter will fail closed for streaming writes, non-trivial predicate overwrite, dynamic overwrite, CTAS, UPDATE, DELETE, MERGE, UPSERT, and unsupported write schemas. Spark catalog DDL is also rejected at the Gravitino facade, including `createTable`, `alterTable`, `dropTable`, `purgeTable`, `renameTable`, `createNamespace`, `alterNamespace`, and `dropNamespace`, rather than relying on the official Doris delegate to reject it. Standalone SQL `TRUNCATE TABLE` is rejected because the specialized table does not implement Spark `TruncatableTable`; only the separately enabled full-table `INSERT OVERWRITE` path may invoke the governed `WriteBuilder.truncate()` callback. Doris unique-key table behavior does not imply that the Spark adapter exposes a governed UPSERT or partial-column-update API. With `spark.sql.gravitino.enableDorisSupport=true`, unsupported Spark or Scala lines fail during specialized catalog initialization instead of silently falling back to generic JDBC. Doris 1.2.x and other unlisted Doris releases have no specialized version gate in the initial implementation: they are not certified and do not automatically fall back, so users must leave the flag disabled to select generic JDBC. Enabling specialized mode on an unlisted Doris release is unsupported and has undefined behavior; no correctness, security, or compatibility guarantee is provided. The first contribution will not claim Arrow Flight SQL, strict TLS identity verification for the specialized hybrid path, performance guarantees, or release-level benchmarks. Users who require verified JDBC TLS reads can continue to use the existing generic JDBC configuration. ### Executable schema, type, and planner contract * Direct scalar columns retain their Spark types only when the logical type and the FE/JDBC physical signature agree. * The initial formal Doris-backed direct-scalar claim is limited to the combinations exercised by the required matrix: Doris `INT`, `VARCHAR(64)` exposed through Spark String representation, `DECIMAL(18,3)`, and their tested nullable/non-nullable cases. Other direct scalar mappings may have defensive unit coverage but are not promoted to the formal support claim by this Issue. * The initial formal type claim also includes three approved normalization exceptions that must be exercised by the required governed matrix; these exceptions are intentional logical/physical mappings rather than schema drift. * A logical `ExternalType("largeint")` may match physical Doris `LARGEINT`. Doris 3 metadata is lossy across paths: a standard Server JDBC round-trip may report `TYPE_NAME=UNKNOWN`, while the specialized compatibility path may observe a logical signed `INTEGER`. The `INTEGER` fallback is allowed only when the official Doris FE schema identifies the column as LARGEINT and JDBC `information_schema.columns.COLUMN_TYPE` independently reports `LARGEINT`; logical `INTEGER` alone never authorizes the exception. Spark then exposes the physical LARGEINT value as a lossless String produced by `CAST(... AS STRING)`. * A logical `ExternalType("json")` may match physical Doris `JSON`; Spark exposes the normalized JSON text as a String produced by `CAST(... AS STRING)`. * A logical no-time-zone `Timestamp(p)` may match physical Doris `DATETIME(p)` or `DATETIMEV2(p)` only when `p` is identical and ranges from 0 through 6; Spark exposes `yyyy-MM-dd HH:mm:ss` for `p=0`, or the same value followed by a decimal point and exactly `p` fractional digits for `p>0`. * A logical type, raw physical type, precision, column order, or nullability combination outside the direct mappings and approved normalization exceptions fails closed as schema drift. * TIME, Binary/VARBINARY, Bitmap/HLL, JSONB, VARIANT, IPv4/IPv6, ARRAY, MAP, STRUCT, unsigned integers, DECIMAL256 or other decimals beyond Catalyst precision, and matching external Doris types remain defensive or probe-only normalization paths in the first contribution and are not part of its formal support claim. Their physical signatures and Doris-version prerequisites must be established by dedicated Doris-backed tests before they can be promoted to certified support; in particular, internal Doris VARBINARY storage is not claimed, and DECIMAL precision above 38 requires a Doris environment with `enable_decimal256=true`. * Normalized values use Spark-visible String comparison and ordering semantics. Predicates over normalized columns remain Spark residuals, and Top-N or aggregate pushdown over those columns is rejected. * Required unit tests must assert native/JDBC builder selection and residual handling. Required Doris-backed tests must assert result parity for native filters, aggregates, Top-N, global limit and offset, verify the JDBC executed plan for configured partitioned SQL reads, and validate normalized value formats and boundary behavior. ### Required compatibility and evidence matrix This matrix applies only when `spark.sql.gravitino.enableDorisSupport=true`. Every specialized row uses Doris Connector 26.0.0. When the flag is `false`, `jdbc-doris` follows the existing generic JDBC path and its separate compatibility contract. | Spark | Scala | Doris | Evidence and support statement | | --- | --- | --- | --- | | 3.5.3 | 2.12 | 3.0.6.2 | Required lower-bound embedded governed Catalog, permission, and authorization matrix | | 3.5.3 | 2.12 | 4.0.6 | Outside the final required matrix; earlier targeted smoke evidence is not promoted to certification for the current cumulative implementation | | 3.5.9 | 2.12 | 4.0.6 | Required upper-bound embedded governed Catalog and permission matrix | | 3.5.9 | 2.12 | 3.0.6.2 | Required two-worker standalone deployment and external-classpath matrix | | Other 3.5.3+ patches | 2.12 | 3.0.6.2 or 4.0.6 | Accepted by the Spark compatibility gate but not individually certified | | 3.5.0–3.5.2 | 2.12 | Any | Outside the specialized build and support contract | | 3.5.x | 2.13 | Any | No specialized adapter support; default generic JDBC behavior remains available | | Spark 3.3, 3.4, 3.6+, or 4.x | Corresponding Scala | Any | No specialized adapter support; where Gravitino otherwise supports the Spark line, the default generic JDBC path remains separate and available | | 3.5.3+ | 2.12 | Doris 1.2.x or another unlisted Doris release | The Server Catalog metadata/DDL declaration is separate; specialized mode is not certified and does not automatically fall back, so users must keep the feature flag disabled to use generic JDBC | | Any other unlisted Spark, Scala, and Doris combination | Any | Any | No support or certification statement; compatibility must not be inferred by combining independently listed evidence points | Acceptance criteria: * the default configuration preserves current generic JDBC behavior for `jdbc-doris` and does not affect other JDBC providers; * read authorization completes before the adapter's physical Doris read path, and an explicit `MODIFY_TABLE` denial prevents the Spark adapter from loading a Doris physical schema or delegate, performing Stream Load, or mutating the target table; * if #12786 or an equivalent Server change merges before this feature is reviewed, the stronger end-to-end contract is enabled and a fresh-session denial test must record zero Server-triggered Doris FE HTTP and JDBC requests; otherwise the feature follows the existing MySQL/PostgreSQL connector boundary, under which an independently authorized Server existence probe may occur after `MODIFY_TABLE` denial; * governed endpoints, credentials, and Stream Load safety options cannot be overridden; * logical and physical schema drift fails closed, and documented Doris-specific type representations preserve their tested Spark semantics; * append, multi-partition writes, separately enabled truncate overwrite, permission failures, and documented non-atomic failure behavior are covered by real Doris tests; * empty append leaves the existing table content unchanged and does not issue a Stream Load request; * empty full-table truncate overwrite executes the authorized truncate operation and leaves the target table empty even when no data batch is produced; * every supported combination and evidence level matches the compatibility matrix above, without inferring support for untested Spark, Scala, or Doris combinations; * unsupported operations fail closed without mutating the Doris table or exposing sensitive configuration; and * the Server Doris Catalog documentation, generic Spark JDBC documentation, and specialized Doris Spark connector documentation cover the new Server Catalog properties and mappings, configuration defaults, Doris privileges, external runtime dependencies, planner lanes, the certified and probe-only type boundaries, version certification, 2PC scope, truncate non-atomicity, TLS boundaries, and unsupported operations. ## Additional context Related community and design history: * [Discussion #12473: Apache Doris support for the Gravitino Spark connector](https://github.com/apache/gravitino/discussions/12473) * [Issue #3567: Support Doris catalog in spark-connector](https://github.com/apache/gravitino/issues/3567) * [Issue #3567 design document](https://docs.google.com/document/d/1XWzvqV38YWh4ajudcxlFk9IkF414n5veE81_6chCqEQ/edit#heading=h.mhiqko16neu0) * [Independent reference connector](https://github.com/jiangxt2/gravitino-doris-spark-connector) [PR #12414](https://github.com/apache/gravitino/pull/12414) is an open Spark connector architecture change, not a dependency of this feature. If it merges before this contribution is reviewed, the Doris adapter would be adapted to the merged version-owned optional-binding architecture while preserving the default generic JDBC fallback and the governed contract described above; until then, this Issue does not treat contributor code from that PR as an upstream baseline. [Issue #12785](https://github.com/apache/gravitino/issues/12785) and [PR #12786](https://github.com/apache/gravitino/pull/12786) track a stronger Server-side isolation boundary that skips the external existence probe after an explicit `MODIFY_TABLE` denial. This Doris feature does not depend on that disputed generic Server behavior and must not be blocked if #12786 is not merged. Without it, the baseline contract matches the existing MySQL/PostgreSQL Spark JDBC path: no Spark-side physical delegate or data write occurs after denial, while the Server may still perform a separately authorized existence probe. If it merges, this feature will adopt and test the stronger end-to-end zero-I/O contract. This governance contract applies only to reads and writes performed through the Gravitino-managed catalog. It does not prevent a user who independently has Doris credentials and permission to control Spark configuration from bypassing Gravitino and calling `format("doris")` or `format("jdbc")` directly. Deployments must therefore restrict direct credential distribution and Spark configuration privileges when all Doris access is expected to pass through Gravitino governance. The independent project is evidence for the value and feasibility of the contract, not the source of Apache Gravitino support claims. All acceptance criteria above require implementation and verification in the Apache Gravitino repository. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
