PaimonExternalMetaCache.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.paimon;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.datasource.CacheException;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.ExternalCatalog;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.NameMapping;
import org.apache.doris.datasource.SchemaCacheValue;
import org.apache.doris.datasource.metacache.AbstractExternalMetaCache;
import org.apache.doris.datasource.metacache.CacheSpec;
import org.apache.doris.datasource.metacache.MetaCacheEntry;
import org.apache.doris.datasource.metacache.MetaCacheEntryDef;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.DataTable;
import org.apache.paimon.table.Table;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
/**
* Paimon engine implementation of {@link AbstractExternalMetaCache}.
*
* <p>Registered entries:
* <ul>
* <li>{@code table}: loaded Paimon table handle per table mapping</li>
* <li>{@code schema}: schema cache keyed by table identity + schema id</li>
* </ul>
*
* <p>Snapshot-related helpers reuse the table/schema entries to derive partition metadata
* and latest snapshot state without introducing extra cache entries.
*
* <p>Invalidation behavior:
* <ul>
* <li>db/table invalidation clears both table and schema entries by matching local names</li>
* <li>partition-level invalidation falls back to table-level invalidation</li>
* </ul>
*/
public class PaimonExternalMetaCache extends AbstractExternalMetaCache {
public static final String ENGINE = "paimon";
public static final String ENTRY_TABLE = "table";
public static final String ENTRY_SCHEMA = "schema";
private static final CacheSpec SCHEMA_CACHE_SPEC = CacheSpec.fromTtlValue(
null, Config.external_cache_expire_time_seconds_after_access, Config.max_external_schema_cache_num);
private final MetaCacheEntryDef<NameMapping, PaimonTableCacheValue> tableEntryDef;
private final MetaCacheEntryDef<PaimonSchemaCacheKey, SchemaCacheValue> schemaEntryDef;
public PaimonExternalMetaCache(ExecutorService refreshExecutor) {
super(ENGINE, refreshExecutor);
tableEntryDef = MetaCacheEntryDef.of(ENTRY_TABLE, NameMapping.class, PaimonTableCacheValue.class,
this::loadTableCacheValue, DEFAULT_ENTRY_CACHE_SPEC);
schemaEntryDef = MetaCacheEntryDef.of(ENTRY_SCHEMA, PaimonSchemaCacheKey.class, SchemaCacheValue.class,
this::loadSchemaCacheValue, SCHEMA_CACHE_SPEC);
registerMetaCacheEntryDef(tableEntryDef);
registerMetaCacheEntryDef(schemaEntryDef);
}
public Table getPaimonTable(ExternalTable dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
return tableEntry(nameMapping.getCtlId()).get(nameMapping).getPaimonTable();
}
public Table getPaimonTable(NameMapping nameMapping) {
return tableEntry(nameMapping.getCtlId()).get(nameMapping).getPaimonTable();
}
public PaimonSnapshotCacheValue getSnapshotCache(ExternalTable dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
PaimonTableCacheValue tableCacheValue = tableEntry(nameMapping.getCtlId()).get(nameMapping);
return tableCacheValue.getSnapshotCacheValue(
() -> loadSnapshot(dorisTable, tableCacheValue.getPaimonTable()));
}
public PaimonSchemaCacheValue getPaimonSchemaCacheValue(NameMapping nameMapping, long schemaId) {
SchemaCacheValue schemaCacheValue = schemaEntry(nameMapping.getCtlId())
.get(new PaimonSchemaCacheKey(nameMapping, schemaId));
return (PaimonSchemaCacheValue) schemaCacheValue;
}
@Override
public void invalidateDb(long catalogId, String dbName) {
tableEntry(catalogId).invalidateIf(key -> key.getCtlId() == catalogId
&& key.getLocalDbName().equals(dbName));
schemaEntry(catalogId).invalidateIf(key -> key.getNameMapping().getCtlId() == catalogId
&& key.getNameMapping().getLocalDbName().equals(dbName));
}
@Override
public void invalidateTable(long catalogId, String dbName, String tableName) {
tableEntry(catalogId).invalidateIf(key -> key.getCtlId() == catalogId
&& key.getLocalDbName().equals(dbName)
&& key.getLocalTblName().equals(tableName));
schemaEntry(catalogId).invalidateIf(key -> key.getNameMapping().getCtlId() == catalogId
&& key.getNameMapping().getLocalDbName().equals(dbName)
&& key.getNameMapping().getLocalTblName().equals(tableName));
}
@Override
public void invalidatePartitions(long catalogId, String dbName, String tableName, List<String> partitions) {
invalidateTable(catalogId, dbName, tableName);
}
private MetaCacheEntry<NameMapping, PaimonTableCacheValue> tableEntry(long catalogId) {
return entry(catalogId, tableEntryDef);
}
private MetaCacheEntry<PaimonSchemaCacheKey, SchemaCacheValue> schemaEntry(long catalogId) {
return entry(catalogId, schemaEntryDef);
}
private PaimonTableCacheValue loadTableCacheValue(NameMapping nameMapping) {
try {
PaimonExternalCatalog externalCatalog = (PaimonExternalCatalog) Env.getCurrentEnv().getCatalogMgr()
.getCatalogOrException(nameMapping.getCtlId(), id -> new IOException("Catalog not found: " + id));
Table table = externalCatalog.getPaimonTable(nameMapping);
return new PaimonTableCacheValue(table);
} catch (Exception e) {
throw new CacheException("failed to load paimon table %s.%s.%s: %s",
e, nameMapping.getCtlId(), nameMapping.getLocalDbName(), nameMapping.getLocalTblName(),
e.getMessage());
}
}
private PaimonSnapshotCacheValue loadSnapshot(ExternalTable dorisTable, Table paimonTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
try {
PaimonSnapshot latestSnapshot = loadLatestSnapshot(paimonTable);
List<Column> partitionColumns = getPaimonSchemaCacheValue(nameMapping,
latestSnapshot.getSchemaId()).getPartitionColumns();
PaimonPartitionInfo partitionInfo = loadPartitionInfo(nameMapping, partitionColumns);
return new PaimonSnapshotCacheValue(partitionInfo, latestSnapshot);
} catch (Exception e) {
throw new CacheException("failed to load paimon snapshot %s.%s.%s: %s",
e, nameMapping.getCtlId(), nameMapping.getLocalDbName(), nameMapping.getLocalTblName(),
e.getMessage());
}
}
private SchemaCacheValue loadSchemaCacheValue(PaimonSchemaCacheKey key) {
ExternalTable dorisTable = findExternalTable(key.getNameMapping());
return dorisTable.initSchemaAndUpdateTime(key).orElseThrow(() ->
new CacheException("failed to load paimon schema cache value for: %s.%s.%s, schemaId: %s",
null, key.getNameMapping().getCtlId(), key.getNameMapping().getLocalDbName(),
key.getNameMapping().getLocalTblName(), key.getSchemaId()));
}
private PaimonPartitionInfo loadPartitionInfo(NameMapping nameMapping, List<Column> partitionColumns)
throws AnalysisException {
if (CollectionUtils.isEmpty(partitionColumns)) {
return PaimonPartitionInfo.EMPTY;
}
PaimonExternalCatalog externalCatalog = (PaimonExternalCatalog) Env.getCurrentEnv().getCatalogMgr()
.getCatalogOrAnalysisException(nameMapping.getCtlId());
List<Partition> paimonPartitions = externalCatalog.getPaimonPartitions(nameMapping);
return PaimonUtil.generatePartitionInfo(partitionColumns, paimonPartitions);
}
private PaimonSnapshot loadLatestSnapshot(Table paimonTable) {
Table snapshotTable = paimonTable;
long latestSnapshotId = PaimonSnapshot.INVALID_SNAPSHOT_ID;
Optional<Snapshot> optionalSnapshot = paimonTable.latestSnapshot();
if (optionalSnapshot.isPresent()) {
latestSnapshotId = optionalSnapshot.get().id();
snapshotTable = paimonTable.copy(
Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), String.valueOf(latestSnapshotId)));
}
DataTable dataTable = (DataTable) paimonTable;
long latestSchemaId = dataTable.schemaManager().latest().map(TableSchema::id).orElse(0L);
return new PaimonSnapshot(latestSnapshotId, latestSchemaId, snapshotTable);
}
@Override
protected Map<String, String> catalogPropertyCompatibilityMap() {
return Collections.singletonMap(
ExternalCatalog.SCHEMA_CACHE_TTL_SECOND,
"meta.cache." + ENGINE + "." + ENTRY_SCHEMA + ".ttl-second");
}
private ExternalTable findExternalTable(NameMapping nameMapping) {
CatalogIf<?> catalog = Env.getCurrentEnv().getCatalogMgr().getCatalog(nameMapping.getCtlId());
if (!(catalog instanceof ExternalCatalog)) {
throw new CacheException("catalog %s is not external when loading paimon schema cache",
null, nameMapping.getCtlId());
}
ExternalCatalog externalCatalog = (ExternalCatalog) catalog;
return externalCatalog.getDb(nameMapping.getLocalDbName())
.flatMap(db -> db.getTable(nameMapping.getLocalTblName()))
.orElseThrow(() -> new CacheException(
"table %s.%s.%s not found when loading paimon schema cache",
null, nameMapping.getCtlId(), nameMapping.getLocalDbName(),
nameMapping.getLocalTblName()));
}
}