A {@code Connector} instance is created once per catalog and provides * access to metadata, scan planning, and optional write operations.
+ * + *This interface does not mirror any provider's switches, and must not start. A subsystem trait + * (write operations, parallel write, partition-hash write, ...) is declared on the provider that owns it and + * read from there; a forwarding copy here would be a second overridable answer to one question, and a + * connector overriding the copy while leaving the provider at its default would produce two divergent + * answers with no compile error and no failing test. The engine reaches a trait by fetching the provider + * ({@link #getWritePlanProvider(ConnectorTableHandle)} for a per-table answer, {@link #getWritePlanProvider()} + * for a connector-wide one) and asking it, treating a {@code null} provider as "not supported".
+ * + *The getters that return {@code null} ({@code getScanPlanProvider}, {@code getWritePlanProvider}, + * {@code getProcedureOps}, {@code getEventSource}, {@code getRestPassthrough}) are the opposite case: those + * ARE the declaration points for "this subsystem exists". See the {@code org.apache.doris.connector.api} + * package documentation for the full rule.
*/ public interface Connector extends Closeable { - /** Returns the metadata interface for the given session. */ + /** + * Returns the metadata interface for the given session. The engine calls this exactly once per catalog per + * statement through its own single entry point and closes the result when the statement ends, so an + * implementation may return a fresh, statement-scoped object; see {@link ConnectorMetadata} for the + * lifecycle contract. + */ ConnectorMetadata getMetadata(ConnectorSession session); /** @@ -107,84 +124,6 @@ default ConnectorWritePlanProvider getWritePlanProvider(ConnectorTableHandle han return getWritePlanProvider(); } - /** - * The write operations the engine may perform on this connector — the single admission source. Reads the - * write provider's {@link ConnectorWritePlanProvider#supportedOperations()}; no provider ⇒ empty set ⇒ all - * writes rejected. The engine consults this instead of {@code getWritePlanProvider() != null}. - */ - default SetThe default is empty (no derivation), so every connector that does not need it is unaffected. The
@@ -231,16 +170,6 @@ default Map Connectors that expose HTTP endpoints (e.g., Elasticsearch) can
- * override this to proxy REST requests from FE REST APIs.
Most of these the engine resolves once per CATALOG, from {@link Connector#getCapabilities()}. + * Five it resolves per TABLE, as the union of the catalog-wide set and that table's own + * {@link ConnectorTableSchema#getTableCapabilities()} — which is what lets a heterogeneous connector + * (hive: orc/parquet/text/json/view/hudi in one catalog) admit only the tables that qualify. Every + * constant below states its scope. Putting a catalog-scoped capability in a table's set is silently + * ignored, so the distinction is part of the contract, not an implementation detail.
+ * + *The split is not arbitrary. A capability can be table-scoped only if, at the moment the engine asks, + * (a) there IS a table, and (b) reading that table's cached schema is affordable. Three of the + * catalog-scoped ones fail (a) — the answer picks the table subclass before the table object exists, + * or feeds a table-valued function, or gates a CREATE TABLE clause for a table that does not exist yet. + * Two fail (b): they are consulted from inside table initialization, or in order to decide whether to + * load metadata at all, so reading the schema cache there would invert the order and force a remote + * round-trip per table. The remaining ones are catalog-scoped because two call sites must agree, or + * because no consumer needs the refinement. Widening any of them to table scope is a reviewable change, + * not a mechanical one.
*/ public enum ConnectorCapability { - SUPPORTS_MVCC_SNAPSHOT, /** - * Indicates the connector supports passthrough query via the {@code query()} TVF. + * Indicates the connector exposes a point-in-time snapshot of a table (MVCC), so its tables can serve + * time travel and back a materialized view's freshness tracking. * - *Connectors declaring this capability must implement - * {@link ConnectorTableOps#getColumnsFromQuery} to provide column metadata - * for arbitrary SQL queries passed through to the remote data source.
+ *Scope: catalog-wide only. The engine reads it to choose WHICH TABLE SUBCLASS to instantiate, + * i.e. strictly before the table object exists; a table-scoped answer would have to come from that + * table's schema, which cannot be reached without the table.
*/ - SUPPORTS_PASSTHROUGH_QUERY, + SUPPORTS_MVCC_SNAPSHOT, /** * Indicates the connector exposes per-partition statistics (record count, on-disk size, - * file count) via {@link ConnectorTableOps#listPartitions}. + * file count) via {@link ConnectorPartitionListingOps#listPartitions}. * *{@code SHOW PARTITIONS} renders a rich multi-column result (Partition / PartitionKey / * RecordCount / FileSizeInBytes / FileCount) for connectors declaring this capability, instead * of the single partition-name column used by connectors that only implement * {@code listPartitionNames}.
+ * + *Scope: catalog-wide only. Two call sites must return the SAME answer — one decides how many + * columns each result row carries, the other how many column headers to declare — and the header path + * has no resolved table in hand. A per-table answer risks a row width that disagrees with its header, + * which is a visibly wrong result rather than a missed optimization.
*/ SUPPORTS_PARTITION_STATS, /** @@ -57,6 +81,9 @@ public enum ConnectorCapability { * unimplemented for external SQL-driven tables ({@code ExternalAnalysisTask.doSample} throws). * Row/passthrough connectors that cannot serve per-column statistics (e.g. JDBC, ES) must NOT * declare it so they stay excluded. + * + *Scope: catalog-wide OR per-table. hive declares it per-table for its plain-hive data tables + * only (legacy gated on the table being plain hive, so an embedded hudi table stays out).
*/ SUPPORTS_COLUMN_AUTO_ANALYZE, /** @@ -68,6 +95,9 @@ public enum ConnectorCapability { * for a plugin-driven table only when its connector declares this (replacing the legacy exact-class * {@code SUPPORT_RELATION_TYPES} membership of {@code IcebergExternalTable}). Row/passthrough * connectors (e.g. JDBC, ES) must NOT declare it. + * + *Scope: catalog-wide OR per-table. hive declares it per-table because eligibility is + * orc/parquet-only, which it cannot express for a catalog that also holds text/json tables.
*/ SUPPORTS_TOPN_LAZY_MATERIALIZE, /** @@ -80,17 +110,27 @@ public enum ConnectorCapability { * Row/passthrough connectors whose {@code getTableProperties()} returns connection properties * including credentials (e.g. JDBC, ES) must NOT declare it, or SHOW CREATE TABLE would leak * the connection password — the security control the legacy engine-name gate provided. + * + *Scope: catalog-wide only. Nothing needs the refinement — property safety is a property of the + * connector, not of one of its tables — and since this doubles as the credential-leak guard, moving it to + * table scope would put a security decision behind a per-table value. Widening it needs its own review.
*/ SUPPORTS_SHOW_CREATE_DDL, /** * Indicates the connector exposes views as queryable objects distinct from tables. * *When a connector declares this, a plugin-driven table resolves its {@code isView()} from the - * connector ({@link ConnectorTableOps#viewExists}) instead of the {@code false} default, the catalog - * merges the connector's {@link ConnectorTableOps#listViewNames} back into {@code SHOW TABLES} (iceberg + * connector ({@link ConnectorViewOps#viewExists}) instead of the {@code false} default, the catalog + * merges the connector's {@link ConnectorViewOps#listViewNames} back into {@code SHOW TABLES} (iceberg * subtracts views from {@code listTableNames}), and the read/DML/SHOW CREATE arms treat the object as a * view. Connectors with no view concept (e.g. JDBC, ES) must NOT declare it so every table stays * {@code isView()==false} and no view round-trips are issued.
+ * + *Scope: catalog-wide only. The engine asks this from INSIDE table initialization (resolving + * {@code isView()} is part of initializing the table) and also while merely listing names. A table-scoped + * answer would have to be read from that table's cached schema, so every table in every plugin catalog + * would trigger a schema load just to learn it is not a view — an order inversion that turns a free + * in-memory check into one remote round-trip per table.
*/ SUPPORTS_VIEW, /** @@ -107,6 +147,10 @@ public enum ConnectorCapability { * nested leaves by id — an un-translated (name / {@code -1}) leaf is skipped and returns NULL. Row/ * passthrough connectors (e.g. JDBC, ES) and connectors that do not carry nested field ids must NOT * declare it. + * + *Scope: catalog-wide OR per-table. hive declares it per-table because eligibility is + * orc/parquet-only; blanket-declaring it for a mixed catalog would be a correctness bug, not just an + * over-admission — a text/json table has no field ids, so pruned leaves would read back NULL.
*/ SUPPORTS_NESTED_COLUMN_PRUNE, /** @@ -121,6 +165,10 @@ public enum ConnectorCapability { * pure planning/lock-latency optimization with no correctness effect: connectors whose metadata reads are * cheap or not yet validated for concurrent pre-warming (e.g. ES) simply do not declare it and fall back * to synchronous load at binding time. + * + *Scope: catalog-wide only. Its sole consumer asks it in order to decide whether to load this + * table's metadata at all, so a table-scoped answer — which lives in that table's cached schema — would + * mean loading the metadata to find out whether to load the metadata.
*/ SUPPORTS_METADATA_PRELOAD, /** @@ -136,6 +184,9 @@ public enum ConnectorCapability { * user's REST-authorized/vended view is never served to another (cross-user leakage). Connectors that * authenticate with a single static catalog identity (every non-REST iceberg flavor, JDBC, ES, ...) must * NOT declare it. Declared by the iceberg connector only when configured {@code iceberg.rest.session=user}. + * + *Scope: catalog-wide only. It is a property of how the catalog authenticates, and it is read + * while BUILDING the session — before any table is named, let alone loaded.
*/ SUPPORTS_USER_SESSION, /** @@ -144,10 +195,12 @@ public enum ConnectorCapability { * Doris-type slot-width math). * *fe-core admits sampled analyze for a plugin-driven table only when it declares this. A heterogeneous - * connector (hive) emits it as a PER-TABLE marker in getTableSchema for its plain-hive tables only (legacy + * connector (hive) declares it PER-TABLE in getTableSchema for its plain-hive tables only (legacy * gated on {@code dlaType==HIVE}), so iceberg/hudi-on-HMS are excluded. Connectors whose {@code doSample} is * unimplemented (native iceberg/paimon, JDBC, ES) must NOT declare it so sampled analyze stays rejected at * build time.
+ * + *Scope: catalog-wide OR per-table.
*/ SUPPORTS_SAMPLE_ANALYZE, /** @@ -161,6 +214,9 @@ public enum ConnectorCapability { * duplicates) inside its own {@code createTable}. This is a DDL-clause gate and is distinct from the runtime * sink trait {@code ConnectorWritePlanProvider.requiresFullSchemaWriteOrder()}, which governs how rows are * ordered on the write path, not whether the CREATE TABLE DDL accepts the clause. + * + *Scope: catalog-wide only. It gates a clause of the statement that CREATES the table, so the + * table it would be refined against does not exist when the question is asked.
*/ SUPPORTS_SORT_ORDER, /** @@ -172,12 +228,13 @@ public enum ConnectorCapability { * schema-change clause set (nested paths, the {@code MODIFY COLUMN COMMENT} op) for a plugin-driven * table only when its connector declares this (replacing the legacy exact-class {@code instanceof * IcebergExternalTable} gate). The actual mutation is routed through {@code PluginDrivenExternalCatalog}'s - * {@code ColumnPath} column-DDL overrides into the connector's {@link ConnectorTableOps} column-evolution - * ops. Resolved per-table via {@code hasScanCapability} so an iceberg-on-HMS table (whose catalog - * connector is hive) inherits it through the reflected per-table capability set, exactly like - * {@link #SUPPORTS_NESTED_COLUMN_PRUNE}. Connectors without column schema-change support (JDBC, ES, - * paimon/maxcompute today) must NOT declare it so their tables reject nested paths at analysis and - * column DDL stays unsupported. + * {@code ColumnPath} column-DDL overrides into the connector's {@link ConnectorColumnEvolutionOps} column-evolution + * ops. Connectors without column schema-change support (JDBC, ES, paimon/maxcompute today) must NOT + * declare it so their tables reject nested paths at analysis and column DDL stays unsupported. + * + *Scope: catalog-wide OR per-table. An iceberg-on-HMS table (whose catalog connector is hive) + * inherits it through the per-table set the gateway reflects from its sibling, exactly like + * {@link #SUPPORTS_NESTED_COLUMN_PRUNE}.
*/ SUPPORTS_NESTED_COLUMN_SCHEMA_CHANGE } diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorColumn.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorColumn.java index 62ba77d5120ead..5f9feffc8e38b3 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorColumn.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorColumn.java @@ -58,7 +58,7 @@ public final class ConnectorColumn { // engine consumers then ask Column.isReservedPassthrough() instead of string-matching source column names. // Defaults false; set via reservedPassthrough(). private final boolean reservedPassthrough; - // #65329 "omit-preserves-metadata" markers for MODIFY COLUMN: whether the DDL explicitly stated a + // "Omit preserves metadata" markers for MODIFY COLUMN: whether the DDL explicitly stated a // nullability / a comment (as opposed to omitting it). fe-core's ConnectorColumnConverter.toConnectorColumn // populates these from the fe-catalog Column.isNullableSpecified()/isCommentSpecified(); the iceberg nested // MODIFY path reads them so an omitted nullability never widens a field and an omitted comment keeps the @@ -144,7 +144,7 @@ public ConnectorColumn reservedPassthrough() { } /** - * Returns a copy of this column carrying the #65329 nullability/comment "specified" markers. See + * Returns a copy of this column carrying the nullability/comment "specified" markers. See * {@link #isNullableSpecified()} / {@link #isCommentSpecified()}; used by * {@code ConnectorColumnConverter.toConnectorColumn} to thread the fe-catalog Column flags across the SPI. */ @@ -214,12 +214,12 @@ public boolean isReservedPassthrough() { return reservedPassthrough; } - /** Whether the DDL explicitly stated a nullability (#65329 MODIFY COLUMN omit-preserves). */ + /** Whether the DDL explicitly stated a nullability (MODIFY COLUMN: omitting it preserves the current one). */ public boolean isNullableSpecified() { return nullableSpecified; } - /** Whether the DDL explicitly stated a comment (#65329 MODIFY COLUMN omit-preserves). */ + /** Whether the DDL explicitly stated a comment (MODIFY COLUMN: omitting it preserves the current one). */ public boolean isCommentSpecified() { return commentSpecified; } diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorColumnEvolutionOps.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorColumnEvolutionOps.java new file mode 100644 index 00000000000000..7d56b586257137 --- /dev/null +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorColumnEvolutionOps.java @@ -0,0 +1,151 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.connector.api; + +import org.apache.doris.connector.api.ddl.ConnectorColumnPath; +import org.apache.doris.connector.api.ddl.ConnectorColumnPosition; +import org.apache.doris.connector.api.handle.ConnectorTableHandle; + +import java.util.List; + +/** + * Column schema evolution: {@code ALTER TABLE ... ADD / DROP / RENAME / MODIFY COLUMN}, flat and nested. + * + *The whole domain is optional — every default fails loud with a "not supported" message, which is + * the correct answer for a connector whose format cannot evolve columns.
+ * + *Minimum implementation set:
+ *A heterogeneous gateway connector must route BOTH groups to its sibling. Routing only the flat six is a + * silent trap: nested {@code ALTER} statements then hit the gateway's own throwing default even though the + * sibling supports them.
+ * + *Dotted column paths (e.g. {@code s.b}, {@code arr.element.c}, {@code m.value}) are carried neutrally by + * {@link ConnectorColumnPath}; a single-part path targets a top-level column. The fe-core bridge routes + * top-level ADD/DROP/RENAME/MODIFY through the flat ops and reserves the {@code *NestedColumn} ops for the + * nested case, except {@link #modifyColumnComment}, which is the sole entrypoint for + * {@code MODIFY COLUMN ... COMMENT} (flat and nested alike). Distinct names (rather than overloads of the flat + * {@code String} / {@link ConnectorColumn} ops) keep {@code Mockito.any()} / null call sites in connector tests + * unambiguous.
+ */ +public interface ConnectorColumnEvolutionOps { + + /** + * Adds a column to the table at the given position. + * + * @param position where to place the column ({@link ConnectorColumnPosition#FIRST} / + * {@link ConnectorColumnPosition#after(String)}); {@code null} appends at the end. + */ + @ConnectorMustImplement(when = "the connector supports column evolution") + default void addColumn(ConnectorSession session, ConnectorTableHandle handle, + ConnectorColumn column, ConnectorColumnPosition position) { + throw new DorisConnectorException("ADD COLUMN not supported"); + } + + /** Adds multiple columns to the table, appended in order. */ + @ConnectorMustImplement(when = "the connector supports column evolution") + default void addColumns(ConnectorSession session, ConnectorTableHandle handle, + ListA connector that has no per-column statistics returns {@code Optional.empty()} from + * {@link ConnectorStatisticsOps#getColumnStatistics} — that, not a sentinel instance, is how + * "unavailable" is expressed. Returning a present value with an individual field set to -1 is + * still allowed and means "this one number is unknown".
*/ public final class ConnectorColumnStatistics { - /** Sentinel indicating no per-column statistics are available. */ - public static final ConnectorColumnStatistics UNKNOWN = - new ConnectorColumnStatistics(-1, -1, -1, -1); - private final long rowCount; private final long ndv; private final long numNulls; diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorContractValidator.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorContractValidator.java index eeacdf232b6441..36ba204b1a4e26 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorContractValidator.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorContractValidator.java @@ -18,6 +18,7 @@ package org.apache.doris.connector.api; import org.apache.doris.connector.api.handle.WriteOperation; +import org.apache.doris.connector.api.write.ConnectorWritePlanProvider; import java.util.Set; @@ -27,11 +28,22 @@ * the doc contracts the removed {@code ConnectorCapability} javadoc stated only in prose. * *Because the invariants are static properties of a connector's own capability declarations, they are - * enforced by the per-connector contract tests (which build each connector and call {@link #validate}), + * meant to be enforced by per-connector contract tests (which build the connector and call {@link #validate}), * not at catalog registration: reading a connector's write capabilities constructs its write plan provider, * which for some connectors (e.g. iceberg) eagerly builds the live remote catalog — too costly and * outage-fragile to run on the FE metadata-replay / CREATE CATALOG path. This class stays available to any * caller that already holds an eagerly-built connector and wants the same check.
+ * + *Actual coverage today (do not read the paragraph above as a statement that every connector is + * checked): five connectors' tests call {@link #validate} — iceberg, elasticsearch, maxcompute, jdbc and + * hive. Every invariant below now has a positive sample on a real connector: maxcompute for the local-sort + * arm, hive for the hash arm and for their mutual exclusion. A connector added without such a test is simply + * unchecked.
+ * + *Note also that this validator reads the CONNECTOR-LEVEL provider, while the engine's write path resolves + * the provider per table ({@code Connector.getWritePlanProvider(ConnectorTableHandle)}). A heterogeneous + * gateway connector can therefore be self-consistent here and still answer differently per table, which is by + * design and out of this validator's scope.
*/ public final class ConnectorContractValidator { @@ -39,30 +51,38 @@ private ConnectorContractValidator() {} /** @throws IllegalStateException if any write-capability invariant is violated. */ public static void validate(Connector connector, String catalogType) { - SetExtends the fine-grained sub-interfaces for schema, table, * pushdown, statistics, and write operations. Each sub-interface * provides sensible defaults so that connectors only need to - * override the methods they actually support.
+ * override the methods they actually support. Because every method has a default, the compiler forces NO + * method on an implementor: an empty implementation compiles and loads but serves nothing, so each + * sub-interface documents its own minimum implementation set. + * + *Lifecycle: exactly one instance per catalog per statement, obtained by the engine through its + * single entry point ({@code PluginDrivenMetadata}, which owns this contract) and closed deterministically + * when the statement ends. One instance must not be shared across threads. Note that + * {@link ConnectorStatisticsOps} — inherited here — is exempt from the query-thread assumption; see its class + * javadoc.
*/ public interface ConnectorMetadata extends ConnectorSchemaOps, @@ -50,11 +56,6 @@ public interface ConnectorMetadata extends ConnectorIdentifierOps, Closeable { - /** Returns connector-level properties. */ - default MapThe predicate is expressed in the connector-neutral {@link ConnectorExpression} pushdown grammar, - * NOT a source-specific shape — the engine NEVER discriminates by connector here; it applies whatever the - * connector returns. This mirrors the engine-agnostic residual-predicate model.
+ * NOT a source-specific shape — the engine NEVER discriminates by connector here. It does, however, accept + * only a narrow subset of that grammar on this path, because it has to translate the expression back into + * an engine predicate: conjunction (AND), the five comparisons EQ / LT / LE / GT / GE, a column reference + * that binds BY NAME to one of the scan's visible output columns, and STRING literals. Anything else + * (OR, NOT, IN, LIKE, arithmetic, a non-STRING literal, an unbindable column) makes the reverse conversion + * throw and fails the query — deliberately loud, so a silently dropped correctness filter is impossible. + * Keep what you return inside that subset. * *The default returns an EMPTY list: a connector with no synthetic scan predicate adds nothing, so the * plan is byte-identical. iceberg/paimon/jdbc/... inherit this empty default; only a connector that opts in @@ -229,6 +235,11 @@ default ConnectorTableHandle applyTopnLazyMaterialization(ConnectorSession sessi } + /** + * Releases whatever this per-statement instance holds. Called by the engine exactly once per statement in + * the normal path, but implementations must be idempotent — a failed statement can reach close from + * more than one unwind path. Must not throw for an already-closed instance. + */ @Override default void close() throws IOException { } diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorMustImplement.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorMustImplement.java new file mode 100644 index 00000000000000..492da94fdbe6b7 --- /dev/null +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorMustImplement.java @@ -0,0 +1,50 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.connector.api; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Marks a default SPI method that a connector is nevertheless expected to override. + * + *
Every method of {@link ConnectorMetadata} and its sub-interfaces has a default body, so the compiler + * forces nothing on an implementor: an empty implementation compiles and loads, and then answers "no tables" + * / "table not found" to users instead of failing. This annotation is the machine-readable half of the fix — + * it says WHICH methods make up the minimum implementation set; the interface's class javadoc says WHY, and + * what the visible symptom of skipping one is.
+ * + *Nothing in production reads this annotation: there is no registration-time validation and no build gate + * (a check that has to understand which capability a connector declared cannot be written as a text match). + * The retention is {@code RUNTIME} only so that a unit test can reflect over it and pin the set, which makes + * promoting a method to "must implement" a deliberate, reviewed change rather than a silent one.
+ */ +@Documented +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.METHOD) +public @interface ConnectorMustImplement { + + /** + * The precondition that makes this method mandatory. Empty means unconditional — every connector must + * override it. Otherwise a short phrase naming the capability or situation that triggers the obligation. + */ + String when() default ""; +} diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorPartitionInfo.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorPartitionInfo.java index 2eb569d2642b3d..a433626757124a 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorPartitionInfo.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorPartitionInfo.java @@ -54,9 +54,13 @@ public final class ConnectorPartitionInfo { * {@link #partitionValueNullFlags} — i.e. value {@code i} is the value of the {@code i}-th * {@code key=value} segment of {@link #partitionName}, decoded exactly as fe-core's legacy name parse * would produce it (so the connector supplies what fe-core used to re-parse out of the name). - * Empty means "not supplied": fe-core then falls back to parsing {@link #partitionName} itself - * (unchanged behavior). A connector that lists partitions for the MVCC partition-item path - * (hive/paimon/iceberg/hudi) supplies this so fe-core does not re-run the hive-style parse. + * + *A connector that lists partitions for the MVCC partition-item path (hive/paimon/iceberg/hudi) + * MUST supply this. There is no name-parsing fallback in fe-core any more. fe-core checks that the + * number of values matches the number of partition columns and fails that partition when it does not; + * the failure is caught per partition, so the visible effect of leaving this empty is not an error but a + * silent degradation: every partition is skipped, the table is treated as unpartitioned, and partition + * pruning is lost (only a warn line in the FE log records it).
*/ private final ListThe unit is load-bearing, not decorative. The engine subtracts this from the + * wall clock to decide whether the table has been quiet long enough to serve a query from + * the SQL cache. Because it clamps "now" to at least this value (a guard against FE/metadata + * clock skew), a value from any other scale - a source-native version, a commit id, a + * timestamp in another unit - makes that difference come out as zero forever and SILENTLY + * disables the SQL cache for this table AND for every table queried alongside it. There is + * no error and no EXPLAIN signal; the only symptom is that the cache never engages. Report + * {@link #UNKNOWN} rather than a value in a different unit. + */ public long getLastModifiedMillis() { return lastModifiedMillis; } diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorPartitionListingOps.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorPartitionListingOps.java new file mode 100644 index 00000000000000..bf31c7468d0597 --- /dev/null +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorPartitionListingOps.java @@ -0,0 +1,64 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.connector.api; + +import org.apache.doris.connector.api.handle.ConnectorTableHandle; +import org.apache.doris.connector.api.pushdown.ConnectorExpression; + +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +/** + * Enumerating a table's partitions. + * + *
Minimum implementation set: a connector whose tables can be partitioned implements + * {@link #listPartitionNames} and {@link #listPartitions}. A connector without partitioned tables implements + * nothing here and the empty defaults are correct for it.
+ * + *When supplying {@link ConnectorPartitionInfo} objects, note that the ordered partition values are + * mandatory on the MVCC partition-item path — see that class for what silently happens when they are + * omitted.
+ */ +public interface ConnectorPartitionListingOps { + + /** + * Lists all partition display names (e.g., {@code "year=2024/month=01"}). + * + *Should be cheap and avoid loading per-partition metadata.
+ */ + @ConnectorMustImplement(when = "the connector has partitioned tables") + default ListConnectors should push the filter into the metastore / catalog when + * possible. {@code filter} is empty when the caller wants the full list.
+ */ + @ConnectorMustImplement(when = "the connector has partitioned tables") + default ListOptional: implement this interface, or do not. It is not part of {@link ConnectorMetadata}, so a + * connector that has no remote SQL dialect never sees these methods and owes them nothing. Implementing the + * interface IS the declaration — there is no accompanying capability flag to keep in step with it (a + * {@code SUPPORTS_PASSTHROUGH_QUERY} flag existed and was removed: it was a second overridable answer to the + * question "can this connector run my SQL", and a connector could declare it while implementing nothing).
+ * + *Both methods take a SQL string the user wrote. A connector that implements them owns the consequences: + * the engine does not parse, rewrite or authorize the statement beyond the catalog-level privilege check on + * the entry points, so an implementation must send it under the catalog's own credentials and must not widen + * what those credentials can reach.
+ * + *Minimum implementation set: whichever of the two the connector actually supports. Each defaults to + * refusing, so implementing the interface for the {@code query()} TVF alone does not silently claim + * {@code CALL EXECUTE_STMT} as well.
+ */ +public interface ConnectorPassthroughSqlOps { + + /** + * Executes a DML statement (INSERT/UPDATE/DELETE) on the remote source verbatim, for + * {@code CALL EXECUTE_STMT(catalog, stmt)}. Nothing is returned: the statement's effect is remote. + */ + default void executeStmt(ConnectorSession session, String stmt) { + throw new DorisConnectorException("executeStmt not supported"); + } + + /** + * Returns the column metadata of an arbitrary remote query, for the {@code query()} table-valued function + * (typically via the remote driver's prepared-statement metadata, without running the query). + */ + default ConnectorTableSchema getColumnsFromQuery(ConnectorSession session, String query) { + throw new DorisConnectorException("getColumnsFromQuery not supported"); + } +} diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorPropertyMetadata.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorPropertyMetadata.java deleted file mode 100644 index f53e2d916c6095..00000000000000 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorPropertyMetadata.java +++ /dev/null @@ -1,120 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -package org.apache.doris.connector.api; - -import java.util.Objects; - -/** - * Describes a configuration property that a connector exposes. - * - * @paramWhen this returns {@code false}, the engine will strip any conjuncts - * containing CAST expressions from the filter before passing it to the connector. - * The default is {@code true} (CASTs are pushed down).
+ *This switch governs ONE of the two pushdown paths. Returning {@code false} makes the engine + * drop CAST-containing conjuncts from the RESIDUAL predicate it builds for the scan node. The + * {@link #applyFilter} path never consults it: there the engine converts every conjunct and hands the + * result to the connector regardless of what this method answers.
+ * + *And on either path a CAST does not arrive as a CAST. The forward converter unwraps it and + * pushes the child expression, so {@code CAST(dt AS INT) = 20240101} reaches the connector as + * {@code dt = 20240101}, with nothing marking it as coerced — a connector cannot detect the situation by + * inspecting what it received. The risk is therefore carried by the connector: a connector that turns such + * a comparison into remote filtering (e.g. metastore partition pruning from equality / IN predicates on + * partition columns) will UNDER-return rows whenever the remote comparison semantics differ from Doris's + * coercion, and BE cannot recover the difference because the data was never scanned.
+ * + *The default is {@code true} (CASTs are pushed down), which is an opt-OUT polarity and the one + * exception to this module's "capabilities default to false" rule; it is kept for compatibility. Do not + * copy the polarity into a new switch. A connector that delegates filtering to a remote system with + * different type coercion rules (e.g. a JDBC database) overrides this to {@code false}, optionally driven + * by session configuration.
* - *Connectors that delegate filtering to remote systems with different type - * coercion rules (e.g., JDBC databases) may override this to disable CAST - * pushdown when the session configuration indicates it is unsafe.
+ *Every shipped connector answers this deliberately; none of them merely inherits. jdbc, paimon + * and maxcompute return {@code false}; iceberg, elasticsearch and the trino bridge state {@code true} at + * their own metadata class, each recording what it does with the predicate and that {@code true} is an + * accepted risk rather than a safety claim. hive and hudi declare nothing because for them the switch is + * INERT — their scan planning ignores the residual filter entirely, and the predicate they do consume + * arrives on the {@link #applyFilter} path, which is never gated on this method. A new connector should + * make the same deliberate choice, starting from {@code false}.
*/ default boolean supportsCastPredicatePushdown(ConnectorSession session) { return true; diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorSchemaOps.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorSchemaOps.java index fe7c6a485128ce..6436bed1a7a04f 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorSchemaOps.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorSchemaOps.java @@ -50,37 +50,34 @@ default ConnectorDatabaseMetadata getDatabase( } /** - * Whether this connector supports CREATE DATABASE. Defaults to false so the FE - * {@code CREATE DATABASE IF NOT EXISTS} remote existence precheck applies only to - * connectors that can actually create databases; connectors that cannot keep their - * existing "CREATE DATABASE not supported" behavior unchanged. + * Creates a new database with the given name and properties. The default throws, which IS the + * declaration that this connector cannot create databases — there is no separate boolean to keep in + * sync with it (a {@code supportsCreateDatabase()} switch existed and was removed: it could only + * restate whether this method was overridden, and getting the two out of step broke + * {@code CREATE DATABASE IF NOT EXISTS} with no compile error and no failing test). + * + *{@code IF NOT EXISTS} is handled by the engine, not here: it consults {@link #databaseExists} + * first and only calls this when the answer is no. A connector that cannot create databases therefore + * still wants a truthful {@link #databaseExists} — that is what makes {@code IF NOT EXISTS} on an + * already-existing database succeed rather than report "not supported".
*/ - default boolean supportsCreateDatabase() { - return false; - } - - /** Creates a new database with the given name and properties. */ default void createDatabase(ConnectorSession session, String dbName, MapThis is the only drop-database entry point: a 3-arg form without {@code force} used to exist and + * this overload defaulted to it, so {@code DROP DATABASE ... FORCE} silently became a non-cascading drop + * that then failed on a non-empty database. A connector that cannot cascade should reject {@code force} + * explicitly instead.
*/ default void dropDatabase(ConnectorSession session, String dbName, boolean ifExists, boolean force) { - dropDatabase(session, dbName, ifExists); + throw new DorisConnectorException( + "DROP DATABASE not supported"); } } diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorSession.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorSession.java index 320805bacd0faf..e4f5fdd833efaf 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorSession.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorSession.java @@ -71,7 +71,20 @@ default OptionalThe two namespaces are merged silently. A name present in both resolves to the session value + * and the caller cannot tell which side answered — convenient for a knob that a catalog sets as a default + * and a user overrides per query, wrong for anything where the origin matters (a catalog-level credential + * or endpoint must not be overridable by a session variable of the same name). When the origin matters, + * read the namespace directly: {@link #getSessionProperties()} or {@link #getCatalogProperties()}.
+ * + * @param name the property name, as spelled in the FE session-variable registry or in the catalog properties + * @param type one of {@code String}, {@code Integer}, {@code Long} or {@code Boolean} (any other type is + * rejected); the stored string is parsed into it + */The whole domain is optional, and only meaningful for a format that has a snapshot log with named + * references and a mutable partition spec. Every default fails loud with a "not supported" message.
+ * + *Minimum implementation set: the two groups are independent, but each is all-or-nothing. Implement all + * four of {@link #createOrReplaceBranch} / {@link #createOrReplaceTag} / {@link #dropBranch} / + * {@link #dropTag} together — a half set lets a user create a branch that they then cannot drop. Likewise + * implement {@link #addPartitionField} / {@link #dropPartitionField} / {@link #replacePartitionField} + * together.
+ */ +public interface ConnectorSnapshotRefOps { + + /** Creates or replaces a named branch (snapshot ref) on the table. */ + @ConnectorMustImplement(when = "the connector exposes branches and tags") + default void createOrReplaceBranch(ConnectorSession session, ConnectorTableHandle handle, + BranchChange branch) { + throw new DorisConnectorException("CREATE/REPLACE BRANCH not supported"); + } + + /** Creates or replaces a named tag (snapshot ref) on the table. */ + @ConnectorMustImplement(when = "the connector exposes branches and tags") + default void createOrReplaceTag(ConnectorSession session, ConnectorTableHandle handle, + TagChange tag) { + throw new DorisConnectorException("CREATE/REPLACE TAG not supported"); + } + + /** Drops a named branch (snapshot ref) from the table. */ + @ConnectorMustImplement(when = "the connector exposes branches and tags") + default void dropBranch(ConnectorSession session, ConnectorTableHandle handle, + DropRefChange branch) { + throw new DorisConnectorException("DROP BRANCH not supported"); + } + + /** Drops a named tag (snapshot ref) from the table. */ + @ConnectorMustImplement(when = "the connector exposes branches and tags") + default void dropTag(ConnectorSession session, ConnectorTableHandle handle, + DropRefChange tag) { + throw new DorisConnectorException("DROP TAG not supported"); + } + + /** Adds a partition field (column reference + optional transform) to the table's partition spec. */ + @ConnectorMustImplement(when = "the connector supports partition-spec evolution") + default void addPartitionField(ConnectorSession session, ConnectorTableHandle handle, + PartitionFieldChange change) { + throw new DorisConnectorException("ADD PARTITION FIELD not supported"); + } + + /** Drops a partition field from the table's partition spec. */ + @ConnectorMustImplement(when = "the connector supports partition-spec evolution") + default void dropPartitionField(ConnectorSession session, ConnectorTableHandle handle, + PartitionFieldChange change) { + throw new DorisConnectorException("DROP PARTITION FIELD not supported"); + } + + /** Replaces a partition field (removes the old field, adds the new one) in the table's partition spec. */ + @ConnectorMustImplement(when = "the connector supports partition-spec evolution") + default void replacePartitionField(ConnectorSession session, ConnectorTableHandle handle, + PartitionFieldChange change) { + throw new DorisConnectorException("REPLACE PARTITION FIELD not supported"); + } +} diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorStatisticsOps.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorStatisticsOps.java index 55fe9dd731b91b..700868d7832dc5 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorStatisticsOps.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorStatisticsOps.java @@ -26,6 +26,20 @@ /** * Operations for retrieving table-level statistics from a connector. + * + *These methods may run OFF the query thread, on engine background pools, and the engine pins no + * thread context classloader on any of them. The pools reached today are the column-statistics cache + * loader ({@code STATS_FETCH}), the analysis-job executor that runs {@code ANALYZE ... WITH SAMPLE}, and the + * external row-count refresh executor. (The snapshot-aware {@code getTableStatistics} overload is invoked on + * the query thread only, but do not rely on that: treat every method here as background-callable.)
+ * + *Consequence for a connector: if an implementation — or a bundled library it calls — loads classes + * reflectively BY NAME, it must pin the thread context classloader to its own plugin classloader for the + * duration of the call and restore it afterwards. Without that pin the name resolves against the engine's own + * copy of the class on a background thread, which surfaces as an intermittent + * {@code ClassCastException} / {@code NoClassDefFoundError} that only appears while statistics are being + * collected. The hive connector's {@code listFileSizes} implementation does this pinning itself and is the + * pattern to copy; the engine's side of the boundary states explicitly that it does not pin here.
*/ public interface ConnectorStatisticsOps { @@ -90,8 +104,10 @@ default long estimateDataSizeByListingFiles(ConnectorSession session, ConnectorT * factor, then does the Doris-type slot-width math itself. Unlike {@link #estimateDataSizeByListingFiles} it * neither partition-samples nor sums, because the sampler needs the individual file sizes. A potentially * expensive full remote listing, so connectors that cannot enumerate files cheaply must NOT override it - * (default empty -> the sampler falls back to scale factor 1). Best-effort: an override must return empty on - * any listing error rather than throw (statistics must not fail a query). + * (default empty -> the sampler falls back to scale factor 1). An override must let a listing error + * propagate rather than swallow it: an empty list is indistinguishable from "this table has no files", so + * swallowing turns an unreachable metastore into a silently wrong scale factor of 1 while the sample still + * runs. Propagating fails only the ANALYZE task that asked, never an unrelated query. */ default ListThe whole domain is optional — a read-only connector implements nothing here and every default + * fails loud with a "not supported" message, which is the correct answer for it.
+ * + *Minimum implementation set: to support {@code CREATE TABLE}, override + * {@link #createTable(ConnectorSession, ConnectorCreateTableRequest)}. There is deliberately only ONE create + * entry point: a narrower {@code (schema, properties)} overload used to exist and the request overload + * degraded to it, silently dropping the partition spec, the bucket spec and {@code IF NOT EXISTS} — a + * connector implementing only the narrow form reported success on a partitioned {@code CREATE TABLE} and + * created an unpartitioned table. Nothing implemented it, so it is gone rather than documented. + * {@link #dropTable}, {@link #renameTable} and {@link #truncateTable} are independent and optional.
+ */ +public interface ConnectorTableDdlOps { + + /** + * Creates a table with full DDL semantics (partition, bucket, {@code IF NOT EXISTS}). + * + *The request carries everything the statement said; a connector that cannot honor part of it must + * reject that part rather than ignore it, because a {@code CREATE TABLE} reporting success after dropping + * the partition spec is indistinguishable from one that worked.
+ * + * @throws DorisConnectorException if the connector cannot create tables, or cannot honor the request + */ + @ConnectorMustImplement(when = "the connector supports CREATE TABLE") + default void createTable(ConnectorSession session, + ConnectorCreateTableRequest request) { + throw new DorisConnectorException( + "CREATE TABLE not supported"); + } + + /** Drops the specified table. */ + default void dropTable(ConnectorSession session, + ConnectorTableHandle handle) { + throw new DorisConnectorException( + "DROP TABLE not supported"); + } + + /** Renames the table identified by {@code handle} to {@code newName} within the same database. */ + default void renameTable(ConnectorSession session, + ConnectorTableHandle handle, String newName) { + throw new DorisConnectorException( + "RENAME TABLE not supported"); + } + + /** + * Truncates the table identified by {@code handle}. When {@code partitions} is non-empty only those + * partitions are truncated; {@code null} / empty truncates the whole table. + * + *Connectors that support {@code TRUNCATE TABLE} override this. The default throws, matching the + * pre-flip behavior of the generic bridge (which had no truncate route for the SPI path).
+ * + * @throws DorisConnectorException if the connector does not support truncate + */ + default void truncateTable(ConnectorSession session, + ConnectorTableHandle handle, ListThis is the one domain no connector can skip. The four methods marked unconditional below are + * overridden by all eight shipped connectors, and {@link #buildTableDescriptor} by seven of them. Leaving them + * at their defaults is not an error the user sees as an error: {@link #listTableNames} answers an empty list, + * so {@code SHOW TABLES} looks like an empty database, and {@link #getTableHandle} answers empty, so a query + * looks like a mistyped table name. Only {@link #getTableSchema} and {@link #getColumnHandles} fail loud.
+ * + *Minimum implementation set:
+ *Note that {@link #getTableComment} addresses a table by NAME, not by handle. A heterogeneous gateway + * connector routes foreign tables to a sibling by the concrete handle type, so it cannot route a name-only + * method that way; if you are writing a gateway, handle it explicitly.
+ */ +public interface ConnectorTableMetadataOps { + + /** Retrieves a table handle for the given database and table name. */ + @ConnectorMustImplement + default OptionalThe names are WITHOUT any "$" prefix; fe-core composes the + * "{baseTable}${sysName}" reference name. Default: empty (no system + * tables). Implemented by connectors that expose system tables.
+ */ + @ConnectorMustImplement(when = "the connector exposes system tables") + default ListThe returned handle is connector-internal and carries whatever the + * connector needs (system-table name, scan-routing hints, etc.); it is + * opaque to fe-core. {@code sysName} is the bare name (no "$").
+ */ + @ConnectorMustImplement(when = "the connector exposes system tables") + default OptionalA connector whose partitioned tables expose their partition rows through the generic + * partition-values TVF (e.g. hive) overrides this to return {@code true} for that sys-table name; + * such a name need NOT return a handle from {@link #getSysTableHandle} (the TVF path never consults + * it). fe-core needs the kind at discovery time (before any handle is fetched), so it cannot be + * inferred from an empty {@code getSysTableHandle}. {@code sysName} is the bare name (no + * {@code "$"}).
+ */ + @ConnectorMustImplement(when = "a system table is served by the generic partition-values function") + default boolean isPartitionValuesSysTable(ConnectorSession session, + ConnectorTableHandle baseTableHandle, String sysName) { + return false; + } + + /** Returns the schema (columns, format, etc.) for the given table. */ + @ConnectorMustImplement + default ConnectorTableSchema getTableSchema( + ConnectorSession session, ConnectorTableHandle handle) { + throw new DorisConnectorException( + "getTableSchema not implemented"); + } + + /** + * Returns the schema AT {@code snapshot.getSchemaId()} — the schema as of the + * pinned snapshot, for time-travel reads under schema evolution. + * + *The default ignores the snapshot and returns the latest schema via + * {@link #getTableSchema(ConnectorSession, ConnectorTableHandle)}. A connector that + * supports schema-at-snapshot overrides this to resolve the schema version.
+ */ + @ConnectorMustImplement(when = "the connector supports time travel or schema evolution") + default ConnectorTableSchema getTableSchema( + ConnectorSession session, ConnectorTableHandle handle, + ConnectorMvccSnapshot snapshot) { + return getTableSchema(session, handle); + } + + /** + * Renders the native {@code SHOW CREATE TABLE} DDL for a table, fetching schema FRESH from the underlying + * metastore at call time (bypassing any connector-side table cache) so the returned statement always + * reflects the latest remote schema. + * + *This is a LAZY, per-call interception point used ONLY by {@code ShowCreateTableCommand}. It intentionally + * does NOT participate in the {@code SUPPORTS_SHOW_CREATE_DDL} capability (which gates the engine-assembled + * DDL in {@code Env.getDdlStmt} for every caller, including delegated sibling tables and the HTTP schema + * endpoint). A connector that does not natively render its own SHOW CREATE returns {@link Optional#empty()}, + * and the command falls through to the generic {@code Env.getDdlStmt} path unchanged.
+ * + * @return the full {@code CREATE TABLE} statement, or {@link Optional#empty()} to defer to the engine + */ + default OptionalThe default ignores the snapshot and returns the latest columns via + * {@link #getColumnHandles(ConnectorSession, ConnectorTableHandle)}. WHY this exists: the generic + * scan node builds column handles BEFORE it pins the snapshot, so without a snapshot-aware overload + * it can only key handles by the LATEST names. A time-travel query binds its slots to the PINNED + * (old) names, so after a RENAME the renamed column's slot misses the latest-keyed map and is + * silently dropped — crashing BE with a field-id dictionary miss on connectors whose native + * projection is name/ordinal-driven (paimon). A connector that supports schema-at-snapshot + * overrides this to key handles by the pinned names (mirrors the + * {@link #getTableSchema(ConnectorSession, ConnectorTableHandle, ConnectorMvccSnapshot)} split) and + * declares {@link #supportsColumnHandleSnapshotPin(ConnectorSession)}.
+ */ + @ConnectorMustImplement(when = "column handles must be keyed by the pinned schema's names") + default MapOnly a connector that returns {@code true} is subject to the generic node's fail-loud check + * that every bound column present in the pinned schema has a handle: for such a connector a missing + * pinned column is a genuine bug (the connector promised pinned handles but dropped one) and must + * fail with a clear error rather than being silently dropped into a BE crash. A connector that + * returns {@code false} keeps the legacy latest-keyed handles and recovers from the drop by its own + * means (e.g. iceberg rebuilds its field-id dictionary from the full pinned schema), so it is left + * on the unchanged silent-skip path.
+ */ + @ConnectorMustImplement(when = "column handles are keyed by the pinned schema's names") + default boolean supportsColumnHandleSnapshotPin(ConnectorSession session) { + return false; + } + + /** Lists all table names within the given database. */ + @ConnectorMustImplement + default ListEach connector constructs its own typed descriptor (e.g., {@code TJdbcTable}, + * {@code TEsTable}) and wraps it in a {@code TTableDescriptor}. This keeps + * connector-specific Thrift logic out of fe-core.
+ * + *The Thrift classes are provided by fe-thrift at compile time and loaded + * from the parent classloader at runtime.
+ * + * @param session connector session + * @param tableId Doris internal table ID + * @param tableName table display name + * @param dbName database name + * @param remoteName remote table name in the external data source + * @param numCols number of columns in the schema + * @param catalogId Doris internal catalog ID + * @return the table descriptor, or {@code null} if the connector does not + * need a typed descriptor (fe-core will use a generic fallback) + */ + @ConnectorMustImplement + default org.apache.doris.thrift.TTableDescriptor buildTableDescriptor( + ConnectorSession session, + long tableId, String tableName, String dbName, + String remoteName, int numCols, long catalogId) { + return null; + } +} diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorTableOps.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorTableOps.java index 22afc267669d32..60510eb64f2a2c 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorTableOps.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorTableOps.java @@ -17,488 +17,37 @@ package org.apache.doris.connector.api; -import org.apache.doris.connector.api.ddl.BranchChange; -import org.apache.doris.connector.api.ddl.ConnectorColumnPath; -import org.apache.doris.connector.api.ddl.ConnectorColumnPosition; -import org.apache.doris.connector.api.ddl.ConnectorCreateTableRequest; -import org.apache.doris.connector.api.ddl.DropRefChange; -import org.apache.doris.connector.api.ddl.PartitionFieldChange; -import org.apache.doris.connector.api.ddl.TagChange; -import org.apache.doris.connector.api.handle.ConnectorColumnHandle; -import org.apache.doris.connector.api.handle.ConnectorTableHandle; -import org.apache.doris.connector.api.mvcc.ConnectorMvccSnapshot; -import org.apache.doris.connector.api.pushdown.ConnectorExpression; - -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.Optional; - /** - * Operations on tables within a connector catalog. + * Operations on tables within a connector catalog: the aggregate of the per-domain table interfaces. + * + *This interface declares nothing itself. It exists so that {@link ConnectorMetadata} keeps one + * table-operations supertype and connectors keep compiling unchanged, while the operations themselves live in + * the domain each belongs to:
+ * + *Passing a SQL string through to the remote source is deliberately NOT here: it is the escape hatch of a + * connector whose source speaks SQL, not a table operation, and it lives in the optional + * {@link ConnectorPassthroughSqlOps}, which a connector implements or does not.
+ * + *Start with each domain's class javadoc: it states that domain's minimum implementation set + * — which methods a connector must override to work at all, which become mandatory once a given + * capability is declared, and which are optional. Every method has a default body, so the compiler demands + * nothing; those lists are the only statement of what is actually required.
*/ -public interface ConnectorTableOps { - - /** Retrieves a table handle for the given database and table name. */ - default OptionalThe names are WITHOUT any "$" prefix; fe-core composes the - * "{baseTable}${sysName}" reference name. Default: empty (no system - * tables). Implemented by connectors that expose system tables.
- */ - default ListThe returned handle is connector-internal and carries whatever the - * connector needs (system-table name, scan-routing hints, etc.); it is - * opaque to fe-core. {@code sysName} is the bare name (no "$").
- */ - default OptionalA connector whose partitioned tables expose their partition rows through the generic - * partition-values TVF (e.g. hive) overrides this to return {@code true} for that sys-table name; - * such a name need NOT return a handle from {@link #getSysTableHandle} (the TVF path never consults - * it). fe-core needs the kind at discovery time (before any handle is fetched), so it cannot be - * inferred from an empty {@code getSysTableHandle}. {@code sysName} is the bare name (no - * {@code "$"}).
- */ - default boolean isPartitionValuesSysTable(ConnectorSession session, - ConnectorTableHandle baseTableHandle, String sysName) { - return false; - } - - /** Returns the schema (columns, format, etc.) for the given table. */ - default ConnectorTableSchema getTableSchema( - ConnectorSession session, ConnectorTableHandle handle) { - throw new DorisConnectorException( - "getTableSchema not implemented"); - } - - /** - * Returns the schema AT {@code snapshot.getSchemaId()} — the schema as of the - * pinned snapshot, for time-travel reads under schema evolution. - * - *The default ignores the snapshot and returns the latest schema via - * {@link #getTableSchema(ConnectorSession, ConnectorTableHandle)}. A connector that - * supports schema-at-snapshot overrides this to resolve the schema version.
- */ - default ConnectorTableSchema getTableSchema( - ConnectorSession session, ConnectorTableHandle handle, - ConnectorMvccSnapshot snapshot) { - return getTableSchema(session, handle); - } - - /** - * Renders the native {@code SHOW CREATE TABLE} DDL for a table, fetching schema FRESH from the underlying - * metastore at call time (bypassing any connector-side table cache) so the returned statement always - * reflects the latest remote schema. - * - *This is a LAZY, per-call interception point used ONLY by {@code ShowCreateTableCommand}. It intentionally - * does NOT participate in the {@code SUPPORTS_SHOW_CREATE_DDL} capability (which gates the engine-assembled - * DDL in {@code Env.getDdlStmt} for every caller, including delegated sibling tables and the HTTP schema - * endpoint). A connector that does not natively render its own SHOW CREATE returns {@link Optional#empty()}, - * and the command falls through to the generic {@code Env.getDdlStmt} path unchanged.
- * - * @return the full {@code CREATE TABLE} statement, or {@link Optional#empty()} to defer to the engine - */ - default OptionalThe default ignores the snapshot and returns the latest columns via - * {@link #getColumnHandles(ConnectorSession, ConnectorTableHandle)}. WHY this exists: the generic - * scan node builds column handles BEFORE it pins the snapshot, so without a snapshot-aware overload - * it can only key handles by the LATEST names. A time-travel query binds its slots to the PINNED - * (old) names, so after a RENAME the renamed column's slot misses the latest-keyed map and is - * silently dropped — crashing BE with a field-id dictionary miss on connectors whose native - * projection is name/ordinal-driven (paimon). A connector that supports schema-at-snapshot - * overrides this to key handles by the pinned names (mirrors the - * {@link #getTableSchema(ConnectorSession, ConnectorTableHandle, ConnectorMvccSnapshot)} split) and - * declares {@link #supportsColumnHandleSnapshotPin(ConnectorSession)}.
- */ - default MapOnly a connector that returns {@code true} is subject to the generic node's fail-loud check - * that every bound column present in the pinned schema has a handle: for such a connector a missing - * pinned column is a genuine bug (the connector promised pinned handles but dropped one) and must - * fail with a clear error rather than being silently dropped into a BE crash. A connector that - * returns {@code false} keeps the legacy latest-keyed handles and recovers from the drop by its own - * means (e.g. iceberg rebuilds its field-id dictionary from the full pinned schema), so it is left - * on the unchanged silent-skip path.
- */ - default boolean supportsColumnHandleSnapshotPin(ConnectorSession session) { - return false; - } - - /** Lists all table names within the given database. */ - default ListConnectors should override this when they support advanced - * {@code CREATE TABLE} options. The default degrades to the legacy - * {@link #createTable(ConnectorSession, ConnectorTableSchema, Map)}, - * dropping partition / bucket / external / {@code ifNotExists} info.
- * - * @throws DorisConnectorException if the connector cannot honor the request - */ - default void createTable(ConnectorSession session, - ConnectorCreateTableRequest request) { - ConnectorTableSchema schema = new ConnectorTableSchema( - request.getTableName(), - request.getColumns(), - null, - request.getProperties()); - createTable(session, schema, request.getProperties()); - } - - /** Drops the specified table. */ - default void dropTable(ConnectorSession session, - ConnectorTableHandle handle) { - throw new DorisConnectorException( - "DROP TABLE not supported"); - } - - /** Renames the table identified by {@code handle} to {@code newName} within the same database. */ - default void renameTable(ConnectorSession session, - ConnectorTableHandle handle, String newName) { - throw new DorisConnectorException( - "RENAME TABLE not supported"); - } - - /** - * Truncates the table identified by {@code handle}. When {@code partitions} is non-empty only those - * partitions are truncated; {@code null} / empty truncates the whole table. - * - *Connectors that support {@code TRUNCATE TABLE} override this. The default throws, matching the - * pre-flip behavior of the generic bridge (which had no truncate route for the SPI path).
- * - * @throws DorisConnectorException if the connector does not support truncate - */ - default void truncateTable(ConnectorSession session, - ConnectorTableHandle handle, ListEach connector constructs its own typed descriptor (e.g., {@code TJdbcTable}, - * {@code TEsTable}) and wraps it in a {@code TTableDescriptor}. This keeps - * connector-specific Thrift logic out of fe-core.
- * - *The Thrift classes are provided by fe-thrift at compile time and loaded - * from the parent classloader at runtime.
- * - * @param session connector session - * @param tableId Doris internal table ID - * @param tableName table display name - * @param dbName database name - * @param remoteName remote table name in the external data source - * @param numCols number of columns in the schema - * @param catalogId Doris internal catalog ID - * @return the table descriptor, or {@code null} if the connector does not - * need a typed descriptor (fe-core will use a generic fallback) - */ - default org.apache.doris.thrift.TTableDescriptor buildTableDescriptor( - ConnectorSession session, - long tableId, String tableName, String dbName, - String remoteName, int numCols, long catalogId) { - return null; - } - - /** - * Lists all partition display names (e.g., {@code "year=2024/month=01"}). - * - *Should be cheap and avoid loading per-partition metadata.
- */ - default ListConnectors should push the filter into the metastore / catalog when - * possible. {@code filter} is empty when the caller wants the full list.
- */ - default ListUsed by the {@code partition_values()} TVF and by column-distinct-value - * optimizations. Inner list order matches {@code partitionColumns}.
- */ - default ListA reserved key is the right carrier only for structure that is naturally a name (a column + * name, a pre-rendered SQL clause). Anything drawn from a closed set gets a typed field instead — see + * {@link #getTableCapabilities()}, which used to be a reserved key holding a CSV of enum constant names.
*/ public static final String INTERNAL_KEY_PREFIX = "__internal."; @@ -73,30 +78,6 @@ public final class ConnectorTableSchema { */ public static final String PARTITION_COLUMNS_KEY = INTERNAL_KEY_PREFIX + "partition_columns"; - /** - * Reserved property key carrying a CSV of the table's primary-key column names (paimon). FE-only; - * stripped from the user-facing SHOW CREATE TABLE PROPERTIES(...) block. - */ - public static final String PRIMARY_KEYS_KEY = INTERNAL_KEY_PREFIX + "primary_keys"; - - /** - * Reserved property key carrying a CSV of {@link ConnectorCapability#name()} values that THIS specific - * table supports, refining the connector-wide {@link Connector#getCapabilities()} set per-table. - * - *A uniform-format connector (e.g. iceberg — every table orc/parquet) declares a scan capability for all - * its tables connector-wide. A heterogeneous connector (e.g. hive — orc/parquet/text/json/csv/view/hudi in - * one catalog) whose eligibility is per-table file-format gated cannot: Top-N lazy materialization and - * nested-column pruning are orc/parquet-only, and blanket-declaring them connector-wide would over-admit a - * text/json table (a correctness bug for nested-column pruning, which reads NULL leaves without field ids). - * Such a connector instead emits the capability name here, per-table, computed from that table's format.
- * - *fe-core reads it ADDITIVELY (a capability counts as supported if it is in the connector-wide set OR in - * this per-table list) from the already-cached schema — no remote round-trip and no file-format inspection - * in fe-core. Single-format connectors never emit it and are unaffected. Stripped from the user-facing - * SHOW CREATE TABLE PROPERTIES(...) block.
- */ - public static final String PER_TABLE_CAPABILITIES_KEY = INTERNAL_KEY_PREFIX + "connector.per-table-capabilities"; - /** * Reserved property key carrying a CSV of the table's distribution (bucketing) column names, already * lowercased. A heterogeneous connector (hive) whose bucketing varies per table cannot express it as a @@ -117,18 +98,29 @@ public final class ConnectorTableSchema { */ public static final SetA uniform-format connector (e.g. iceberg — every table orc/parquet) declares its capabilities + * connector-wide. A heterogeneous connector (e.g. hive — orc/parquet/text/json/csv/view/hudi in one + * catalog) whose eligibility is per-table file-format gated cannot: Top-N lazy materialization and + * nested-column pruning are orc/parquet-only, and blanket-declaring them connector-wide would over-admit a + * text/json table (a correctness bug for nested-column pruning, which reads NULL leaves without field + * ids). Such a connector computes this set from the table's own format instead.
+ * + *fe-core reads it ADDITIVELY (a capability counts as supported if it is in the connector-wide set OR + * here) from the already-cached schema — no remote round-trip and no file-format inspection in fe-core. + * Only the capabilities {@link ConnectorCapability} documents as table-scoped are consulted here; a + * catalog-scoped one placed in this set is ignored, so put it in {@link Connector#getCapabilities()}.
+ */ + public SetA connector that has no statistics at all returns {@code Optional.empty()} from + * {@link ConnectorStatisticsOps#getTableStatistics} — that, not a sentinel instance, is how + * "unavailable" is expressed. Returning a present value with an individual field set to -1 is + * still allowed and means "this one number is unknown" (e.g. hive knows the total size but not + * the row count).
*/ public final class ConnectorTableStatistics { - /** Sentinel value indicating statistics are not available. */ - public static final ConnectorTableStatistics UNKNOWN = - new ConnectorTableStatistics(-1, -1); - private final long rowCount; private final long dataSize; diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorTestResult.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorTestResult.java index 2249af7e56c407..031f1dff837534 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorTestResult.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorTestResult.java @@ -17,56 +17,39 @@ package org.apache.doris.connector.api; -import java.util.Collections; -import java.util.Map; import java.util.Objects; /** * Result of a connector connectivity test. * *Connectors return this from {@link Connector#testConnection} to report - * whether the data source is reachable. Individual sub-components (e.g., - * metastore, object storage) can report separate results via - * {@link #getComponentResults()}.
+ * whether the data source is reachable. A connector that probes several + * sub-components (e.g. metastore plus object storage) reports the first + * failure it hits, in {@link #getMessage()}. */ public class ConnectorTestResult { private final boolean success; private final String message; - private final MapShape contract for complex types (enforced in the canonical constructor, so every factory + * and convenience constructor is covered):
+ *This is checked eagerly because a violation is invisible downstream: fe-core's converter fills a
+ * missing STRUCT field name with {@code "col" + index} and turns a childless ARRAY/MAP into
+ * {@code ARRAY
The whole domain is optional — a connector without views implements nothing here, and the engine + * never enters this domain unless the connector declares {@link ConnectorCapability#SUPPORTS_VIEW} (it checks + * the capability before merging view names into {@code SHOW TABLES}).
+ * + *Minimum implementation set, once {@code SUPPORTS_VIEW} is declared:
+ *Faithful, lossless neutralization of the fe-core {@code org.apache.doris.analysis.ColumnPath} * (an ordered, non-empty list of identifier parts). A single-part path targets a top-level column; diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ddl/ConnectorColumnPosition.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ddl/ConnectorColumnPosition.java index 4ab5fafc536c6f..18b7d326a2337a 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ddl/ConnectorColumnPosition.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ddl/ConnectorColumnPosition.java @@ -21,8 +21,8 @@ /** * The position of a column in an {@code ALTER TABLE ADD/MODIFY COLUMN} clause, carried neutrally - * across the SPI by {@link org.apache.doris.connector.api.ConnectorTableOps#addColumn} / - * {@link org.apache.doris.connector.api.ConnectorTableOps#modifyColumn}. + * across the SPI by {@link org.apache.doris.connector.api.ConnectorColumnEvolutionOps#addColumn} / + * {@link org.apache.doris.connector.api.ConnectorColumnEvolutionOps#modifyColumn}. * *
Faithful, lossless neutralization of the fe-catalog {@code ColumnPosition}, which is exactly
* {@code FIRST | AFTER Carries partition / bucket / external / {@code IF NOT EXISTS} information
- * absent from the legacy
- * {@code createTable(session, ConnectorTableSchema, Map Carries everything the statement said: the columns, the partition and bucket specs,
+ * {@code IF NOT EXISTS} and the user properties. A connector that cannot honor one of them must reject it
+ * rather than ignore it — reporting success after dropping the partition spec produces a table the user did
+ * not ask for, with no error to go on. {@code partitionSpec} and {@code bucketSpec} are nullable when the
* underlying DDL omits them. {@code initialValues} is only meaningful for {@code LIST} / {@code RANGE} styles. For {@code LIST} partitions, {@code values} contains the literal list of
- * permitted values (each inner list is one tuple matching the partition columns).
- * For {@code RANGE} partitions, {@code values} contains exactly two tuples
- * representing the [lower, upper) bound.
Extends the marker {@link ConnectorTransactionHandle} so that existing - * APIs that traffic in opaque handles continue to work without change.
*/ -public interface ConnectorTransaction extends ConnectorTransactionHandle, Closeable { +public interface ConnectorTransaction extends Closeable { /** Stable transaction ID assigned by the connector. */ long getTransactionId(); @@ -76,7 +73,7 @@ default long getUpdateCnt() { /** * Applies an optional engine-extracted, target-only write constraint used for write-time optimistic - * conflict detection (O5-2). The engine extracts, from the analyzed DELETE/UPDATE/MERGE plan, the + * conflict detection. The engine extracts, from the analyzed DELETE/UPDATE/MERGE plan, the * conjuncts that reference only the target table's own columns (slot origin-table == target, excluding * synthetic {@code $row_id} / metadata / join columns) and hands the connector a neutral * {@link ConnectorPredicate} at plan time, before {@code begin}/{@code commit}. diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/handle/ConnectorTransactionHandle.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/handle/ConnectorTransactionHandle.java deleted file mode 100644 index 6320a186f744de..00000000000000 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/handle/ConnectorTransactionHandle.java +++ /dev/null @@ -1,24 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -package org.apache.doris.connector.api.handle; - -/** - * Opaque transaction handle. - */ -public interface ConnectorTransactionHandle { -} diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/handle/ConnectorWriteHandle.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/handle/ConnectorWriteHandle.java index 2b90af2a70153c..70534c6c3191eb 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/handle/ConnectorWriteHandle.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/handle/ConnectorWriteHandle.java @@ -30,7 +30,7 @@ * *Carries the engine-resolved facts about a single DML write: the target * table handle, the column list, whether it is an OVERWRITE, and the static - * partition spec ({@link #getWriteContext}). The connector reads these to build + * partition spec ({@link #getStaticPartitionSpec}). The connector reads these to build * its Thrift data sink.
*/ public interface ConnectorWriteHandle { @@ -47,13 +47,14 @@ public interface ConnectorWriteHandle { /** * The static partition spec (partition column name -> value) for a statically partitioned write, * carried from the bound sink to {@code planWrite}; an EMPTY map when the write is not statically - * partitioned. Despite the "write context" name (once envisioned as a free-form bag), the only content - * the sole producer ({@code PluginDrivenTableSink.bindDataSink} -> - * {@code PluginDrivenInsertCommandContext.getStaticPartitionSpec}) ever populates is the static - * partition spec, and the write providers (hive/iceberg/maxcompute) all consume it as such (iceberg - * ships it verbatim as {@code TDataSink.static_partition_values}). + * partitioned. Both sides now spell it the same way: the sole producer is + * {@code PluginDrivenTableSink.bindDataSink} -> {@code PluginDrivenInsertCommandContext + * .getStaticPartitionSpec}, and the write providers (hive/iceberg/maxcompute) consume it as such + * (iceberg ships it verbatim as {@code TDataSink.static_partition_values}). It was once called + * {@code getWriteContext} and envisioned as a free-form bag; nothing ever put anything else in it, so the + * name was corrected rather than the contract widened. A future free-form channel would be a new method. */ - MapKept OFF {@link ConnectorTransaction} so the shared transaction contract carries no source-specific - * methods: the engine rewrite driver ({@code ConnectorRewriteDriver}) checks {@code instanceof - * RewriteCapableTransaction} before it calls, turning "unsupported" from a runtime throw into a type - * mismatch. A connector whose transaction is not rewrite-capable simply does not implement this.
+ *Kept OFF {@link ConnectorTransaction} so the shared transaction contract carries no methods only some + * connectors can honor: the engine rewrite driver probes for this capability before it calls, turning + * "unsupported" from a runtime throw into a type mismatch. A connector whose transaction is not + * rewrite-capable simply does not implement this.
*/ public interface RewriteCapableTransaction { @@ -45,8 +45,8 @@ public interface RewriteCapableTransaction { /** * Compaction rewrite: the number of new compacted data files this transaction added, available only * AFTER {@code ConnectorTransaction#commit()} (the count is materialized from the BE-reported commit - * fragments during commit). Feeds the procedure's {@code added_data_files_count} result column — the one - * rewrite-result statistic the engine driver cannot compute from its planning groups. + * fragments during commit). This is the one rewrite statistic the engine cannot compute from the planning + * groups, which is why it has to be reported back rather than summed. */ int getRewriteAddedDataFilesCount(); } diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/mvcc/ConnectorMvccPartitionView.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/mvcc/ConnectorMvccPartitionView.java index 993ec0e2693784..5226742a954ee6 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/mvcc/ConnectorMvccPartitionView.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/mvcc/ConnectorMvccPartitionView.java @@ -26,7 +26,7 @@ * refresh path of a plugin-driven MVCC table. * *The generic table model ({@code PluginDrivenMvccExternalTable}) builds its partition view from - * {@link org.apache.doris.connector.api.ConnectorTableOps#listPartitions} by default, which always + * {@link org.apache.doris.connector.api.ConnectorPartitionListingOps#listPartitions} by default, which always * yields {@code LIST} partitions keyed on a last-modified timestamp. Connectors whose partitions are * intrinsically ranges (e.g. iceberg's {@code YEAR}/{@code MONTH}/{@code DAY}/{@code HOUR} time * transforms) need {@code RANGE} partition items plus a snapshot-id freshness marker to preserve the diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/mvcc/ConnectorMvccSnapshot.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/mvcc/ConnectorMvccSnapshot.java index 39e0296edb28cf..53fbad7dddfe8d 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/mvcc/ConnectorMvccSnapshot.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/mvcc/ConnectorMvccSnapshot.java @@ -28,22 +28,22 @@ * *
Returned by {@code ConnectorMetadata.beginQuerySnapshot} and friends. * Used by the engine as the MVCC pin for all subsequent reads of the same - * table handle within a query, and serialized into BE scan ranges so the - * read path sees a consistent version.
+ * table handle within a query. + * + *The pin lives entirely inside FE: this type is not serializable and the engine never places it in a + * scan range. What reaches BE is whatever the connector itself put there — the engine hands the snapshot back + * to the connector ({@code ConnectorMetadata.applySnapshot}), the connector weaves the version into its own + * table handle, and its scan plan provider decides what to emit.
*/ public final class ConnectorMvccSnapshot { private final long snapshotId; - private final long timestampMillis; - private final String description; private final long schemaId; private final MapRead this page before adding a connector or extending these interfaces. It records the rules the + * existing connectors follow, and — where a rule is not yet fully honored — says so explicitly instead of + * describing the target state as if it were already true. Every rule below can be checked against the code + * that is cited with it; when the two disagree, the code wins and this page is the bug.
+ * + *Use {@code Optional} for a missing value and {@code null} only for an absent + * subsystem; they are not two competing idioms for the same thing.
+ * + *Current deviations. (a) A new switch must default to {@code false} (opt-in), so that adding a + * capability can never change an existing connector's behavior. One legacy switch violates this and defaults + * to {@code true}: {@code ConnectorPushdownOps.supportsCastPredicatePushdown(ConnectorSession)} — see the + * risk paragraph in its own javadoc, and do not copy its polarity. (b) The {@code supportsXxx()} shape also + * appears on non-provider interfaces ({@code ConnectorPushdownOps}, {@code ConnectorSchemaOps}, + * {@code ConnectorTableOps}); those are per-operation switches on the interface that owns the operation, + * which is consistent with layer 2, but there is no provider object to attach them to. (c) Per-table + * refinement of {@link ConnectorCapability} is honored for only five of the thirteen constants today, and + * each constant's own javadoc states its scope (catalog-only, or catalog ∪ per-table) and why; do not + * assume a newly added constant can be refined per table.
+ * + *Two engine paths translate connector exceptions into user-facing errors, and they accept different + * types. Getting this wrong does not break anything visibly; it only degrades the error the user sees into an + * internal FE exception.
+ * + *Everything else ({@code UnsupportedOperationException}, {@code IllegalStateException}, plain + * {@code RuntimeException}) is untranslated and surfaces as an internal error. The one accepted use of + * {@code UnsupportedOperationException} is the "this optional SPI method is not implemented" idiom.
+ * + *Subclass {@link DorisConnectorException} when the CALLER must distinguish outcomes, not to build + * a taxonomy for its own sake. Shipped precedent: {@code HiveDirectoryListingException} exists so the scan + * path can catch only a skippable listing failure while a plain {@link DorisConnectorException} + * still fails the query loud.
+ * + *Fail loud vs. degrade silently. Anything the user explicitly asked for — DDL, writes, an explicit + * {@code ANALYZE ... WITH SAMPLE} — must fail loud, because a swallowed error there produces a wrong result + * or a wrong statistic that then looks authoritative. Anything the engine attempts opportunistically for + * optimization must degrade silently (return empty / a default, and log), because it must never be able to + * fail a query that would otherwise succeed.
+ * + *BE is C++ and has no plugin mechanism, so a payload destined for BE has to be a thrift type. That is the + * only reason thrift appears in this module, and the complete list of places it does is:
+ * + *Do not add a method that takes a thrift type as a PARAMETER. A new return value that must carry a BE + * payload may only reuse a type already on that list. {@code fe-thrift} is a {@code provided} dependency + * here: it is supplied by the parent classloader at runtime and is not part of a plugin package.
+ * + *Current deviation. {@code fe-connector-spi} deliberately avoids thrift even for BE-bound data + * (it passes an enum NAME as a {@code String}, a neutral broker-address type, and a thrift enum's integer + * VALUE as an {@code int}). That is a local convention in that module, not a module-level invariant: it + * depends on this module, which depends on {@code fe-thrift}. Whether to unify the two styles is out of + * scope here.
+ * + *The split is by WHO IMPLEMENTS the type. A type the connector implements and the engine consumes + * belongs here ({@code fe-connector-api}). A type the engine implements and the connector consumes, + * plus the service-discovery entry point, belongs in {@code fe-connector-spi}. The dependency runs one way, + * {@code spi} → {@code api}; nothing here may reference the {@code spi} package.
+ * + *The two module names are inverted relative to common usage (elsewhere, including Trino, + * "SPI" names what the plugin implements and "API" names the engine services it consumes). Here + * {@code fe-connector-api} holds the types a connector implements and {@code fe-connector-spi} holds the + * handful it consumes. The names are deliberately NOT being changed — renaming would touch every connector + * pom and the engine's dependencies for a naming win only — so read the content, not the name.
+ * + *Current deviation. {@link ConnectorSession}, {@code ConnectorHttpSecurityHook} and + * {@code ConnectorValidationContext} are engine-implemented yet live here. This is a known misplacement, + * left alone because moving them would rewrite imports in every connector.
+ * + *These interfaces are the only behavioral contract a connector author has: connectors are packaged and + * class-loaded separately and cannot read the engine. A javadoc line that describes what the engine does with + * a value, and is wrong, is worse than no line at all — it produces an implementation that compiles, runs, + * and silently returns wrong results. So when documenting engine behavior, name the engine class (and + * ideally the method) that does it, so the next reader can re-verify instead of trusting the sentence.
+ * + *Pick the channel by the SCOPE of the value. Only the second one obliges anyone to touch the engine, so + * a knob that can be catalog-scoped should be.
+ * + *There is deliberately no declarative property-descriptor mechanism here. One was carried over + * from Trino ({@code ConnectorPropertyMetadata} plus two {@link Connector} getters) and shipped for several + * releases with zero implementors and zero consumers; it was removed rather than wired up, because in Doris + * it would not have solved the thing it looks like it solves — there is no {@code SET SESSION + * catalog.property} syntax to feed, and per-catalog values already arrive through the map above. If you find + * yourself wanting the engine to validate unknown catalog property keys, that belongs in + * {@code ConnectorProvider.validateProperties}, which already exists and already runs.
+ */ +package org.apache.doris.connector.api; diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/procedure/ConnectorProcedureOps.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/procedure/ConnectorProcedureOps.java index 77d044e68bb506..8f5b626481ea0a 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/procedure/ConnectorProcedureOps.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/procedure/ConnectorProcedureOps.java @@ -46,8 +46,14 @@ public interface ConnectorProcedureOps { /** - * Returns the procedure names this connector supports for {@code ALTER TABLE EXECUTE}, used by the - * engine for routing, validation, and {@code SHOW}-style discovery. + * Returns the procedure names this connector supports for {@code ALTER TABLE EXECUTE}. + * + *The engine does NOT use this list to route or to validate. Routing is by table type plus the + * execution mode reported by {@link #getExecutionMode}; an unknown procedure name is rejected by the + * connector inside {@link #execute} (the engine's "is this supported" check answers {@code true} + * unconditionally and says so at its call site). The engine's only reader of this list is a + * {@code SHOW}-style discovery helper that has no production caller today, so a name omitted here still + * executes and a name added here does not by itself become reachable.
*/ ListOnly meaningful for a procedure whose {@link #getExecutionMode} is {@code DISTRIBUTED}
* (today: iceberg {@code rewrite_data_files}). The default throws — a connector with no distributed
@@ -96,6 +103,33 @@ default List Implementations must render LOCALLY: no table load, no remote call, no authorization scope. The
+ * engine also calls this on the "nothing to rewrite" path, where it deliberately has not opened a
+ * transaction (and therefore has nothing to scope an authorized call to). Only meaningful for a procedure whose {@link #getExecutionMode} is {@code DISTRIBUTED}. The default
+ * throws rather than returning an empty result, so a connector that declares {@code DISTRIBUTED} and
+ * forgets to implement this fails loudly instead of silently answering with an empty result set. This is the planning analogue of {@link ConnectorProcedureResult}: the connector owns the
- * file-selection / bin-pack / partition-grouping decision (iceberg's {@code rewrite_data_files} criteria —
- * file size range, delete ratio, min-input-files, max group size, partition grouping); the engine only
- * orchestrates the distributed reads/writes. No live SDK object crosses the seam — only neutral
- * {@code String} paths and primitive counts. The model is atomic replacement by file path: each group's files are read once and replaced by what the
+ * rewrite writes. This is the planning analogue of {@link ConnectorProcedureResult}: the connector owns which
+ * files to touch and how to group them (size ranges, delete ratios, minimum inputs, per-partition grouping —
+ * all of it connector-defined); the engine only orchestrates the distributed reads/writes. No live SDK object
+ * crosses the seam — only neutral {@code String} paths and primitive counts. Three of the four — {@code dataFileCount}, {@code totalSizeBytes}, {@code deleteFileCount} — are straight
+ * sums of the {@link ConnectorRewriteGroup}s the connector itself planned; the engine ran one
+ * {@code INSERT-SELECT} per group, so counting the groups is part of orchestrating them, and no unit
+ * conversion or reinterpretation happens on the way. The remaining one, {@code addedDataFileCount}, comes from
+ * {@code RewriteCapableTransaction#getRewriteAddedDataFilesCount()} and is valid only after commit; it is the
+ * one figure the engine cannot derive from planning. Note it is the SECOND constructor argument, not the last
+ * — the constructor follows the result row's conventional column order, not this split. On the "nothing to
+ * rewrite" path the engine reports all zeros without having opened a transaction. The field names mirror {@code ConnectorRewriteGroup}'s getters so the summation is obvious, and the
+ * constructor order matches the order these values conventionally appear in a rewrite result row, so there is
+ * one ordering to remember rather than two. All four are neutral words: how they are NAMED and TYPED in the
+ * result set is the connector's decision, not this class's. Supported operators: EQ, NE, LT, LE, GT, GE, EQ_FOR_NULL. {@code left} is normally a {@link ConnectorColumnRef} and {@code right} a {@link ConnectorLiteral}, but
+ * neither is guaranteed: the engine builds this node from whatever the two operands converted to. A connector
+ * that only supports column-op-literal must check both sides and drop the conjunct otherwise (see the package
+ * javadoc, Rule 1) rather than assume. Operator semantics, all with SQL three-valued logic — a comparison against NULL is UNKNOWN, not false: {@code EQ_FOR_NULL} must be split into two cases; collapsing them loses rows: A connector whose dialect has no null-safe form must drop the whole conjunct (package javadoc, Rule 1).
+ * It must never substitute a narrower predicate. Shipped precedent for the correct shape: the iceberg and
+ * trino converters branch on {@code isNull()}; the maxcompute converter has no null-safe operator remotely and
+ * therefore refuses to push the conjunct at all. {@code pattern} is normally a {@link ConnectorLiteral} holding a {@code String}, but the engine does not
+ * guarantee it — a non-literal pattern must be dropped, not guessed at (see the package javadoc, Rule 1). {@code LIKE} dialect (Doris semantics — translate to these, not to your remote system's defaults): {@code REGEXP} is UNANCHORED (Doris/MySQL semantics): the pattern may match anywhere inside the
+ * value. A remote engine whose regex is whole-string anchored — Lucene's {@code regexp} query, for example —
+ * is not a valid target for a verbatim hand-off: {@code REGEXP 'bc'} matches {@code 'abcd'} in Doris
+ * and matches nothing when anchored. Anchoring narrows the predicate, which Rule 1 forbids; either rewrite
+ * the pattern into the remote form or drop the conjunct. Do not turn a pattern into a prefix/suffix/contains match unless it is provably equivalent.
+ * {@code 'abc%'} is a prefix match. {@code 'a_c%'} is not — {@code _} is a wildcard, so {@code 'abc'} must
+ * match the pattern but does not start with {@code a_c}. Neither is {@code 'a\%%'} (that is "starts with
+ * {@code a%}"), nor anything with a {@code %} left in the body. A pushed prefix that is stricter than the
+ * user's pattern makes the connector skip files, and rows skipped at planning time can never be recovered by
+ * BE. Case folding: do not introduce it. Neither operator carries a collation here, so translate
+ * case-sensitively. A remote form that matches a case-insensitive SUPERSET is permitted by Rule 1 (BE
+ * re-checks the original predicate); one that matches fewer rows is not. The value is a standard Java type. The engine produces exactly these eight shapes — see the package
+ * javadoc (Rule 4) for the full Doris-type-to-Java-class table: {@code null}, {@code Boolean}, {@code Long},
+ * {@code Double}, {@code BigDecimal}, {@code String}, {@code LocalDate}, {@code LocalDateTime}. Two consequences worth stating outright, because both have produced wrong connector code:
+ * {@code Integer} never arrives (every integral type, {@code TINYINT} through {@code BIGINT}, is a
+ * {@code Long}; the {@link #ofInt} factory exists for tests), and {@code LARGEINT} arrives as a decimal
+ * {@code String}, not as a {@code BigInteger}. A converter that switches on the Java class must have a
+ * fall-through for the {@code String} case rather than assuming a numeric column carries a numeric object. A NULL literal is a legal operand of any comparison, so check {@link #isNull()} FIRST. In particular
+ * {@code ConnectorComparison.Operator#EQ_FOR_NULL} means {@code IS NULL} only when this is null and plain
+ * equality otherwise; conflating the two loses rows (see {@link ConnectorComparison}). The engine only acts on {@code null}. {@code PluginDrivenScanNode.convertPredicate} clears
+ * every conjunct when this is {@code null}, and removes NOTHING when it is non-null — matching residual
+ * sub-expressions back to individual conjuncts is not implemented, so returning "the half I could not
+ * push" earns no credit for the half that was pushed. The connector-side consequence is asymmetric: Per-conjunct credit is available, but through the other protocol:
+ * {@code ConnectorScanPlanProvider.getScanNodePropertiesResult} reporting not-pushed conjunct indices. This is the ONLY vocabulary a connector receives for filter pushdown. A connector translates it into its
+ * own dialect (an iceberg {@code Expression}, a paimon {@code Predicate}, an ES query DSL document, a SQL
+ * {@code WHERE} fragment). Getting a translation subtly wrong does not fail the query — it silently returns
+ * the wrong number of rows, and {@code EXPLAIN} shows nothing unusual. So read Rule 1 before writing a
+ * converter; the three shipped row-loss bugs fixed in this area were all the same mistake. Rules 1–3 are the contract. Rules 4–6 record what the engine actually produces and what it actually does
+ * with your answer, each citing the engine class so it can be re-verified (see Rule 6 of
+ * {@code org.apache.doris.connector.api} — the engine is not on a connector's classpath, so an unverifiable
+ * sentence here is worse than no sentence). A pushed predicate that is too WIDE is recoverable: the engine keeps the conjuncts and BE re-evaluates
+ * them, so the extra rows are filtered out before the user sees them. A pushed predicate that is too NARROW
+ * is not recoverable by anything: the connector has already skipped files, partitions or row groups, and the
+ * rows never reach BE. That asymmetry — not convenience — is why dropping is always allowed and narrowing
+ * never is. The precondition on "wide is safe". It holds only while the engine still has the conjuncts. If
+ * you tell the engine you consumed them (a {@code null} remaining filter, or index-level tracking — Rule 5),
+ * then a predicate that is too wide starts returning EXTRA rows. Read Rule 5 before claiming full pushdown. Two concrete instances of the mistake, both shipped and both fixed: The same expression tree is used for three purposes, and "drop it" is only safe in two of them: The engine already enforces the third row on its side: {@code UnboundExpressionToConnectorPredicateConverter}
+ * is strictly all-or-nothing and throws when any part of the {@code WHERE} cannot be represented neutrally.
+ * A connector consuming such a predicate must hold the symmetric invariant — a conjunct it cannot turn into
+ * file pruning is a hard error there, not a silent drop. {@link ConnectorColumnRef#getColumnName()} carries the name as the engine knows it. Case handling is
+ * currently decided per connector and is deliberately not specified: paimon lower-cases both sides
+ * before matching field names, jdbc maps through its own remote-name table. If your remote system is
+ * case-sensitive, decide explicitly rather than assuming the engine normalized anything. One producer. Every tree is built by {@code ExprToConnectorExpressionConverter} (fe-core). The
+ * other two entry points — {@code NereidsToConnectorExpressionConverter} for DELETE/UPDATE/MERGE and
+ * {@code UnboundExpressionToConnectorPredicateConverter} for {@code EXECUTE ... WHERE} — route literals and
+ * types back through it, so literal encoding is identical on all three paths. Shape of the root. {@code convertConjuncts} returns the single node directly when there is
+ * exactly one conjunct (the root is NOT a {@link ConnectorAnd}), and a boolean {@code true}
+ * {@link ConnectorLiteral} when there are none. Do not assume an {@code AND} root. Two arrival paths, only one of which strips CAST. Literal value domain. {@link ConnectorLiteral#getValue()} is one of exactly eight Java shapes.
+ * Note that {@code Integer} is never produced and that {@code LARGEINT} arrives as a decimal
+ * {@code String}: Consequence for everyone else: all three {@code applyFilter} implementations shipped today return the
+ * original expression as the residual, so their predicates are also evaluated on BE. That is correct but not
+ * free — and it is the reason a slightly-too-wide pushdown is survivable today. {@code ExprToConnectorExpressionConverter.fallback} turns any expression it does not model into a
+ * {@link ConnectorFunctionCall} whose name is the rendered Doris SQL text and whose argument list is
+ * empty. It is not a function call at all. Matching function names blindly will make a connector
+ * translate {@code my_udf(a, b) = 1} as a call to a function literally named {@code "my_udf(a, b) = 1"}.
+ * jdbc handles this deliberately: an argument-less call whose name is not a plain identifier is emitted as a
+ * pre-rendered SQL fragment ({@code JdbcQueryBuilder}). Two related shapes to expect: Anything you cannot interpret confidently falls under Rule 1: drop the conjunct. This exists for FE HTTP endpoints that deliberately speak a specific source's HTTP dialect — today
+ * {@code ESCatalogAction}, whose two endpoints proxy an Elasticsearch mapping lookup and search. Such an
+ * endpoint narrows to the catalog type it emulates BEFORE reaching for this capability; the capability itself
+ * says only "this connector can forward an HTTP request", and it is the caller that knows what shape of
+ * request the source understands. Consequently a connector implementing this is NOT agreeing to serve arbitrary requests from arbitrary
+ * engine code: the path is composed by an endpoint that was written for that specific source. {@link #NULL_PARTITION_NAME} is how "this partition column's value is a genuine SQL NULL" is spelled
+ * inside a partition NAME. A connector whose source has its own spelling (paimon's
+ * {@code partition.default-name}, for instance) normalizes to this one when it renders a partition name. The literal is frozen byte for byte. It is the historical hive value, and a partition name is a
+ * persisted, user-visible identity: it lands in view and materialized-view definitions, in the
+ * {@code partition_values()} table-function output, and in the {@code columns_from_path} bytes BE parses.
+ * Only the Java symbol may be renamed; changing the string breaks already-persisted objects. This constant does not replace the structured null flag on {@code ConnectorPartitionInfo}, and the
+ * flag does not replace this constant — see {@code PluginDrivenMvccExternalTable#toListPartitionItem}. The flag
+ * exists so FE can build a TYPED {@code NullLiteral} (parsing this string as an INT or DATE partition value
+ * would throw and silently drop the partition, making a partitioned table look unpartitioned); the name exists
+ * for partition identity and for BE's path parsing. Whether a value IS null must be declared by the connector
+ * through the flag: two connectors render this identical string with different NULL semantics, so fe-core
+ * never decides nullness by comparing against it. There is deliberately no shared "normalize a partition value" helper here. The rule for turning a value
+ * into a null flag is per-connector — hudi's directory-name rule also treats a literal {@code \N} as NULL,
+ * which would corrupt real data for a connector whose partition values are typed — so each connector derives
+ * its own flags (see {@code HudiScanRange}, {@code HiveScanRange}, {@code PaimonScanRange},
+ * {@code IcebergScanRange}). The engine uses this to determine which Thrift scan range structure
- * to generate. For example, {@link ConnectorScanRangeType#FILE_SCAN}
- * produces TFileScanRange. Some connectors (e.g., JDBC) can push the limit into the remote query
- * to reduce data transfer. The default delegates to the 4-arg planScan,
- * ignoring the limit. This is the one method a scanning connector must implement. Everything the engine can tell it about
+ * the scan — columns, remaining filter, row limit, pruned partitions, {@code COUNT(*)} pushdown — arrives
+ * on {@link ConnectorScanRequest}, and a connector consumes what it can serve and ignores the rest. It
+ * replaced a chain of four overloads in which only the shortest was abstract, so implementing the obvious
+ * one silently discarded the limit, the partition pruning and the count signal. The engine computes partition pruning (Nereids {@code SelectedPartitions}) and
- * threads the surviving partitions here so partition-aware connectors can build a read
- * session over only those partitions instead of the whole table. The default ignores
- * {@code requiredPartitions} and delegates to the 5-arg variant, so connectors that do
- * not support partition pushdown are unaffected. Contract for {@code requiredPartitions}: The "pruned to zero partitions" case (a partition predicate that matches nothing) is
- * short-circuited by the engine before this method is called, so an empty list here always
- * means "not pruned / scan all", never "scan nothing". When {@code countPushdown} is true, the engine has determined the query is a no-grouping
- * {@code COUNT(*)} (Nereids {@code getPushDownAggNoGroupingOp()==COUNT}) and BE is already in
- * count mode. A connector that can produce a precomputed row count for (some of) its splits
- * should emit it so BE serves the count from metadata instead of materializing rows
- * (e.g. Paimon's {@code DataSplit.mergedRowCount()}). The default ignores the flag and delegates
- * to the 6-arg variant, so connectors without a metadata row count are unaffected and keep the
- * normal scan. {@code false} (the default) keeps a connector out of that governance, which is right for anything
+ * read through JNI or over a remote protocol: it never populates the BE file cache, so evaluating an
+ * admission rule for it only spends the lookup. A connector whose ranges BE reads natively (the lake
+ * formats reading parquet/orc off object storage) returns {@code true}, and does so regardless of which
+ * catalog type its tables live under — that is the point of asking the connector rather than matching a
+ * catalog type name. Mirrors {@link #supportsTableSample}'s opt-in shape. Called once per partition batch when the engine drives batch-mode split generation
* (see {@link #supportsBatchScan}). Each call should build a read session over exactly the
- * given {@code partitionBatch} and return that batch's scan ranges. The default delegates to
- * the 6-arg {@link #planScan} with {@code partitionBatch} as the required partitions, which is
- * correct for connectors whose {@code planScan} builds one read session per partition set
- * (e.g. MaxCompute). A connector whose {@code planScan} is not partition-set-scoped must
- * override this method (and {@link #supportsBatchScan}) before enabling batch mode.
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ * Rule 1 — translate exactly, or drop the whole conjunct. Never narrow it
+ *
+ *
+ * can express it exactly -> push it down
+ * cannot express it exactly -> drop the WHOLE conjunct and let BE evaluate it
+ * never -> push a "close enough" predicate that matches FEWER rows
+ *
+ *
+ *
+ *
+ *
+ * Rule 2 — which direction is safe depends on what the predicate is FOR
+ *
+ *
+ *
+ *
+ * Use Dropping a conjunct means Correct behavior Scan pushdown ({@link org.apache.doris.connector.api.ConnectorPushdownOps#applyFilter},
+ * {@code ConnectorScanPlanProvider.planScan})
+ * the filter widens; BE re-evaluates and covers it
+ * dropping allowed, narrowing forbidden Write-time conflict detection
+ * ({@link org.apache.doris.connector.api.handle.ConnectorTransaction#applyWriteConstraint})
+ * conflict detection widens, i.e. gets more conservative
+ * dropping allowed {@code ALTER TABLE ... EXECUTE ... WHERE} rewrite scoping
+ * MORE files get rewritten — dropping the whole WHERE rewrites the entire table
+ * must throw; dropping is not allowed Rule 3 — column names, and what is NOT specified here
+ *
+ * Rule 4 — what the engine actually produces
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ * Doris type Java class NULL literal (any type) {@code null} — see {@link ConnectorLiteral#isNull()} BOOLEAN {@code Boolean} TINYINT / SMALLINT / INT / BIGINT {@code Long} (never {@code Integer}) FLOAT / DOUBLE {@code Double} DECIMAL* {@code BigDecimal} CHAR / VARCHAR / STRING {@code String} DATE / DATEV2 {@code LocalDate} DATETIME / DATETIMEV2 {@code LocalDateTime} anything else, incl. LARGEINT, IPV4/IPV6, JSON
+ * {@code String} from {@code Expr.getStringValue()} Rule 5 — telling the engine what you consumed: two protocols, one of which does nothing
+ *
+ *
+ *
+ *
+ * What you return What the engine does {@link FilterApplicationResult#getRemainingFilter()} == {@code null}
+ * {@code PluginDrivenScanNode.convertPredicate} clears ALL conjuncts. BE re-evaluates nothing, so
+ * every predicate you were given must have been pushed EXACTLY. {@code getRemainingFilter()} != {@code null} (any expression, including the original one)
+ * Nothing is removed. The engine keeps every conjunct; matching residual sub-expressions back
+ * to individual conjuncts is not implemented. Returning "the half I did not push" does not give you
+ * credit for the half you did. {@code ScanNodePropertiesResult} with not-pushed conjunct indices
+ * {@code PluginDrivenScanNode.pruneConjunctsFromNodeProperties} really does prune: conjuncts whose
+ * index you did NOT report are removed. This is the only protocol with per-conjunct granularity;
+ * it is used by exactly one shipped connector (es). Rule 6 — {@link ConnectorFunctionCall} is also the fallback carrier. Do not match it by name
+ *
+ *
+ *
+ *
+ *
- *
- *
- *
The keys the engine reads are declared in {@link ScanNodePropertyKeys}; use those constants rather + * than string literals. Any other key is connector-private (the engine passes the map back to + * {@link #populateScanLevelParams} and {@link #appendExplainInfo}, so a connector can carry its own + * information through it) and should be namespaced with the connector's own name.
+ * + *Override this face OR {@link #getScanNodePropertiesResult}, not both. The engine only ever + * calls the result face, whose default implementation delegates here; a connector that overrides both + * with different content has the map returned by this method silently discarded.
+ * * @param session the current session * @param handle the table handle (may have been updated by applyFilter) * @param columns the columns to read @@ -422,18 +338,6 @@ default MapThe engine may use this to pre-allocate resources or decide - * scan parallelism before calling {@link #planScan}.
- */ - default long estimateScanRangeCount(ConnectorSession session, - ConnectorTableHandle handle) { - return -1; - } - /** * Returns scan-node-level properties along with filter pushdown results. * @@ -443,8 +347,22 @@ default long estimateScanRangeCount(ConnectorSession session, * refer to the AND children of the filter expression, in the same order as * the conjuncts list. * - *The default wraps {@link #getScanNodeProperties} with an empty not-pushed set, - * meaning all conjuncts are assumed to have been pushed.
+ *The indices are into the conjunct list AS RECEIVED, i.e. after + * {@code PluginDrivenScanNode.buildRemainingFilter} has dropped CAST-wrapped conjuncts when + * {@link org.apache.doris.connector.api.ConnectorPushdownOps#supportsCastPredicatePushdown} is false; the + * engine maps them back to the original positions itself and always keeps the conjuncts it never sent. + * When {@code filter} is a single non-AND node (the engine does not wrap a lone conjunct in an AND), + * index 0 refers to that whole expression.
+ * + *The default wraps {@link #getScanNodeProperties} in a result WITHOUT conjunct tracking, which makes + * {@code PluginDrivenScanNode.pruneConjunctsFromNodeProperties} remove nothing: every conjunct is still + * evaluated on BE. That is the safe default — it is NOT "all conjuncts are assumed pushed". Reporting an + * empty not-pushed set through {@link ScanNodePropertiesResult#withPushdownTracking} is the opposite + * claim (everything was pushed, prune them all), so use it only when the pushdown was exact.
+ * + *This is the face the engine calls. A connector that needs conjunct tracking overrides this + * one and leaves {@link #getScanNodeProperties} alone — overriding both is how the {@code Map} face's + * result gets silently discarded.
* * @param session the current session * @param handle the table handle (may have been updated by applyFilter) @@ -457,7 +375,7 @@ default ScanNodePropertiesResult getScanNodePropertiesResult( ConnectorTableHandle handle, ListCurrently used by Paimon to pass the serialized Paimon Table - * object to BE for JNI-based reading.
- * - * @param nodeProperties the scan node properties - * @return serialized table string, or null - */ - default String getSerializedTable(MapWhere that transaction may be kept. This method releases state opened by a DIFFERENT call, yet + * a provider instance cannot carry it: providers are built fresh per acquisition ({@code + * Connector.getScanPlanProvider}), so the instance that opened the transaction is generally not the one + * asked to release it. The state therefore belongs to the CONNECTOR — a connector-scoped, query-id-keyed + * registry that both the opening {@code planScan} and this callback reach through the connector instance + * they were built from (this is what hive does). A provider that stashes the transaction in its own field + * leaks it, and no test will say so. The same rule applies to any other cross-call state a provider + * appears to need: put it on the connector, keyed by query id, and clean it up here.
+ * * @param queryId the finishing query's id (== {@link ConnectorSession#getQueryId()}) */ default void releaseReadTransaction(String queryId) { diff --git a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/scan/ConnectorScanRange.java b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/scan/ConnectorScanRange.java index 7f2ab1508b4632..902e64691fde47 100644 --- a/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/scan/ConnectorScanRange.java +++ b/fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/scan/ConnectorScanRange.java @@ -30,18 +30,17 @@ /** * Represents a unit of work (a split/range) for scanning a connector table. * - *Each scan range maps to one BE scan task. The {@link #getRangeType() range type} - * determines how the engine converts this range into the appropriate Thrift - * scan range structure for BE execution.
+ *Each scan range maps to one BE scan task. The Thrift shape is chosen by the range itself: override + * {@link #populateRangeParams} to build a format-specific {@code TTableFormatFileDesc} (iceberg, hudi, paimon + * and es all do), and use {@link #getTableFormatType()} to select the BE-side reader. The default + * {@code populateRangeParams} dumps {@link #getProperties()} into the generic {@code jdbc_params} map, which + * is what the JNI-based readers consume.
* *Connectors produce scan ranges via {@link ConnectorScanPlanProvider#planScan}, * and the engine converts them to {@code TScanRangeLocations} for dispatch.
*/ public interface ConnectorScanRange extends Serializable { - /** Returns the scan range type, which determines BE processing. */ - ConnectorScanRangeType getRangeType(); - /** Returns the file path, if applicable. */ default OptionalThe unit is connector-defined, NOT universally bytes. A file-backed connector reports the byte + * count to read from {@link #getStart()} (hive, iceberg), but a connector whose split is not a byte range + * reports what its own SDK gave it — MaxCompute's row-offset splits carry a ROW count, and its default + * splits, like Paimon's JNI ranges, carry -1. BE is told the value only through the range's own thrift, so + * each connector stays self-consistent.
+ * + *Consequently the ENGINE must not read this as a byte size to drive a generic size-based decision + * (sampling, split merging, parallelism): a value meaning rows would silently mis-size the plan for that + * connector. Such a feature is gated on an explicit capability the connector opts into (mirroring + * {@code supportsTableSample}), and only a connector whose {@code getLength} is genuinely a byte count may + * opt in.
+ */ default long getLength() { return -1; } @@ -60,9 +73,8 @@ default long getLength() { /** * Returns the file format (e.g., "parquet", "orc", "csv", "jni"). * - *For {@link ConnectorScanRangeType#FILE_SCAN}, this determines the - * BE reader. For non-file types (JDBC, ES, etc.), return "jni" to use - * the JNI scanner framework.
+ *For a range read by a native file reader this determines that reader. For a range served through a + * JNI scanner (jdbc, es, and the system-table paths), return "jni".
*/ default String getFileFormat() { return "jni"; @@ -182,7 +194,6 @@ default boolean isNativeReadRange() { default void populateRangeParams(TTableFormatFileDesc formatDesc, TFileRangeDesc rangeDesc) { MapEach type maps to a specific Thrift scan range variant in the - * execution layer. Connectors choose the appropriate type based on - * their data access pattern:
- *Why a parameter object. These fields arrived one at a time, each as another {@code planScan} + * overload that delegated to the previous one — four in the end, of which only the shortest was abstract. A + * connector that implemented that abstract method, the obvious thing to do when reading the interface, ended + * up silently ignoring the row limit, the pruned partition set and the {@code COUNT(*)} signal: everything + * still worked, just slower, with nothing to point at. There is now one method to implement and one object + * carrying whatever the engine has; a signal added later becomes a field with a default, and no connector + * silently loses it.
+ * + *Immutable. The engine builds one per scan; {@link #withRequiredPartitions} makes the batched variant + * ({@link ConnectorScanPlanProvider#planScanForPartitionBatch}) without copying the rest by hand.
+ */ +public final class ConnectorScanRequest { + + private final ConnectorTableHandle tableHandle; + private final ListNever means "scan nothing": a predicate that prunes everything away is short-circuited by the engine + * before the connector is asked, except for a predicate-driven connector + * ({@link ConnectorScanPlanProvider#ignorePartitionPruneShortCircuit}), which is handed scan-all and + * re-plans from the filter instead.
+ */ + public ListThis is the only residual protocol the engine acts on per conjunct, and exactly one shipped + * connector uses it (es). The other channel — returning a non-null remaining filter from + * {@code ConnectorPushdownOps.applyFilter} — makes the engine keep every conjunct; see + * {@code FilterApplicationResult.getRemainingFilter} and the {@code pushdown} package javadoc, Rule 5. + * Because a pruned conjunct is not re-evaluated anywhere, only report an index as pushed when the + * translation was EXACT: a widened pushdown that would merely cost extra BE work in the other channel + * returns extra rows here.
*/ public class ScanNodePropertiesResult { @@ -39,29 +47,37 @@ public class ScanNodePropertiesResult { private final SetTwo directions share that one map:
+ *Keys NOT listed here are connector-private: the engine never reads them, so a connector may use its + * own namespaced keys (e.g. {@code paimon.*}) to carry information from its property builder to its own + * {@code populateScanLevelParams} / {@code appendExplainInfo}. Prefix them with the connector's own name so + * they cannot collide with a future engine-read key.
+ * + *Why this class exists: every literal here used to be duplicated between the engine's private constants + * and each connector's inline string. A one-letter mismatch compiles, starts, and returns wrong data + * silently (a mistyped {@link #TEXT_COLUMN_SEPARATOR} puts the whole text row into the first column). The + * literals are the historical wire values and must not be changed — only referenced.
+ */ +public final class ScanNodePropertyKeys { + + // ------------------------------------------------------------------------ + // connector -> engine + // ------------------------------------------------------------------------ + + /** + * Selects the BE file reader. Recognized values: {@code parquet}, {@code orc}, {@code text}, + * {@code csv}, {@code json}, {@code avro}, {@code es_http}. Any other value — including the key being + * absent — selects the JNI reader. + * + *This is a SCAN-level decision (BE picks the scanner implementation before it looks at any single + * range), so a connector that reads some ranges natively and some through JNI must not emit + * {@code jni} here; see {@link ConnectorScanRange#isNativeReadRange()}.
+ */ + public static final String FILE_FORMAT_TYPE = "file_format_type"; + + /** + * Comma-separated Doris column names that are encoded in the file PATH rather than in the file body, + * so BE does not try to decode them from the data. Names are taken verbatim (same case the connector + * reported in its schema). + */ + public static final String PATH_PARTITION_KEYS = "path_partition_keys"; + + /** + * Prefix for the storage configuration BE needs to read the files. The engine strips the prefix and + * passes the remaining key/value pairs through as the range's location properties, so + * {@code LOCATION_PREFIX + "s3.access_key"} arrives at BE as {@code s3.access_key}. + */ + public static final String LOCATION_PREFIX = "location."; + + /** + * The remote query text a passthrough connector rendered, used ONLY for the {@code QUERY:} line of + * EXPLAIN. It is not sent to BE through this map. + */ + public static final String REMOTE_QUERY = "query"; + + // ------------------------------------------------------------------------ + // connector -> engine: text-family file attributes + // ------------------------------------------------------------------------ + + /** + * Prefix of the file attributes for the text family of formats (TEXT / CSV / JSON). The literal is the + * historical wire value and is NOT hive-specific: hudi, iceberg and any other connector reading a + * text-family file goes through the same engine code path. + * + *Only the suffixes declared below are read by the engine. A connector may put additional + * {@code TEXT_PROPERTY_PREFIX}-prefixed keys in the map for its own consumption, but the engine + * ignores them.
+ */ + public static final String TEXT_PROPERTY_PREFIX = "hive.text."; + + /** Fully qualified name of the SerDe class; selects the text sub-dialect on BE. */ + public static final String TEXT_SERDE_LIB = TEXT_PROPERTY_PREFIX + "serde_lib"; + + /** Number of leading lines to skip (header rows). Parsed as an integer. */ + public static final String TEXT_SKIP_LINES = TEXT_PROPERTY_PREFIX + "skip_lines"; + + /** Column separator. May be a multi-byte string. */ + public static final String TEXT_COLUMN_SEPARATOR = TEXT_PROPERTY_PREFIX + "column_separator"; + + /** Line delimiter. May be a multi-byte string. */ + public static final String TEXT_LINE_DELIMITER = TEXT_PROPERTY_PREFIX + "line_delimiter"; + + /** Separator between a map entry's key and value. */ + public static final String TEXT_MAPKV_DELIMITER = TEXT_PROPERTY_PREFIX + "mapkv_delimiter"; + + /** Separator between the elements of an array / map / struct. */ + public static final String TEXT_COLLECTION_DELIMITER = TEXT_PROPERTY_PREFIX + "collection_delimiter"; + + /** Escape character; a single character. */ + public static final String TEXT_ESCAPE = TEXT_PROPERTY_PREFIX + "escape"; + + /** The literal that represents SQL NULL in the file. */ + public static final String TEXT_NULL_FORMAT = TEXT_PROPERTY_PREFIX + "null_format"; + + /** Quote character enclosing a field; a single character. */ + public static final String TEXT_ENCLOSE = TEXT_PROPERTY_PREFIX + "enclose"; + + /** {@code "true"} to strip the enclosing quotes from field values. */ + public static final String TEXT_TRIM_DOUBLE_QUOTES = TEXT_PROPERTY_PREFIX + "trim_double_quotes"; + + /** {@code "true"} when the rows are JSON documents rather than delimited text. */ + public static final String TEXT_IS_JSON = TEXT_PROPERTY_PREFIX + "is_json"; + + /** {@code "true"} to tolerate malformed JSON rows instead of failing the scan. Read only when + * {@link #TEXT_IS_JSON} is {@code "true"}. */ + public static final String TEXT_OPENX_IGNORE_MALFORMED = TEXT_PROPERTY_PREFIX + "openx_ignore_malformed"; + + // ------------------------------------------------------------------------ + // engine -> connector (never forwarded to BE as a property) + // ------------------------------------------------------------------------ + + /** + * Number of scan ranges this node will read with a native BE reader, counted from + * {@link ConnectorScanRange#isNativeReadRange()}. Injected by the engine into the property map passed to + * {@link ConnectorScanPlanProvider#appendExplainInfo}, for connectors that surface a native/JNI split + * distinction in EXPLAIN. + */ + public static final String SYNTHETIC_NATIVE_READ_SPLITS = "__native_read_splits"; + + /** Total number of scan ranges this node will read. Companion of {@link #SYNTHETIC_NATIVE_READ_SPLITS}. */ + public static final String SYNTHETIC_TOTAL_READ_SPLITS = "__total_read_splits"; + + /** + * Present with value {@code "true"} only while the engine renders a VERBOSE EXPLAIN, so a connector can + * gate VERBOSE-only output from {@link ConnectorScanPlanProvider#appendExplainInfo} without an SPI + * signature change. Absent otherwise. + */ + public static final String SYNTHETIC_EXPLAIN_VERBOSE = "__explain_verbose"; + + /** + * The row limit the engine pushed down to this scan, as a decimal string; {@code "-1"} (or any + * non-positive value) means there is none. Injected on BOTH the {@code appendExplainInfo} and the + * {@code populateScanLevelParams} paths, because a connector that can ask its source to stop early needs + * it while building thrift, not only while printing EXPLAIN. + */ + public static final String SYNTHETIC_PUSHDOWN_LIMIT = "__pushdown_limit"; + + /** + * {@code "true"} when NO filtering is left for the engine to do after the scan — every conjunct was taken + * by the connector. Companion of {@link #SYNTHETIC_PUSHDOWN_LIMIT}: limiting rows at the source is only + * correct when the source is not going to hand back rows that a leftover engine-side filter would then + * discard. + * + *PRECONDITION, or this key lies: the engine can only know which conjuncts were taken if the connector + * reported them, i.e. returned {@link ScanNodePropertiesResult#withPushdownTracking}. A connector that + * returns the plain result reads {@code "false"} here even when it did in fact push everything. Report + * tracking, or do not use this key.
+ */ + public static final String SYNTHETIC_ALL_CONJUNCTS_PUSHED = "__all_conjuncts_pushed"; + + private ScanNodePropertyKeys() { + } +} diff --git a/fe/fe-connector/fe-connector-api/src/test/java/org/apache/doris/connector/api/ConnectorMetadataSurfaceTest.java b/fe/fe-connector/fe-connector-api/src/test/java/org/apache/doris/connector/api/ConnectorMetadataSurfaceTest.java new file mode 100644 index 00000000000000..ed2a28985a0144 --- /dev/null +++ b/fe/fe-connector/fe-connector-api/src/test/java/org/apache/doris/connector/api/ConnectorMetadataSurfaceTest.java @@ -0,0 +1,179 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.connector.api; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.TreeSet; + +/** + * Freezes the public method surface of {@link ConnectorMetadata} and pins the minimum implementation set. + * + *WHY this matters: every method of {@link ConnectorMetadata} and its six {@code Ops} + * sub-interfaces has a default body, so the compiler forces nothing on an implementor and no existing test + * fails when a method silently disappears, gains a parameter, or is moved between interfaces. That is exactly + * what happened when {@code ConnectorTableOps} was split into per-domain super-interfaces: the split is only + * safe if the resulting method set is IDENTICAL, and "identical" was previously unverifiable. This test makes + * the surface a reviewed artifact — {@code src/test/resources/connector-metadata-methods.txt} is the recorded + * set, and any change to it has to be made deliberately.
+ * + *When this test goes red on purpose: a batch that intentionally deletes or adds SPI methods must + * regenerate the baseline in the same commit. That is the point — it is a speed bump on the public connector + * surface, not an obstacle. Regenerate by running this test and copying the "actual" set from the failure + * message.
+ */ +public class ConnectorMetadataSurfaceTest { + + private static final String BASELINE_RESOURCE = "/connector-metadata-methods.txt"; + + /** The six domain interfaces {@link ConnectorTableOps} aggregates. Each states its own minimum set. */ + private static final ListWHY this matters: a complex type is described by PARALLEL lists — one of child types, one
+ * of STRUCT field names, plus optional per-child metadata. Nothing downstream can detect that they
+ * disagree: fe-core's converter fills a missing STRUCT field name with {@code "col" + index} and turns
+ * a childless ARRAY/MAP into {@code ARRAY WHY this matters: connectors translate this node arm by arm into their own predicate
+ * dialect. The class documents itself as "two or more disjuncts", but nothing used to enforce it,
+ * and a degenerate or mutated node does not fail loudly on the consumer side - it silently produces
+ * a NARROWER pushed-down predicate, which makes the source skip data the query should have returned.
+ * Missing rows have no error, no warning and no EXPLAIN signal, so the invariant has to be enforced
+ * where it is cheap to check: at construction.
+ */
+public class ConnectorOrTest {
+
+ private static ConnectorExpression arm(String column) {
+ return new ConnectorColumnRef(column, ConnectorType.of("INT"));
+ }
+
+ @Test
+ public void testEmptyDisjunctListRejected() {
+ IllegalArgumentException e = Assertions.assertThrows(IllegalArgumentException.class,
+ () -> new ConnectorOr(Collections.emptyList()));
+ Assertions.assertTrue(e.getMessage().contains("at least two disjuncts"), e.getMessage());
+ }
+
+ @Test
+ public void testSingleDisjunctRejected() {
+ // A one-armed OR is the caller's logic error (the arm should have been used directly).
+ // Absorbing it silently is what lets a lost arm reach a connector unnoticed.
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> new ConnectorOr(Collections.singletonList(arm("a"))));
+ }
+
+ @Test
+ public void testNullDisjunctListRejected() {
+ Assertions.assertThrows(NullPointerException.class, () -> new ConnectorOr(null));
+ }
+
+ @Test
+ public void testTwoOrMoreDisjunctsAccepted() {
+ Assertions.assertEquals(2, new ConnectorOr(Arrays.asList(arm("a"), arm("b")))
+ .getDisjuncts().size());
+ Assertions.assertEquals(3, new ConnectorOr(Arrays.asList(arm("a"), arm("b"), arm("c")))
+ .getDisjuncts().size());
+ }
+
+ @Test
+ public void testCallerListMutationDoesNotChangeNode() {
+ // The node used to wrap the caller's list as an unmodifiable VIEW, so a caller that kept
+ // building its own list afterwards would mutate an already-constructed predicate node.
+ // ConnectorIn in this same package copies defensively; this pins the two to one behavior.
+ List WHY the literal is frozen: a partition name is persisted, user-visible identity. It is baked into view and
+ * materialized-view definitions (e.g. a view filtering {@code t_int != "__HIVE_DEFAULT_PARTITION__"}), into
+ * {@code partition_values()} table-function output, and into the {@code columns_from_path} bytes BE parses; BE
+ * also hardcodes the same literal on the hive write path ({@code vhive_utils.cpp}). Changing the string orphans
+ * every object already persisted with it. Renaming the SYMBOL is free; changing the VALUE is not. Why this matters: these defaults are what keep the change zero-break for the other
* connectors (es/jdbc/hive/paimon/hudi/trino). {@code supportsBatchScan} MUST default to false so no
- * connector silently enters batch mode without opting in; {@code planScanForPartitionBatch} MUST
- * delegate to the 6-arg {@code planScan} with the batch as the required partitions (and forward the
- * limit), so a connector whose {@code planScan} is partition-set-scoped — like MaxCompute — gets
+ * connector silently enters batch mode without opting in; {@code planScanForPartitionBatch} MUST call
+ * {@code planScan} with the batch as the required partitions AND everything else about the request
+ * unchanged, so a connector whose {@code planScan} is partition-set-scoped — like MaxCompute — gets
* correct per-batch behaviour without overriding it. Why this matters: the two behaviours the javadoc promises are both silent when wrong. A
+ * connector that overrides only the {@code Map} face must still reach the engine (otherwise its scan
+ * properties vanish and, for instance, its storage credentials never reach BE); and the delegation must
+ * NOT claim conjunct tracking, because "tracking with an empty not-pushed set" means "every conjunct was
+ * pushed exactly, prune them all" — asserting that for a connector which never reported anything would
+ * drop filters and return extra rows. WHY this matters: the generic translator asks every write-capable connector for its
- * write-sort columns. The default MUST be an empty list so connectors that declare no write sort
- * (jdbc / maxcompute) produce no {@code TSortInfo} and keep their byte-identical unsorted sink output —
- * only iceberg (with a {@code WRITE ORDERED BY}) overrides it. This is a conscious acceptance of the risk the SPI documents, not a claim of safety: the residual
+ * predicate is compiled into the ES query DSL ({@code EsScanPlanProvider.buildQueryDsl}) and evaluated by
+ * Elasticsearch, so a comparison whose literal ES matches differently than Doris coerced it drops rows AT
+ * THE SOURCE. It stays {@code true} for parity with the legacy {@code EsScanNode}, which built the same
+ * DSL; unconvertible conjuncts are already reported back as not-pushed and re-evaluated by BE. Every other foreign-table operation is diverted to the embedded iceberg sibling by the concrete handle
+ * type, but this method addresses a table by NAME and never receives a handle, so there is nothing to
+ * discriminate on. It is answered directly instead: iceberg's HMS catalog mirrors the table's iceberg
+ * properties into the HMS table parameters on every commit, {@code comment} included — the same
+ * parameter {@code HiveShowCreateTableRenderer} lifts into the {@code COMMENT} clause — so the single
+ * (cached) {@code getTable} this connector already performs carries it, with no sibling build and no
+ * iceberg metadata load. Without this, the same table shows its comment through a dedicated iceberg catalog
+ * and shows nothing through an HMS gateway catalog, in {@code SHOW CREATE TABLE},
+ * {@code information_schema.tables.TABLE_COMMENT} and {@code SHOW TABLE STATUS}. Deliberately restricted to iceberg tables: a plain hive table keeps the empty comment that legacy
+ * {@code HMSExternalTable.getComment} returned, so nothing about plain hive changes. A missing or
+ * unreadable table degrades to the empty comment rather than failing the caller, matching
+ * {@link #viewExists} — this is an opportunistic display value, not a user-requested operation. fe-core couplings broken per design: query profiling dropped (D4); metastore access via the plugin
* {@link HmsClient} SPI instead of {@code HiveMetadataOps} (D10); staging renames/deletes and object-store
* multipart-upload complete/abort go through the engine-owned {@link FileSystem} borrowed via
- * {@code context.getFileSystem(session)} (a per-catalog {@code SpiSwitchingFileSystem}, all schemes), with the
+ * {@code storage().getFileSystem(session)} (a per-catalog {@code SpiSwitchingFileSystem}, all schemes), with the
* concrete {@link ObjFileSystem} resolved per object-store location via {@link FileSystem#forLocation} for the
* MPU narrowing (D6); plugin-owned async pool threads each auth-wrapped (D5); full-ACID writes hard-rejected at
* begin (D7); {@code rollback()} deletes staging + aborts MPUs (D9).
*
- * Live since the HMS cutover: {@code hms} is in {@code SPI_READY_TYPES}, so a {@code type=hms} table INSERT
+ * Live since the HMS cutover: an {@code hms} catalog is served by this connector plugin, so a
+ * {@code type=hms} table INSERT
* routes through {@code HiveWritePlanProvider.planWrite} → {@link #beginWrite} → this class. The engine
* {@link FileSystem} is borrowed (never closed here — the catalog owns its lifecycle).
*/
@@ -114,7 +116,7 @@ public class HiveConnectorTransaction implements ConnectorTransaction {
private final ExecutorService fileSystemExecutor;
private final Executor authWrappingExecutor;
- // Captured at beginWrite (D6). Threaded to context.getFileSystem(session) so the engine hands back the
+ // Captured at beginWrite (D6). Threaded to storage().getFileSystem(session) so the engine hands back the
// per-catalog borrowed FileSystem; the session reserves per-user identity (getUser()) — the current engine
// impl resolves the FS at catalog level and ignores it, so a null session (rollback-before-begin) is safe.
private ConnectorSession session;
@@ -747,13 +749,13 @@ private synchronized void dropPartition(
/**
* Returns the engine-owned {@link FileSystem} for this write, borrowed via
- * {@code context.getFileSystem(session)} (a per-catalog {@code SpiSwitchingFileSystem} that routes every
+ * {@code storage().getFileSystem(session)} (a per-catalog {@code SpiSwitchingFileSystem} that routes every
* scheme — HDFS and object stores alike). The engine lazily builds and caches it per catalog, so this is a
* cheap lookup; the connector borrows and must never close it (D6). MPU sites narrow to the concrete
* {@link ObjFileSystem} via {@link FileSystem#forLocation}.
*/
private FileSystem getFileSystem() {
- FileSystem engineFs = context.getFileSystem(session);
+ FileSystem engineFs = storage().getFileSystem(session);
if (engineFs == null) {
throw new DorisConnectorException("No engine FileSystem available for hive write transaction "
+ transactionId + " (catalog has no storage properties)");
@@ -1691,4 +1693,9 @@ THivePartitionUpdate getHivePartitionUpdate() {
return hivePartitionUpdate;
}
}
+
+ /** This catalog's engine-owned storage services (see {@link ConnectorContext#getStorageContext()}). */
+ private ConnectorStorageContext storage() {
+ return context.getStorageContext();
+ }
}
diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveFileListingCache.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveFileListingCache.java
index 6c9d0efcb54960..a62967c33760f2 100644
--- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveFileListingCache.java
+++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveFileListingCache.java
@@ -102,7 +102,8 @@ public class HiveFileListingCache {
* The raw directory lister: the engine-injected Doris {@link FileSystem} in production
* ({@link #listFromFileSystem}), a fake in unit tests. Injected so the cache's hit/miss/invalidation behaviour
* is testable without a live filesystem (mirrors {@code HiveConnectorMetadata.estimateDataSize} injecting its
- * {@code ToLongFunction} size source). The {@code fs} is borrowed from {@code ConnectorContext.getFileSystem}
+ * {@code ToLongFunction} size source). The {@code fs} is borrowed from the engine's
+ * {@code ConnectorStorageContext.getFileSystem}
* (engine-owned, per-catalog, must NOT be closed by the connector).
*/
@FunctionalInterface
diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
index 4da1b40adbd9b3..fe6aa3f0b10e55 100644
--- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
+++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
@@ -24,10 +24,12 @@
import org.apache.doris.connector.api.pushdown.ConnectorExpression;
import org.apache.doris.connector.api.scan.ConnectorScanPlanProvider;
import org.apache.doris.connector.api.scan.ConnectorScanRange;
-import org.apache.doris.connector.api.scan.ConnectorScanRangeType;
+import org.apache.doris.connector.api.scan.ConnectorScanRequest;
+import org.apache.doris.connector.api.scan.ScanNodePropertyKeys;
import org.apache.doris.connector.hms.HmsClient;
import org.apache.doris.connector.hms.HmsPartitionInfo;
import org.apache.doris.connector.spi.ConnectorContext;
+import org.apache.doris.connector.spi.ConnectorStorageContext;
import org.apache.doris.filesystem.FileEntry;
import org.apache.doris.filesystem.FileSystem;
import org.apache.doris.thrift.TFileCompressType;
@@ -78,10 +80,6 @@ public class HiveScanPlanProvider implements ConnectorScanPlanProvider {
/** Maximum number of partitions to list from HMS. */
private static final int MAX_PARTITIONS = 100000;
- /** Scan node property keys. */
- public static final String PROP_FILE_FORMAT_TYPE = "file_format_type";
- public static final String PROP_PATH_PARTITION_KEYS = "path_partition_keys";
- public static final String PROP_LOCATION_PREFIX = "location.";
/**
* Connector-internal signal (consumed by {@link #populateScanLevelParams}) marking a full-ACID
* (transactional) Hive scan; drives the scan-level {@code table_format_type} stamp. Not a BE-facing key.
@@ -94,7 +92,7 @@ public class HiveScanPlanProvider implements ConnectorScanPlanProvider {
private final HmsClient hmsClient;
private final Map Live path. Post-cutover {@code hms} is in {@code SPI_READY_TYPES}, so an hms-catalog
+ * Live path. Post-cutover an {@code hms} catalog is plugin-driven, so an hms-catalog
* transactional table is a {@code PluginDrivenExternalTable} routed through {@code PluginDrivenScanNode}
* straight into this method on a live query (the only read gate is the full-ACID ORC-format check below).
* It opens a real metastore transaction/lock; the matching commit (lock release) is driven by
@@ -386,12 +378,12 @@ public Map These properties are passed to BE via scan node properties so that
* the text/CSV/JSON file reader can parse the data correctly. The property keys mirror the constants used in fe-core's
- * {@code HiveProperties} and {@code HiveMetaStoreClientHelper}. The output keys are the text-family keys declared in {@link ScanNodePropertyKeys} — they are shared
+ * with the engine rather than mirrored, so a mistyped suffix (which would silently disable that one
+ * attribute, e.g. leaving the whole row in the first column) is a compile error. Live since the hms flip. {@code hms} is in {@code SPI_READY_TYPES}, so a type=hms INSERT routes
+ * Live since the hms flip. An {@code hms} catalog is served by this connector plugin, so a
+ * type=hms INSERT routes
* here via {@code PhysicalPlanTranslator} → {@code getWritePlanProvider} → {@link #planWrite};
* {@link #planWrite} requires the executor-bound connector transaction and fails loud if absent. WHY (Rule 9): hive is the ONLY connector that declares {@code requiresPartitionHashWrite}, and its tests
+ * did not call the validator — so invariants #4 (hash write implies parallel write AND full-schema write
+ * order) and #5 (the two partition-distribution arms are mutually exclusive) were exercised solely by
+ * fe-core's fake connector. A fake proves the validator rejects a bad declaration; only a real connector
+ * proves the shipped declaration is a good one. Dropping any of the three traits hive declares, or adding
+ * {@code requiresPartitionLocalSort} alongside the hash arm, must fail loud here. The provider is built directly ({@code HiveWritePlanProvider}'s constructor is pure field assignment)
+ * with a null client: the validator reads only the declared traits, none of which touch the metastore. The
+ * no-arg {@link HiveConnector#getWritePlanProvider()} is not used because it builds a real HmsClient, whose
+ * Hadoop stack is absent from connector unit tests. WHY this matters: the comment accessor takes a database and table NAME, never a
+ * {@code ConnectorTableHandle}, so the gateway cannot route it to the iceberg sibling the way it routes every
+ * other foreign-table operation — there is no handle to discriminate on. Before this was handled explicitly,
+ * an iceberg table showed its comment when read through a dedicated iceberg catalog and showed nothing when
+ * read through an HMS gateway catalog: a blank {@code COMMENT} clause in SHOW CREATE TABLE, a blank
+ * {@code information_schema.tables.TABLE_COMMENT} and a blank SHOW TABLE STATUS Comment column, for the same
+ * table. Iceberg's HMS catalog mirrors the comment into the HMS table parameters on every commit, so the
+ * gateway answers from the metastore table it already reads. The plain-hive assertions are the other half of the contract: a plain hive table must keep the empty
+ * comment legacy returned, so this change cannot alter what any existing hive deployment displays.