PluginDrivenSysExternalTable.java
// 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.datasource.plugin;
import org.apache.doris.analysis.TableScanParams;
import org.apache.doris.analysis.TableSnapshot;
import org.apache.doris.catalog.Column;
import org.apache.doris.connector.api.Connector;
import org.apache.doris.connector.api.ConnectorMetadata;
import org.apache.doris.connector.api.ConnectorSession;
import org.apache.doris.connector.api.ConnectorTableSchema;
import org.apache.doris.connector.api.handle.ConnectorTableHandle;
import org.apache.doris.connector.api.mvcc.ConnectorMvccSnapshot;
import org.apache.doris.connector.api.scan.ConnectorScanPlanProvider;
import org.apache.doris.datasource.SchemaCacheKey;
import org.apache.doris.datasource.SchemaCacheValue;
import org.apache.doris.datasource.mvcc.MvccSnapshot;
import org.apache.doris.datasource.mvcc.MvccTable;
import org.apache.doris.datasource.mvcc.PluginDrivenMvccSnapshot;
import org.apache.doris.datasource.systable.SysTable;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Predicate;
import java.util.function.Supplier;
/**
* Generic {@link PluginDrivenExternalTable} for a connector system table (e.g. {@code tbl$snapshots}).
*
* <p>Created transiently by {@link org.apache.doris.datasource.systable.PluginDrivenSysTable} during
* planning/describe (via {@code createSysExternalTable}); it is NEVER added to a persisted table map
* and is NOT GSON-registered, mirroring legacy sys ExternalTables (e.g.
* {@code PaimonSysExternalTable}).</p>
*
* <p>It reports {@link org.apache.doris.catalog.TableIf.TableType#PLUGIN_EXTERNAL_TABLE} (inherited);
* no connector-specific table type is introduced. The whole schema/partition/row-count path is reused
* from the base class; the only behavioral change is {@link #resolveConnectorTableHandle}, which threads
* the connector's system-table handle (not the base handle) through every base-class site.</p>
*/
public class PluginDrivenSysExternalTable extends PluginDrivenExternalTable {
private final PluginDrivenExternalTable sourceTable;
private final String sysTableName;
private volatile Optional<SchemaCacheValue> cachedSchemaValue;
/** See {@link #resolveScanPin}: one resolution per selector, shared by binding and scanning. */
private final Map<String, Optional<MvccSnapshot>> scanPinMemo = new ConcurrentHashMap<>();
/**
* @param source the underlying base table being wrapped
* @param sysName the bare system-table name (e.g. "snapshots"), no "$" prefix
*/
public PluginDrivenSysExternalTable(PluginDrivenExternalTable source, String sysName) {
super(generateSysTableId(source.getId(), sysName),
source.getName() + "$" + sysName,
source.getRemoteName() + "$" + sysName,
source.getCatalog(),
source.getDb());
this.sourceTable = source;
this.sysTableName = sysName;
}
/**
* Generate a unique ID from the source table ID and system table name (legacy parity with
* {@code PaimonSysExternalTable.generateSysTableId}).
*/
private static long generateSysTableId(long sourceTableId, String sysName) {
return sourceTableId ^ (sysName.hashCode() * 31L);
}
/**
* Resolve the connector handle for THIS system table: first acquire the BASE table handle using the
* source's remote name (NOT this sys table's "$"-suffixed remote name), then ask the connector for
* the system-table handle. Returning the sys handle here threads it through
* {@code initSchema}/{@code getNameToPartitionItems}/{@code fetchRowCount} automatically, so a sys
* query reads the sys table rather than the base.
*/
@Override
public Optional<ConnectorTableHandle> resolveConnectorTableHandle(
ConnectorSession session, ConnectorMetadata metadata) {
String dbName = db != null ? db.getRemoteName() : "";
Optional<ConnectorTableHandle> baseHandle =
metadata.getTableHandle(session, dbName, sourceTable.getRemoteName());
if (!baseHandle.isPresent()) {
return Optional.empty();
}
return metadata.getSysTableHandle(session, baseHandle.get(), sysTableName);
}
/**
* A system/metadata table (e.g. {@code tbl$snapshots}) is never a view. Short-circuit to {@code false}
* so the base {@code resolveIsView} does not issue a {@code viewExists} round-trip on this synthetic
* {@code "$"}-suffixed name (which would be wasted work and could fail on an unparseable identifier).
*/
@Override
protected boolean resolveIsView() {
return false;
}
/**
* A system/metadata table (e.g. {@code tbl$snapshots}) can NEVER take part in Top-N lazy materialization,
* regardless of what the connector declares. Lazy materialization reads the sort key plus the engine-wide
* row-id ({@code __DORIS_GLOBAL_ROWID_COL__}) first, then re-fetches the surviving rows' other columns by
* row-id ��� which requires a file+position row-id. A system table is served by the connector's JNI
* serialized-split metadata reader, which synthesizes rows from table metadata and produces no such row-id,
* so the injected row-id column comes back empty and BE aborts the scan
* ({@code __DORIS_GLOBAL_ROWID_COL__... return column size 0 not equal to expected size 1}).
*
* <p>This restores legacy parity: legacy sys tables ({@code IcebergSysExternalTable} /
* {@code PaimonSysExternalTable}) extend {@code ExternalTable} ��� NOT the base file-scan table class ��� so
* they are absent from {@code MaterializeProbeVisitor.SUPPORT_RELATION_TYPES} and were never lazy-
* materialized. The base {@link PluginDrivenExternalTable#supportsTopNLazyMaterialize()} keys off the
* connector capability alone and would otherwise (wrongly) admit a flipped sys table; this override is the
* sys-table opt-out. Nested-column prune is similarly opted out for sys tables ��� see
* {@link #supportsNestedColumnPrune()}.
*/
@Override
public boolean supportsTopNLazyMaterialize() {
return false;
}
/**
* A system/metadata table (e.g. {@code tbl$snapshots}) can NEVER take part in nested-column pruning,
* regardless of what the connector declares. Pruning has two stages in fe-core: (L1) generate name-based
* access paths ({@code LogicalFileScan.supportPruneNestedColumn}), then (L2) rewrite each access-path top
* element from the column NAME to its numeric field id ({@code SlotTypeReplacer.replaceAccessPathToFieldId},
* gated on {@link PluginDrivenExternalTable#supportsNestedColumnPrune()}). BE can only field-id-match a
* complex column when the scan ships a field-id dictionary, but a system-table scan intentionally ships NONE
* ({@code IcebergScanPlanProvider} skips {@code SCHEMA_EVOLUTION_PROP} when {@code systemTable}), so
* {@code column->has_identifier_field_id()} is false and BE rejects the field-id access path with
* {@code AccessPathParser access path N does not match slot X}.
*
* <p>Legacy parity: on master the L2 field-id rewrite was gated on {@code instanceof IcebergExternalTable},
* and legacy sys tables ({@code IcebergSysExternalTable}) extend {@code ExternalTable} ��� NOT
* {@code IcebergExternalTable} ��� so L2 never fired for them and their access paths stayed name-based (BE
* matched by name). The migrated gate keys off the connector capability alone, which a flipped sys table
* inherits as {@code true}; this override is the sys-table opt-out. It disables BOTH stages, so no access
* paths are emitted and BE reads the whole (tiny) metadata-table complex column ��� which is correct.
*/
@Override
public boolean supportsNestedColumnPrune() {
return false;
}
/**
* Compute the schema directly on this transient instance instead of going through the base
* {@link ExternalTable#getSchemaCacheValue()}, which routes through {@code ExternalCatalog.getSchema()}
* and re-resolves the table by name in the db map. A system table (e.g. {@code tbl$snapshots}) is never
* registered in that map, so the base path fails with "failed to load schema cache value". Memoized
* (double-checked) to avoid repeated connector round-trips, mirroring legacy
* {@code PaimonSysExternalTable.getSchemaCacheValue}. {@code initSchema()} (inherited from
* {@link PluginDrivenExternalTable}) honors this class's {@link #resolveConnectorTableHandle}, so it
* resolves the system-table schema rather than the base table's.
*/
@Override
public Optional<SchemaCacheValue> getSchemaCacheValue() {
if (cachedSchemaValue == null) {
synchronized (this) {
if (cachedSchemaValue == null) {
cachedSchemaValue = initSchema();
}
}
}
return cachedSchemaValue;
}
@Override
public Optional<SchemaCacheValue> initSchema(SchemaCacheKey key) {
return getSchemaCacheValue();
}
/**
* Delegate to the source table so DESCRIBE/SHOW on a system table still lists its sibling system
* tables (legacy parity with {@code PaimonSysExternalTable.getSupportedSysTables}).
*/
@Override
public Map<String, SysTable> getSupportedSysTables() {
return sourceTable.getSupportedSysTables();
}
@Override
public String getComment() {
return "Plugin system table: " + sysTableName + " for " + sourceTable.getName();
}
public PluginDrivenExternalTable getSourceTable() {
return sourceTable;
}
public String getSysTableName() {
return sysTableName;
}
/**
* This reference's own pin, or empty when it has none / must not have one.
*
* <p>A system table is NOT an {@link MvccTable} and {@code BindRelation} returns from
* {@code handleMetaTable} BEFORE {@code StatementContext.loadSnapshots}, so the statement's MVCC map
* never holds an entry for it and {@code MvccUtil.getSnapshotFromContext} answers empty. The pin lives
* on the SOURCE table, so resolve it from there.
*
* <p><b>Memoized per (tableSnapshot, scanParams) on this instance</b>, which is what keeps binding and
* scanning on ONE resolution. The instance is built per relation by
* {@code PluginDrivenSysTable.createSysExternalTable} and then carried on the {@code LogicalFileScan}
* into the scan node, so every consumer ��� {@code LogicalFileScan.computePluginDrivenOutput},
* {@code PluginDrivenScanNode.resolveSysTableSnapshotPin}, {@code PluginDrivenScanNode.buildColumnHandles}
* ��� shares one entry. Resolving independently per consumer would re-open the very skew this closes:
* a MUTABLE selector ({@code scan.mode=latest}, a wall-clock {@code scan.timestamp-millis}) is resolved
* against the LIVE table on each call, so a commit landing between bind and scan would hand the two
* different versions.
*
* <p>Returns empty ��� i.e. falls back to the latest schema ��� whenever the connector declines this
* selector on this view, so {@code PluginDrivenScanNode.checkSysTableScanConstraints} keeps ownership of
* the user-facing rejection instead of this path failing first with a worse message.
*/
public Optional<MvccSnapshot> resolveScanPin(Optional<TableSnapshot> tableSnapshot,
Optional<TableScanParams> scanParams) {
if (!tableSnapshot.isPresent() && !scanParams.isPresent()) {
return Optional.empty();
}
if (!(sourceTable instanceof MvccTable)) {
return Optional.empty();
}
return scanPinMemo.computeIfAbsent(pinKeyOf(tableSnapshot, scanParams), key -> {
if (!selectorSupported(scanParams)) {
return Optional.empty();
}
return Optional.of(((MvccTable) sourceTable).loadSnapshot(tableSnapshot, scanParams));
});
}
/**
* The full schema of this system table AS OF {@code tableSnapshot}/{@code scanParams}, falling back to
* the latest schema when this reference carries no pin.
*
* <p>Several metadata views derive their columns from the base table ({@code $audit_log} is
* {@code rowkind} plus the base row type; so are {@code $ro} and {@code $binlog}), so their schema moves
* with the selected snapshot. Binding them from the latest schema while the scan reads the pinned one
* made a since-renamed column fail to bind at all and ��� worse ��� bound a since-retyped column silently at
* the wrong type.
*/
public List<Column> getFullSchemaAt(Optional<TableSnapshot> tableSnapshot,
Optional<TableScanParams> scanParams) {
Optional<MvccSnapshot> pin = resolveScanPin(tableSnapshot, scanParams);
if (!pin.isPresent()) {
return getFullSchema();
}
return schemaCacheValueAt(pin.get())
.map(SchemaCacheValue::getSchema)
.orElseGet(this::getFullSchema);
}
/**
* Reads this view's schema with {@code pin} threaded onto the SYS handle. Deliberately does NOT go
* through {@link #getSchemaCacheValue()}: that memo is the LATEST schema and is shared by the
* version-blind callers (DESCRIBE, {@code information_schema}).
*/
private Optional<SchemaCacheValue> schemaCacheValueAt(MvccSnapshot pin) {
if (!(pin instanceof PluginDrivenMvccSnapshot) || !(catalog instanceof PluginDrivenExternalCatalog)) {
return Optional.empty();
}
ConnectorMvccSnapshot connectorSnapshot = ((PluginDrivenMvccSnapshot) pin).getConnectorSnapshot();
if (connectorSnapshot == null) {
return Optional.empty();
}
PluginDrivenExternalCatalog pluginCatalog = (PluginDrivenExternalCatalog) catalog;
Connector connector = pluginCatalog.getConnector();
if (connector == null) {
return Optional.empty();
}
ConnectorSession session = pluginCatalog.buildConnectorSession();
ConnectorMetadata metadata = PluginDrivenMetadata.get(session, connector);
Optional<ConnectorTableHandle> sysHandle = resolveConnectorTableHandle(session, metadata);
if (!sysHandle.isPresent()) {
return Optional.empty();
}
String dbName = db != null ? db.getRemoteName() : "";
ConnectorTableSchema schema =
metadata.getTableSchema(session, sysHandle.get(), connectorSnapshot);
return Optional.of(toSchemaCacheValue(metadata, session, dbName, getRemoteName(), schema));
}
/**
* Whether the connector honors THIS selector on THIS view ��� the exact mirror of
* {@code PluginDrivenScanNode.sysTableSelectorSupported}, so a pin is resolved for precisely the queries
* that guard lets through. A connector with no scan provider contributes no capability and declines.
*
* <p>Package-private + overridable so {@link #resolveScanPin} stays unit-testable without a live
* connector, the same reason {@code PluginDrivenScanNode}'s capability questions are.
*/
boolean selectorSupported(Optional<TableScanParams> scanParams) {
String bareName = sysTableName == null ? "" : sysTableName.toLowerCase(Locale.ROOT);
if (!scanParams.isPresent()) {
return askScanProvider(ConnectorScanPlanProvider::supportsSystemTableTimeTravel);
}
TableScanParams params = scanParams.get();
if (params.incrementalRead()) {
return askScanProvider(p -> p.supportsSystemTableIncrementalRead(bareName));
}
if (params.isOptions()) {
return askScanProvider(p -> p.supportsSystemTableOptions(bareName));
}
return askScanProvider(ConnectorScanPlanProvider::supportsSystemTableTimeTravel);
}
private boolean askScanProvider(Predicate<ConnectorScanPlanProvider> question) {
if (!(catalog instanceof PluginDrivenExternalCatalog)) {
return false;
}
PluginDrivenExternalCatalog pluginCatalog = (PluginDrivenExternalCatalog) catalog;
Connector connector = pluginCatalog.getConnector();
if (connector == null) {
return false;
}
ConnectorSession session = pluginCatalog.buildConnectorSession();
ConnectorMetadata metadata = PluginDrivenMetadata.get(session, connector);
Optional<ConnectorTableHandle> sysHandle = resolveConnectorTableHandle(session, metadata);
if (!sysHandle.isPresent()) {
return false;
}
ConnectorScanPlanProvider scanProvider = connector.getScanPlanProvider(sysHandle.get());
if (scanProvider == null) {
return false;
}
return onPluginClassLoader(scanProvider, () -> question.test(scanProvider));
}
private static <T> T onPluginClassLoader(ConnectorScanPlanProvider provider, Supplier<T> body) {
ClassLoader previous = Thread.currentThread().getContextClassLoader();
try {
Thread.currentThread().setContextClassLoader(provider.getClass().getClassLoader());
return body.get();
} finally {
Thread.currentThread().setContextClassLoader(previous);
}
}
/**
* The memo key for one reference's selector, derived exactly the way
* {@code StatementContext.versionKeyOf} derives its version key ��� field by field.
*
* <p>It must NOT be the selector objects themselves, nor their {@code toString()}:
* {@link TableScanParams} defines neither value equality nor {@code toString()}, so an
* identity-keyed memo would miss on every lookup and silently reintroduce the double resolution
* {@link #resolveScanPin} exists to prevent.
*/
private static String pinKeyOf(Optional<TableSnapshot> tableSnapshot,
Optional<TableScanParams> scanParams) {
StringBuilder key = new StringBuilder();
if (tableSnapshot != null && tableSnapshot.isPresent()) {
TableSnapshot ts = tableSnapshot.get();
key.append("v:").append(ts.getType()).append(':').append(ts.getValue());
}
if (scanParams != null && scanParams.isPresent()) {
TableScanParams sp = scanParams.get();
key.append("p:").append(sp.getParamType()).append(':').append(sp.getMapParams())
.append(':').append(sp.getListParams());
}
return key.toString();
}
}