Repository navigation
[improvement](datasource) Version Iceberg partition views and SQL cache metadata #69299
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -25,6 +25,7 @@ | |
| import org.apache.doris.connector.spi.ConnectorMetadata; | ||
| import org.apache.doris.connector.spi.ConnectorPartitionInfo; | ||
| import org.apache.doris.connector.spi.ConnectorSession; | ||
| import org.apache.doris.connector.spi.ConnectorStatementScope; | ||
| import org.apache.doris.connector.spi.ConnectorStorageContext; | ||
| import org.apache.doris.connector.spi.ConnectorTableSchema; | ||
| import org.apache.doris.connector.spi.ConnectorTableStatistics; | ||
|
|
@@ -45,6 +46,7 @@ | |
| import org.apache.doris.connector.spi.handle.WriteOperation; | ||
| import org.apache.doris.connector.spi.mvcc.ConnectorMvccPartitionView; | ||
| import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot; | ||
| import org.apache.doris.connector.spi.mvcc.ConnectorSqlCacheVersion; | ||
| import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec; | ||
| import org.apache.doris.connector.spi.pushdown.ConnectorAnd; | ||
| import org.apache.doris.connector.spi.pushdown.ConnectorComparison; | ||
|
|
@@ -79,6 +81,7 @@ | |
| import org.apache.iceberg.SnapshotSummary; | ||
| import org.apache.iceberg.SortOrder; | ||
| import org.apache.iceberg.Table; | ||
| import org.apache.iceberg.TableMetadata; | ||
| import org.apache.iceberg.TableOperations; | ||
| import org.apache.iceberg.TableProperties; | ||
| import org.apache.iceberg.TableScan; | ||
|
|
@@ -342,6 +345,17 @@ public List<String> listTableNames(ConnectorSession session, String dbName) { | |
| } | ||
| } | ||
|
|
||
| @Override | ||
| public List<String> listRelationNames(ConnectorSession session, String dbName) { | ||
| try { | ||
| return executeAuthenticated(() -> catalogOps.listRelationNames(dbName)); | ||
| } catch (RuntimeException e) { | ||
| throw e; | ||
| } catch (Exception e) { | ||
| throw new RuntimeException("Failed to list relation names, error message is: " + e.getMessage(), e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public List<String> listViewNames(ConnectorSession session, String dbName) { | ||
| // Mirror legacy IcebergMetadataOps.listViewNames: wrap in the auth context; a RuntimeException | ||
|
|
@@ -455,26 +469,59 @@ private String loadTableComment(ConnectorSession session, String dbName, String | |
| @Override | ||
| public Optional<ConnectorTableHandle> getTableHandle( | ||
| ConnectorSession session, String dbName, String tableName) { | ||
| IcebergTableHandle handle = new IcebergTableHandle(dbName, tableName); | ||
| if (IcebergStatementScope.hasReadTable(session, dbName, tableName) | ||
| || (tableCache != null && tableCache.contains(TableIdentifier.of(dbName, tableName)))) { | ||
|
924060929 marked this conversation as resolved.
|
||
| return Optional.of(handle); | ||
| } | ||
| // Offline callers cannot retain a cache lease. Pure existence checks use tableExists instead. | ||
| if (session == null || session.getStatementScope() == ConnectorStatementScope.NONE) { | ||
| return tableExists(session, dbName, tableName) ? Optional.of(handle) : Optional.empty(); | ||
| } | ||
| try { | ||
| // Loading proves existence and pins the same metadata used by schema/snapshot/scan planning. | ||
| // Borrow through the existing generation-aware cache so concurrent cold readers share one load. | ||
| executeAuthenticated(() -> resolveTableForRead(session, handle)); | ||
|
Contributor
There was a problem hiding this comment. [P2] Keep DROP handle lookup independent of scan planning For an existing REST Iceberg table whose load response sets
Contributor
Author
There was a problem hiding this comment. Fixed in 3bde61a. DROP retains its live existence preflight and now requests a DDL handle instead of the scan-planning read handle. Iceberg returns an identifier-only DDL handle after checking live existence, so server-side REST scan planning cannot block DROP or ALTER. Query reads still use the frozen scan-capable handle path. Added a server-planned-table test that reproduces the read-handle rejection and verifies DROP succeeds; the Iceberg (10) and generic DDL routing (90) unit tests pass.
Contributor
There was a problem hiding this comment. [P2] Preserve metadata-only EXECUTE actions on server-planned tables With a cold or disabled table cache,
Contributor
Author
There was a problem hiding this comment. Fixed in 068d3c7. ConnectorExecuteAction now resolves the DDL handle before per-handle procedure dispatch. SINGLE_CALL actions use it directly, so a server-planned REST table is not rejected by the scan-planning read path. DISTRIBUTED rewrites reacquire a scan-capable read handle before file planning. The test uses distinct DDL and scan handles and verifies planRewrite receives the scan handle; all 16 ConnectorExecuteActionTest cases pass. |
||
| return Optional.of(handle); | ||
| } catch (Exception e) { | ||
| if (ExceptionUtils.getThrowableList(e).stream().anyMatch(NoSuchTableException.class::isInstance)) { | ||
| return Optional.empty(); | ||
| } | ||
| throw IcebergExceptionUtils.wrapTableLoadFailure(handle, e, | ||
| "Failed to load table, error message is:"); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<ConnectorTableHandle> getTableHandleForDdl( | ||
| ConnectorSession session, String dbName, String tableName) { | ||
| return tableExists(session, dbName, tableName) | ||
| ? Optional.of(new IcebergTableHandle(dbName, tableName)) : Optional.empty(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean tableExists( | ||
| ConnectorSession session, String dbName, String tableName) { | ||
| // Existence checks are used by DDL and must observe out-of-band drops even while a | ||
| // read-table cache entry or an in-flight reader still owns the previous metadata. | ||
| // Query planning reuses pinned/cached metadata through getTableHandle instead. | ||
| // Mirror legacy IcebergMetadataOps.tableExist: wrap the remote existence check in the auth context | ||
| // without loading and pinning a table merely to answer a DDL existence check. | ||
| boolean exists; | ||
| try { | ||
| exists = executeAuthenticated(() -> catalogOps.tableExists(dbName, tableName)); | ||
| } catch (Exception e) { | ||
| // Preserve Optional.empty even when an auth/catalog layer wraps NoSuchTableException. | ||
| if (ExceptionUtils.getThrowableList(e).stream() | ||
| .anyMatch(NoSuchTableException.class::isInstance)) { | ||
| return false; | ||
| } | ||
| // Existence checks in several catalogs load metadata internally, so normalize them at the shared | ||
| // handle boundary just like explicit table loads. | ||
| throw IcebergExceptionUtils.wrapTableLoadFailure(new IcebergTableHandle(dbName, tableName), e, | ||
| "Failed to check table exist, error message is:"); | ||
| } | ||
| return exists; | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -2423,6 +2470,32 @@ public boolean supportsCastPredicatePushdown(ConnectorSession session) { | |
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<ConnectorSqlCacheVersion> getSqlCacheVersion( | ||
| ConnectorSession session, ConnectorTableHandle handle) { | ||
| IcebergTableHandle iceHandle = (IcebergTableHandle) handle; | ||
| try { | ||
| return executeAuthenticated(() -> { | ||
| Table table = resolveTableForRead(session, iceHandle); | ||
| String version = metadataVersion(table, iceHandle).orElseThrow( | ||
| () -> new UnsupportedOperationException("Iceberg SQL cache requires published table metadata")); | ||
| long updateMillis = iceHandle.getSnapshotId() < 0 ? 0 | ||
| : table.snapshot(iceHandle.getSnapshotId()).timestampMillis(); | ||
| return Optional.of(new ConnectorSqlCacheVersion(version, updateMillis)); | ||
| }); | ||
| } catch (Exception e) { | ||
| throw IcebergExceptionUtils.wrapTableLoadFailure(iceHandle, e, | ||
| "Failed to get iceberg SQL cache version, error message is:"); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public boolean supportsNativePartitionPlanning(ConnectorSession session, ConnectorTableHandle handle) { | ||
| return !((IcebergTableHandle) handle).isSystemTable() | ||
| && (session == null || !Boolean.parseBoolean(session.getSessionProperties() | ||
| .getOrDefault("enable_iceberg_internal_partition_pruning", "false"))); | ||
| } | ||
|
|
||
| // ========== B-2: partition enumeration (MTMV RANGE view + SHOW PARTITIONS) ========== | ||
|
|
||
| /** | ||
|
|
@@ -2527,6 +2600,39 @@ public List<String> listPartitionNames(ConnectorSession session, ConnectorTableH | |
| } | ||
| } | ||
|
|
||
| private static Optional<String> metadataVersion(Table table, IcebergTableHandle handle) { | ||
| // Custom catalog tables and unpublished metadata cannot supply a stable metadata generation. | ||
| if (!(table instanceof BaseTable)) { | ||
| return Optional.empty(); | ||
| } | ||
| TableMetadata metadata = ((BaseTable) table).operations().current(); | ||
| String location = metadata.metadataFileLocation(); | ||
| if (location == null) { | ||
| return Optional.empty(); | ||
| } | ||
| return Optional.of(metadata.uuid() + ":" + handle.getSnapshotId() + ":" | ||
| + handle.getSchemaId() + ":" + location); | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<String> getPartitionViewVersion(ConnectorSession session, ConnectorTableHandle handle) { | ||
| IcebergTableHandle iceHandle = (IcebergTableHandle) handle; | ||
| if (!iceHandle.hasSnapshotPin() || session == null | ||
| || session.getStatementScope() == ConnectorStatementScope.NONE) { | ||
| // Without a statement-frozen table, metadata could change between version lookup and listing. | ||
| return Optional.empty(); | ||
| } | ||
| try { | ||
| return executeAuthenticated(() -> { | ||
| Table table = resolveTableForRead(session, iceHandle); | ||
| return metadataVersion(table, iceHandle); | ||
| }); | ||
| } catch (Exception e) { | ||
| throw IcebergExceptionUtils.wrapTableLoadFailure(iceHandle, e, | ||
| "Failed to resolve iceberg partition view version, error message is:"); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * The physical iceberg partitions of {@code handle} with per-partition value maps, so the generic | ||
| * {@code PluginDrivenExternalTable.getNameToPartitionItems} can populate {@code selectedPartitionNum} | ||
|
|
@@ -2595,7 +2701,7 @@ private List<ConnectorPartitionInfo> listPartitionsUncached( | |
| } | ||
|
|
||
| private List<ConnectorPartitionInfo> listPartitions(Table table, IcebergTableHandle iceHandle) { | ||
| return IcebergPartitionUtils.listPartitions(table, iceHandle.getSnapshotId(), | ||
| TableIdentifier.of(iceHandle.getDbName(), iceHandle.getTableName()), partitionCache); | ||
| } | ||
|
|
||
|
|
@@ -2607,12 +2713,13 @@ private List<ConnectorPartitionInfo> listPartitions(Table table, IcebergTableHan | |
| * snapshot id (or {@code -1} for an empty table — iceberg DOES support MVCC, so it still pins, mirroring | ||
| * paimon), and the LATEST schema id (NOT {@code currentSnapshot().schemaId()} — a schema-only change without | ||
| * a new snapshot advances the schema while the snapshot's id lags; legacy reads {@code table.schema() | ||
| * .schemaId()}). An already frozen statement read takes precedence. With a raw table cache, the pin comes | ||
| * from that same table generation: its eviction and the separate latest-snapshot cache can occur at | ||
| * different times, so consulting both could pair a current table with an older snapshot. Catalogs without | ||
| * a raw table cache use the per-catalog {@link IcebergLatestSnapshotCache}, which keeps a stable pin within | ||
| * its TTL. The pin carries BOTH ids atomically so a schema-only ALTER cannot skew snapshotId vs schemaId. | ||
| * A disabled cache ({@code meta.cache.iceberg.table.enable=false}, {@code ttl-second <= 0} or | ||
| * {@code capacity=0}) reads live every call (the legacy no-cache catalog). | ||
| */ | ||
| @Override | ||
| public Optional<ConnectorMvccSnapshot> beginQuerySnapshot( | ||
|
|
@@ -2623,11 +2730,25 @@ public Optional<ConnectorMvccSnapshot> beginQuerySnapshot( | |
| // per-user loadTable authorization) reads live per-user every call, so authorization runs every time (no | ||
| // stale-authz window); mirrors resolveTableForRead's tableCache null-fallback. A disabled cache (the table | ||
| // entry turned off) is a non-null cache that reads live internally. | ||
| IcebergLatestSnapshotCache.CachedSnapshot pin; | ||
| if (IcebergStatementScope.hasReadTable(session, iceHandle.getDbName(), iceHandle.getTableName())) { | ||
| // Handle/schema resolution may have frozen an older generation before another query's REFRESH. | ||
| // A newer shared pin must not point outside that frozen metadata, nor may this reader publish its | ||
| // old pin over the refreshed cache entry. | ||
| pin = latestSnapshotPin(loadTable(session, iceHandle)); | ||
|
Contributor
There was a problem hiding this comment. [P2] Keep warm snapshot pins aligned after raw-table eviction With SQL caching disabled, a long table-cache TTL and
Contributor
Author
There was a problem hiding this comment. Fixed in 4cedfb0. The latest snapshot pin now comes from the same raw table cache generation whenever that cache exists; the independent latest-snapshot cache remains for catalogs without a raw table cache. Added a capacity-1 A/B eviction regression with an external append: the new test failed on the previous code (warm A returned S1 after cold A saw S2) and all 11 IcebergTableHandleReuseTest cases pass after the fix. Existing in-flight readers still use their frozen statement table. |
||
| } else if (tableCache != null) { | ||
| // The table cache owns this generation. Its capacity eviction must evict the latest pin as well; | ||
| // the independently sized latestSnapshotCache cannot establish that relationship. | ||
| pin = loadLatestSnapshotPin(session, iceHandle); | ||
| } else { | ||
| pin = latestSnapshotCache != null | ||
| ? latestSnapshotCache.getOrLoad(id, () -> loadLatestSnapshotPin(session, iceHandle)) | ||
| : loadLatestSnapshotPin(session, iceHandle); | ||
| } | ||
| // SQL result-cache identity and column binding must describe the same read generation. The engine | ||
| // must not combine this snapshot/schema pin with an independently expiring name-keyed schema cache. | ||
| ConnectorMvccSnapshot.Builder snapshot = ConnectorMvccSnapshot.builder() | ||
| .snapshotId(pin.snapshotId).schemaId(pin.schemaId).retainSchema(true); | ||
| if (pin.snapshotId < 0) { | ||
| snapshot.property(EMPTY_PARTITION_STYLE_PROPERTY, pin.emptyPartitionStyle.name()); | ||
| } | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.