This is an automated email from the ASF dual-hosted git repository.
baiyangtx pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 41bc8cf28 [Improvement]: Introduce TableRuntimeFactory interface
(#3760)
41bc8cf28 is described below
commit 41bc8cf2825d6f6853642653c3346b36515fea21
Author: baiyangtx <[email protected]>
AuthorDate: Mon Sep 8 11:39:53 2025 +0800
[Improvement]: Introduce TableRuntimeFactory interface (#3760)
tableRUntime factory
Co-authored-by: zhangyongxiang.alpha <[email protected]>
---
.../apache/amoro/server/AmoroServiceContainer.java | 7 +-
.../org/apache/amoro/server/AmsServiceMetrics.java | 2 +-
.../server/manager/AbstractPluginManager.java | 31 +++++---
.../apache/amoro/server/manager/MetricManager.java | 2 +-
.../amoro/server/manager/PluginConfiguration.java | 16 +++-
.../server/optimizing/OptimizerGroupMetrics.java | 2 +-
.../persistence/mapper/TableRuntimeMapper.java | 5 +-
.../amoro/server/table/AbstractTableMetrics.java | 2 +-
.../amoro/server/table/DefaultTableRuntime.java | 23 +-----
.../server/table/DefaultTableRuntimeFactory.java | 66 ++++++++++++++++
.../server/table/DefaultTableRuntimeStore.java | 26 +++++++
.../amoro/server/table/DefaultTableService.java | 91 +++++++++++++++-------
.../amoro/server/table/TableOptimizingMetrics.java | 2 +-
.../table/TableOrphanFilesCleaningMetrics.java | 2 +-
...andler.java => TableRuntimeFactoryManager.java} | 15 ++--
.../amoro/server/table/TableRuntimeHandler.java | 7 +-
.../apache/amoro/server/table/TableService.java | 2 +
.../amoro/server/table/TableSummaryMetrics.java | 2 +-
.../org.apache.amoro.table.TableRuntimeFactory | 19 +++++
.../apache/amoro/server/AMSServiceTestBase.java | 15 +++-
.../org/apache/amoro/server/AmsEnvironment.java | 11 +++
.../apache/amoro/server/TestAmsServiceMetrics.java | 2 +-
.../server/manager/TestAbstractPluginManager.java | 26 +++----
.../server/optimizing/TestOptimizingQueue.java | 2 +-
.../table/TestDefaultTableRuntimeHandler.java | 12 ++-
.../table/TestSyncTableOfExternalCatalog.java | 2 +-
.../amoro/server/table/TestTableManagerQuery.java | 2 +-
.../main/java/org/apache/amoro/TableRuntime.java | 8 ++
.../org/apache/amoro}/metrics/MetricRegistry.java | 7 +-
.../apache/amoro/table/TableRuntimeFactory.java | 23 ++++--
.../apache/amoro}/metrics/TestMetricRegistry.java | 8 +-
charts/amoro/templates/_pod.tpl | 7 ++
charts/amoro/templates/amoro-configmap.yaml | 5 ++
dist/src/main/amoro-bin/conf/config.yaml | 6 ++
.../conf/plugins/table-runtime-factories.yaml | 25 ++++++
35 files changed, 360 insertions(+), 123 deletions(-)
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
index ddbf8a1c7..ddad83834 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
@@ -50,6 +50,7 @@ import org.apache.amoro.server.table.DefaultTableManager;
import org.apache.amoro.server.table.DefaultTableService;
import org.apache.amoro.server.table.RuntimeHandlerChain;
import org.apache.amoro.server.table.TableManager;
+import org.apache.amoro.server.table.TableRuntimeFactoryManager;
import org.apache.amoro.server.table.TableService;
import org.apache.amoro.server.terminal.TerminalManager;
import org.apache.amoro.server.utils.ThriftServiceProxy;
@@ -167,7 +168,11 @@ public class AmoroServiceContainer {
}
public void startOptimizingService() throws Exception {
- tableService = new DefaultTableService(serviceConfig, catalogManager);
+ TableRuntimeFactoryManager tableRuntimeFactoryManager = new
TableRuntimeFactoryManager();
+ tableRuntimeFactoryManager.initialize();
+
+ tableService =
+ new DefaultTableService(serviceConfig, catalogManager,
tableRuntimeFactoryManager);
optimizingService =
new DefaultOptimizingService(serviceConfig, catalogManager,
optimizerManager, tableService);
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/AmsServiceMetrics.java
b/amoro-ams/src/main/java/org/apache/amoro/server/AmsServiceMetrics.java
index 6fc26ed03..560e4da77 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/AmsServiceMetrics.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmsServiceMetrics.java
@@ -24,7 +24,7 @@ import org.apache.amoro.metrics.Gauge;
import org.apache.amoro.metrics.Metric;
import org.apache.amoro.metrics.MetricDefine;
import org.apache.amoro.metrics.MetricKey;
-import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap;
import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/manager/AbstractPluginManager.java
b/amoro-ams/src/main/java/org/apache/amoro/server/manager/AbstractPluginManager.java
index 05820b720..e3c70c41f 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/manager/AbstractPluginManager.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/manager/AbstractPluginManager.java
@@ -43,8 +43,8 @@ import java.net.URLClassLoader;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
-import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -52,6 +52,8 @@ import java.util.ServiceLoader;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
/**
* Parent class of all plugin managers, which provide the common method to
help load, install, and
@@ -65,7 +67,7 @@ public abstract class AbstractPluginManager<T extends
ActivePlugin> implements P
private final Map<String, T> installedPlugins = new ConcurrentHashMap<>();
private final Map<String, T> foundedPlugins = new ConcurrentHashMap<>();
- private final Map<String, PluginConfiguration> pluginConfigs =
Maps.newConcurrentMap();
+ private final Map<String, PluginConfiguration> pluginConfigs =
Maps.newLinkedHashMap();
private final String pluginCategory;
private final Class<T> pluginType;
@@ -81,12 +83,14 @@ public abstract class AbstractPluginManager<T extends
ActivePlugin> implements P
/** Initialize the plugin manager, and install all plugins. */
public void initialize() {
List<PluginConfiguration> pluginConfigs = loadPluginConfigurations();
- pluginConfigs.forEach(
- config -> {
- PluginConfiguration exists =
this.pluginConfigs.putIfAbsent(config.getName(), config);
- Preconditions.checkArgument(
- exists == null, "Duplicate plugin name found: %s",
config.getName());
- });
+ pluginConfigs.stream()
+ .sorted(Comparator.comparing(PluginConfiguration::getPriority))
+ .forEach(
+ config -> {
+ PluginConfiguration exists =
this.pluginConfigs.putIfAbsent(config.getName(), config);
+ Preconditions.checkArgument(
+ exists == null, "Duplicate plugin name found: %s",
config.getName());
+ });
foundAvailablePlugins();
for (PluginConfiguration pluginConfig : pluginConfigs) {
@@ -145,7 +149,13 @@ public abstract class AbstractPluginManager<T extends
ActivePlugin> implements P
@Override
public List<T> installedPlugins() {
- return new ArrayList<>(this.installedPlugins.values());
+ return installedPluginsStream().collect(Collectors.toList());
+ }
+
+ protected Stream<T> installedPluginsStream() {
+ return pluginConfigs.values().stream()
+ .filter(p -> installedPlugins.containsKey(p.getName()))
+ .map(p -> installedPlugins.get(p.getName()));
}
/**
@@ -182,8 +192,7 @@ public abstract class AbstractPluginManager<T extends
ActivePlugin> implements P
* @param visitor function to visit all installed plugins.
*/
protected void forEach(Consumer<? super T> visitor) {
- this.installedPlugins
- .values()
+ installedPluginsStream()
.forEach(
plugin -> {
try (ClassLoaderContext ignored = new
ClassLoaderContext(plugin)) {
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/manager/MetricManager.java
b/amoro-ams/src/main/java/org/apache/amoro/server/manager/MetricManager.java
index 13916ae55..d89b13e59 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/manager/MetricManager.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/manager/MetricManager.java
@@ -19,8 +19,8 @@
package org.apache.amoro.server.manager;
import org.apache.amoro.metrics.MetricRegisterListener;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.metrics.MetricReporter;
-import org.apache.amoro.server.metrics.MetricRegistry;
/** Metric plugins manager and registry */
public class MetricManager extends AbstractPluginManager<MetricReporter> {
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/manager/PluginConfiguration.java
b/amoro-ams/src/main/java/org/apache/amoro/server/manager/PluginConfiguration.java
index dafde3340..1c35ae9a0 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/manager/PluginConfiguration.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/manager/PluginConfiguration.java
@@ -32,14 +32,18 @@ public class PluginConfiguration {
private static final String NAME = "name";
private static final String ENABLED = "enabled";
private static final String PROPERTIES = "properties";
+ private static final String PRIORITY = "priority";
private final String name;
private final boolean enabled;
+ private final int priority;
private final Map<String, String> properties;
- public PluginConfiguration(String name, boolean enabled, Map<String, String>
properties) {
+ public PluginConfiguration(
+ String name, boolean enabled, int priority, Map<String, String>
properties) {
this.name = name;
this.enabled = enabled;
this.properties = properties;
+ this.priority = priority;
}
public static PluginConfiguration fromJSONObject(JsonNode configOptions) {
@@ -48,12 +52,16 @@ public class PluginConfiguration {
String name = nameNode.textValue();
boolean enabled = JacksonUtil.getBoolean(configOptions, ENABLED, true);
+ Integer priority = JacksonUtil.getInteger(configOptions, PRIORITY);
+ if (priority == null) {
+ priority = 0;
+ }
Map<String, String> props =
JacksonUtil.getMap(configOptions, PROPERTIES, new
TypeReference<Map<String, String>>() {});
if (props == null) {
props = ImmutableMap.of();
}
- return new PluginConfiguration(name, enabled, props);
+ return new PluginConfiguration(name, enabled, priority, props);
}
/** @return Plugin name. */
@@ -70,4 +78,8 @@ public class PluginConfiguration {
public Map<String, String> getProperties() {
return Collections.unmodifiableMap(properties);
}
+
+ public int getPriority() {
+ return priority;
+ }
}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java
index dfa30a5b1..5b70a4e83 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java
@@ -31,7 +31,7 @@ import org.apache.amoro.metrics.Gauge;
import org.apache.amoro.metrics.Metric;
import org.apache.amoro.metrics.MetricDefine;
import org.apache.amoro.metrics.MetricKey;
-import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.server.resource.OptimizerInstance;
import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap;
import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableRuntimeMapper.java
b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableRuntimeMapper.java
index f9fd54375..d93599057 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableRuntimeMapper.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableRuntimeMapper.java
@@ -41,9 +41,8 @@ public interface TableRuntimeMapper {
+ TABLE_NAME
+ " (table_id, group_name, status_code, table_config, table_summary)
"
+ "VALUES (#{tableId}, #{groupName}, #{statusCode}, "
- + " "
- + "
#{tableConfig,typeHandler=org.apache.amoro.server.persistence.converter.Map2StringConverter},
"
- + "
#{tableSummary,typeHandler=org.apache.amoro.server.persistence.converter.JsonObjectConverter})")
+ +
"#{tableConfig,typeHandler=org.apache.amoro.server.persistence.converter.Map2StringConverter},
"
+ +
"#{tableSummary,typeHandler=org.apache.amoro.server.persistence.converter.JsonObjectConverter})")
int insertRuntime(TableRuntimeMeta meta);
/* ---------- update ---------- */
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/AbstractTableMetrics.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/AbstractTableMetrics.java
index 864c77c01..74e52f0b7 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/AbstractTableMetrics.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/AbstractTableMetrics.java
@@ -22,7 +22,7 @@ import org.apache.amoro.ServerTableIdentifier;
import org.apache.amoro.metrics.Metric;
import org.apache.amoro.metrics.MetricDefine;
import org.apache.amoro.metrics.MetricKey;
-import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap;
import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
index 70f8a7fb0..0fbe4d1ac 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
@@ -26,6 +26,7 @@ import org.apache.amoro.api.BlockableOperation;
import org.apache.amoro.config.OptimizingConfig;
import org.apache.amoro.config.TableConfiguration;
import org.apache.amoro.iceberg.Constants;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.optimizing.OptimizingType;
import org.apache.amoro.optimizing.TableRuntimeOptimizingState;
import org.apache.amoro.optimizing.plan.AbstractOptimizingEvaluator;
@@ -33,7 +34,6 @@ import org.apache.amoro.process.AmoroProcess;
import org.apache.amoro.process.ProcessFactory;
import org.apache.amoro.process.TableProcessState;
import org.apache.amoro.server.AmoroServiceConstants;
-import org.apache.amoro.server.metrics.MetricRegistry;
import org.apache.amoro.server.optimizing.OptimizingProcess;
import org.apache.amoro.server.optimizing.OptimizingStatus;
import org.apache.amoro.server.optimizing.TaskRuntime;
@@ -44,7 +44,6 @@ import org.apache.amoro.server.resource.OptimizerInstance;
import org.apache.amoro.server.table.blocker.TableBlocker;
import org.apache.amoro.server.utils.IcebergTableUtil;
import org.apache.amoro.server.utils.SnowflakeIdGenerator;
-import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
import
org.apache.amoro.shade.zookeeper3.org.apache.curator.shaded.com.google.common.collect.Maps;
import org.apache.amoro.table.BaseTable;
@@ -66,6 +65,7 @@ import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
+/** Default table runtime implementation. */
public class DefaultTableRuntime extends AbstractTableRuntime
implements TableRuntime, SupportsProcessPlugins {
@@ -87,7 +87,6 @@ public class DefaultTableRuntime extends AbstractTableRuntime
public static final List<StateKey<?>> REQUIRED_STATES =
Lists.newArrayList(OPTIMIZING_STATE_KEY, PENDING_INPUT_KEY,
PROCESS_ID_KEY);
- private final TableRuntimeHandler tableHandler;
private final Map<Action, TableProcessContainer> processContainerMap =
Maps.newConcurrentMap();
private final TableOptimizingMetrics optimizingMetrics;
private final TableOrphanFilesCleaningMetrics orphanFilesCleaningMetrics;
@@ -96,16 +95,13 @@ public class DefaultTableRuntime extends
AbstractTableRuntime
private volatile OptimizingProcess optimizingProcess;
private final List<TaskRuntime.TaskQuota> taskQuotas = new
CopyOnWriteArrayList<>();
- public DefaultTableRuntime(TableRuntimeStore store, TableRuntimeHandler
tableHandler) {
+ public DefaultTableRuntime(TableRuntimeStore store) {
super(store);
this.optimizingMetrics =
new TableOptimizingMetrics(store.getTableIdentifier(),
store.getGroupName());
this.orphanFilesCleaningMetrics =
new TableOrphanFilesCleaningMetrics(store.getTableIdentifier());
this.tableSummaryMetrics = new
TableSummaryMetrics(store.getTableIdentifier());
-
- Preconditions.checkNotNull(tableHandler, "TableRuntimeHandler must not be
null.");
- this.tableHandler = tableHandler;
}
public void recover(OptimizingProcess optimizingProcess) {
@@ -116,6 +112,7 @@ public class DefaultTableRuntime extends
AbstractTableRuntime
this.optimizingProcess = optimizingProcess;
}
+ @Override
public void registerMetric(MetricRegistry metricRegistry) {
// TODO: extract method to interface.
this.optimizingMetrics.register(metricRegistry);
@@ -279,7 +276,6 @@ public class DefaultTableRuntime extends
AbstractTableRuntime
summary.setTotalFileCount(pendingFileCount);
})
.commit();
- tableHandler.handleTableChanged(this, OptimizingStatus.IDLE);
}
public void setTableSummary(AbstractOptimizingEvaluator.PendingInput
tableSummary) {
@@ -323,23 +319,17 @@ public class DefaultTableRuntime extends
AbstractTableRuntime
return s;
})
.commit();
-
- if (configChanged) {
- tableHandler.handleTableChanged(this, oldConfiguration);
- }
return this;
}
public void beginPlanning() {
OptimizingStatus originalStatus = getOptimizingStatus();
store().begin().updateStatusCode(code ->
OptimizingStatus.PLANNING.getCode()).commit();
- tableHandler.handleTableChanged(this, originalStatus);
}
public void planFailed() {
OptimizingStatus originalStatus = getOptimizingStatus();
store().begin().updateStatusCode(code ->
OptimizingStatus.PENDING.getCode()).commit();
- tableHandler.handleTableChanged(this, originalStatus);
}
public void beginProcess(OptimizingProcess optimizingProcess) {
@@ -354,8 +344,6 @@ public class DefaultTableRuntime extends
AbstractTableRuntime
OptimizingStatus.ofOptimizingType(optimizingProcess.getOptimizingType()).getCode())
.updateState(PENDING_INPUT_KEY, any -> new
AbstractOptimizingEvaluator.PendingInput())
.commit();
-
- tableHandler.handleTableChanged(this, originalStatus);
}
public void completeProcess(boolean success) {
@@ -383,7 +371,6 @@ public class DefaultTableRuntime extends
AbstractTableRuntime
optimizingMetrics.processComplete(processType, success,
optimizingProcess.getPlanTime());
optimizingProcess = null;
- tableHandler.handleTableChanged(this, originalStatus);
}
public void completeEmptyProcess() {
@@ -403,7 +390,6 @@ public class DefaultTableRuntime extends
AbstractTableRuntime
})
.updateState(PENDING_INPUT_KEY, any -> new
AbstractOptimizingEvaluator.PendingInput())
.commit();
- tableHandler.handleTableChanged(this, originalStatus);
}
}
@@ -425,7 +411,6 @@ public class DefaultTableRuntime extends
AbstractTableRuntime
public void beginCommitting() {
OptimizingStatus originalStatus = getOptimizingStatus();
store().begin().updateStatusCode(code ->
OptimizingStatus.COMMITTING.getCode()).commit();
- tableHandler.handleTableChanged(this, originalStatus);
}
@Override
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntimeFactory.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntimeFactory.java
new file mode 100644
index 000000000..86e748cc5
--- /dev/null
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntimeFactory.java
@@ -0,0 +1,66 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.TableRuntime;
+import org.apache.amoro.table.StateKey;
+import org.apache.amoro.table.TableRuntimeFactory;
+import org.apache.amoro.table.TableRuntimeStore;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+public class DefaultTableRuntimeFactory implements TableRuntimeFactory {
+ @Override
+ public void open(Map<String, String> properties) {}
+
+ @Override
+ public void close() {}
+
+ @Override
+ public String name() {
+ return "default";
+ }
+
+ @Override
+ public Optional<TableRuntimeCreator> accept(
+ ServerTableIdentifier tableIdentifier, Map<String, String>
tableProperties) {
+ if (tableIdentifier
+ .getFormat()
+ .in(TableFormat.MIXED_ICEBERG, TableFormat.MIXED_HIVE,
TableFormat.ICEBERG)) {
+ return Optional.of(new TableRuntimeCreatorImpl());
+ }
+ return Optional.empty();
+ }
+
+ private static class TableRuntimeCreatorImpl implements
TableRuntimeFactory.TableRuntimeCreator {
+ @Override
+ public List<StateKey<?>> requiredStateKeys() {
+ return DefaultTableRuntime.REQUIRED_STATES;
+ }
+
+ @Override
+ public TableRuntime create(TableRuntimeStore store) {
+ return new DefaultTableRuntime(store);
+ }
+ }
+}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntimeStore.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntimeStore.java
index 834af645e..3f4d08226 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntimeStore.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntimeStore.java
@@ -19,8 +19,11 @@
package org.apache.amoro.server.table;
import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.TableRuntime;
+import org.apache.amoro.config.TableConfiguration;
import org.apache.amoro.exception.AmoroRuntimeException;
import org.apache.amoro.exception.PersistenceException;
+import org.apache.amoro.server.optimizing.OptimizingStatus;
import org.apache.amoro.server.persistence.NestedSqlSession;
import org.apache.amoro.server.persistence.PersistentBase;
import org.apache.amoro.server.persistence.TableRuntimeMeta;
@@ -46,6 +49,7 @@ import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
+/** Default table runtime store implementation. */
public class DefaultTableRuntimeStore extends PersistentBase implements
TableRuntimeStore {
private static final Logger LOG =
LoggerFactory.getLogger(DefaultTableRuntimeStore.class);
private final Lock tableLock = new ReentrantLock();
@@ -53,6 +57,9 @@ public class DefaultTableRuntimeStore extends PersistentBase
implements TableRun
private final TableRuntimeMeta meta;
private final Map<String, TableRuntimeState> states = new
ConcurrentHashMap<>();
+ private TableRuntimeHandler runtimeHandler;
+ private TableRuntime tableRuntime;
+
public DefaultTableRuntimeStore(
ServerTableIdentifier tableIdentifier,
TableRuntimeMeta meta,
@@ -67,6 +74,14 @@ public class DefaultTableRuntimeStore extends PersistentBase
implements TableRun
restoreStates(requiredStates, restoredStates);
}
+ public void setTableRuntime(TableRuntime tableRuntime) {
+ this.tableRuntime = tableRuntime;
+ }
+
+ public void setRuntimeHandler(TableRuntimeHandler runtimeHandler) {
+ this.runtimeHandler = runtimeHandler;
+ }
+
@Override
public ServerTableIdentifier getTableIdentifier() {
return tableIdentifier;
@@ -163,6 +178,8 @@ public class DefaultTableRuntimeStore extends
PersistentBase implements TableRun
private final Map<String, String> oldStates = new ConcurrentHashMap<>();
private boolean metaOperation = false;
+ private final List<Consumer<TableRuntimeHandler>> handlerCallback =
Lists.newArrayList();
+
public TableRuntimeOperationImpl() {
this.oldMeta = meta.copy();
states.forEach((k, v) -> oldStates.put(k, v.getStateValue()));
@@ -202,6 +219,8 @@ public class DefaultTableRuntimeStore extends
PersistentBase implements TableRun
Integer newStatusCode = updater.apply(oldMeta.getStatusCode());
oldMeta.setStatusCode(newStatusCode);
});
+ OptimizingStatus status =
OptimizingStatus.ofCode(oldMeta.getStatusCode());
+ handlerCallback.add(handler -> handler.handleTableChanged(tableRuntime,
status));
metaOperation = true;
return this;
}
@@ -214,6 +233,9 @@ public class DefaultTableRuntimeStore extends
PersistentBase implements TableRun
updater.accept(tableConfig);
oldMeta.setTableConfig(tableConfig);
});
+ TableConfiguration oldConfiguration =
+ TableConfigurations.parseTableConfig(oldMeta.getTableConfig());
+ handlerCallback.add(handler -> handler.handleTableChanged(tableRuntime,
oldConfiguration));
metaOperation = true;
return this;
}
@@ -277,6 +299,10 @@ public class DefaultTableRuntimeStore extends
PersistentBase implements TableRun
TableRuntimeState state = states.get(key);
state.setStateValue(oldStates.get(key));
});
+
+ if (runtimeHandler != null && tableRuntime != null) {
+ handlerCallback.forEach(c -> c.accept(runtimeHandler));
+ }
}
}
}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
index 97f73894f..a427ccfb9 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
@@ -44,7 +44,6 @@ import
org.apache.amoro.shade.guava32.com.google.common.base.Objects;
import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
import
org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder;
-import org.apache.amoro.table.TableRuntimeStore;
import org.apache.amoro.table.TableSummary;
import org.apache.amoro.utils.TablePropertyUtil;
import org.slf4j.Logger;
@@ -86,14 +85,19 @@ public class DefaultTableService extends PersistentBase
implements TableService
private final CompletableFuture<Boolean> initialized = new
CompletableFuture<>();
private final Configurations serverConfiguration;
private final CatalogManager catalogManager;
+ private final TableRuntimeFactoryManager tableRuntimeFactoryManager;
private RuntimeHandlerChain headHandler;
private ExecutorService tableExplorerExecutors;
- public DefaultTableService(Configurations configuration, CatalogManager
catalogManager) {
+ public DefaultTableService(
+ Configurations configuration,
+ CatalogManager catalogManager,
+ TableRuntimeFactoryManager tableRuntimeFactoryManager) {
this.catalogManager = catalogManager;
this.externalCatalogRefreshingInterval =
configuration.get(AmoroManagementConf.REFRESH_EXTERNAL_CATALOGS_INTERVAL).toMillis();
this.serverConfiguration = configuration;
+ this.tableRuntimeFactoryManager = tableRuntimeFactoryManager;
}
@Override
@@ -133,16 +137,14 @@ public class DefaultTableService extends PersistentBase
implements TableService
}
@Override
- public void handleTableChanged(
- DefaultTableRuntime tableRuntime, OptimizingStatus originalStatus) {
+ public void handleTableChanged(TableRuntime tableRuntime, OptimizingStatus
originalStatus) {
if (headHandler != null) {
headHandler.fireStatusChanged(tableRuntime, originalStatus);
}
}
@Override
- public void handleTableChanged(
- DefaultTableRuntime tableRuntime, TableConfiguration originalConfig) {
+ public void handleTableChanged(TableRuntime tableRuntime, TableConfiguration
originalConfig) {
if (headHandler != null) {
headHandler.fireConfigChanged(tableRuntime, originalConfig);
}
@@ -170,20 +172,29 @@ public class DefaultTableService extends PersistentBase
implements TableService
}));
List<TableRuntime> tableRuntimes = new
ArrayList<>(tableRuntimeMetaList.size());
- tableRuntimeMetaList.stream()
- .filter(t -> identifierMap.containsKey(t.getTableId()))
- .forEach(
- tableRuntimeMeta -> {
- ServerTableIdentifier identifier =
identifierMap.get(tableRuntimeMeta.getTableId());
- List<TableRuntimeState> states =
statesMap.get(tableRuntimeMeta.getTableId());
- TableRuntimeStore store =
- new DefaultTableRuntimeStore(
- identifier, tableRuntimeMeta,
DefaultTableRuntime.REQUIRED_STATES, states);
- DefaultTableRuntime tableRuntime = new
DefaultTableRuntime(store, this);
- tableRuntimeMap.put(tableRuntimeMeta.getTableId(), tableRuntime);
-
tableRuntime.registerMetric(MetricManager.getInstance().getGlobalRegistry());
- tableRuntimes.add(tableRuntime);
- });
+
+ for (TableRuntimeMeta tableRuntimeMeta : tableRuntimeMetaList) {
+ ServerTableIdentifier identifier =
identifierMap.get(tableRuntimeMeta.getTableId());
+ if (identifier == null) {
+ LOG.warn(
+ "No available table identifier found for table runtime meta id={}",
+ tableRuntimeMeta.getTableId());
+ continue;
+ }
+ List<TableRuntimeState> states =
statesMap.get(tableRuntimeMeta.getTableId());
+ Optional<TableRuntime> tableRuntime =
+ createTableRuntime(identifier, tableRuntimeMeta, states);
+ if (!tableRuntime.isPresent()) {
+ LOG.warn("No available table runtime factory found for table {}",
identifier);
+ continue;
+ }
+ tableRuntime.ifPresent(
+ t -> {
+ t.registerMetric(MetricManager.getInstance().getGlobalRegistry());
+ tableRuntimeMap.put(t.getTableIdentifier().getId(), t);
+ tableRuntimes.add(t);
+ });
+ }
if (headHandler != null) {
headHandler.initialize(tableRuntimes);
@@ -218,7 +229,7 @@ public class DefaultTableService extends PersistentBase
implements TableService
}
@VisibleForTesting
- public void setRuntime(DefaultTableRuntime tableRuntime) {
+ public void setRuntime(TableRuntime tableRuntime) {
checkStarted();
tableRuntimeMap.put(tableRuntime.getTableIdentifier().getId(),
tableRuntime);
}
@@ -481,15 +492,18 @@ public class DefaultTableService extends PersistentBase
implements TableService
meta.setStatusCode(OptimizingStatus.IDLE.getCode());
meta.setGroupName(configuration.getOptimizingConfig().getOptimizerGroup());
meta.setTableSummary(new TableSummary());
+ meta.setGroupName(configuration.getOptimizingConfig().getOptimizerGroup());
+ meta.setTableSummary(new TableSummary());
doAs(TableRuntimeMapper.class, mapper -> mapper.insertRuntime(meta));
- DefaultTableRuntimeStore store =
- new DefaultTableRuntimeStore(
- serverTableIdentifier,
- meta,
- DefaultTableRuntime.REQUIRED_STATES,
- Collections.emptyList());
- DefaultTableRuntime tableRuntime = new DefaultTableRuntime(store, this);
+ Optional<TableRuntime> tableRuntimeOpt =
+ createTableRuntime(serverTableIdentifier, meta,
Collections.emptyList());
+ if (!tableRuntimeOpt.isPresent()) {
+ LOG.warn("No available table runtime factory found for table {}",
serverTableIdentifier);
+ return false;
+ }
+
+ TableRuntime tableRuntime = tableRuntimeOpt.get();
tableRuntimeMap.put(serverTableIdentifier.getId(), tableRuntime);
tableRuntime.registerMetric(MetricManager.getInstance().getGlobalRegistry());
if (headHandler != null) {
@@ -498,6 +512,27 @@ public class DefaultTableService extends PersistentBase
implements TableService
return true;
}
+ private Optional<TableRuntime> createTableRuntime(
+ ServerTableIdentifier identifier,
+ TableRuntimeMeta runtimeMeta,
+ List<TableRuntimeState> restoredStates) {
+ return tableRuntimeFactoryManager.installedPlugins().stream()
+ .map(f -> f.accept(identifier, runtimeMeta.getTableConfig()))
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .findFirst()
+ .map(
+ creator -> {
+ DefaultTableRuntimeStore store =
+ new DefaultTableRuntimeStore(
+ identifier, runtimeMeta, creator.requiredStateKeys(),
restoredStates);
+ store.setRuntimeHandler(this);
+ TableRuntime tableRuntime = creator.create(store);
+ store.setTableRuntime(tableRuntime);
+ return tableRuntime;
+ });
+ }
+
private void revertTableRuntimeAdded(
ExternalCatalog externalCatalog, TableIdentity tableIdentity) {
ServerTableIdentifier tableIdentifier =
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableOptimizingMetrics.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableOptimizingMetrics.java
index e92685e07..72014873c 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableOptimizingMetrics.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableOptimizingMetrics.java
@@ -27,9 +27,9 @@ import org.apache.amoro.metrics.Gauge;
import org.apache.amoro.metrics.Metric;
import org.apache.amoro.metrics.MetricDefine;
import org.apache.amoro.metrics.MetricKey;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.optimizing.OptimizingType;
import org.apache.amoro.server.AmoroServiceConstants;
-import org.apache.amoro.server.metrics.MetricRegistry;
import org.apache.amoro.server.optimizing.OptimizingStatus;
import org.apache.amoro.server.optimizing.maintainer.IcebergTableMaintainer;
import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap;
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableOrphanFilesCleaningMetrics.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableOrphanFilesCleaningMetrics.java
index a307eff68..9692b01e7 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableOrphanFilesCleaningMetrics.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableOrphanFilesCleaningMetrics.java
@@ -23,7 +23,7 @@ import static
org.apache.amoro.metrics.MetricDefine.defineCounter;
import org.apache.amoro.ServerTableIdentifier;
import org.apache.amoro.metrics.Counter;
import org.apache.amoro.metrics.MetricDefine;
-import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.metrics.MetricRegistry;
/** Table Orphan Files Cleaning metrics. */
public class TableOrphanFilesCleaningMetrics extends AbstractTableMetrics {
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeHandler.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeFactoryManager.java
similarity index 68%
copy from
amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeHandler.java
copy to
amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeFactoryManager.java
index 45a032a91..65a906108 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeHandler.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeFactoryManager.java
@@ -18,14 +18,13 @@
package org.apache.amoro.server.table;
-import org.apache.amoro.config.TableConfiguration;
-import org.apache.amoro.server.optimizing.OptimizingStatus;
+import org.apache.amoro.server.manager.AbstractPluginManager;
+import org.apache.amoro.table.TableRuntimeFactory;
-public interface TableRuntimeHandler {
+public class TableRuntimeFactoryManager extends
AbstractPluginManager<TableRuntimeFactory> {
+ public static final String PLUGIN_CATEGORY = "table-runtime-factories";
- void addHandlerChain(RuntimeHandlerChain handler);
-
- void handleTableChanged(DefaultTableRuntime tableRuntime, OptimizingStatus
originalStatus);
-
- void handleTableChanged(DefaultTableRuntime tableRuntime, TableConfiguration
originalConfig);
+ public TableRuntimeFactoryManager() {
+ super(PLUGIN_CATEGORY);
+ }
}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeHandler.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeHandler.java
index 45a032a91..922553968 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeHandler.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeHandler.java
@@ -18,14 +18,13 @@
package org.apache.amoro.server.table;
+import org.apache.amoro.TableRuntime;
import org.apache.amoro.config.TableConfiguration;
import org.apache.amoro.server.optimizing.OptimizingStatus;
public interface TableRuntimeHandler {
- void addHandlerChain(RuntimeHandlerChain handler);
+ void handleTableChanged(TableRuntime tableRuntime, OptimizingStatus
originalStatus);
- void handleTableChanged(DefaultTableRuntime tableRuntime, OptimizingStatus
originalStatus);
-
- void handleTableChanged(DefaultTableRuntime tableRuntime, TableConfiguration
originalConfig);
+ void handleTableChanged(TableRuntime tableRuntime, TableConfiguration
originalConfig);
}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableService.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableService.java
index ae711639f..152f95164 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableService.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableService.java
@@ -25,6 +25,8 @@ import org.apache.amoro.server.catalog.InternalCatalog;
public interface TableService extends TableRuntimeHandler {
+ void addHandlerChain(RuntimeHandlerChain handler);
+
void initialize();
void dispose();
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableSummaryMetrics.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableSummaryMetrics.java
index 07b09a40a..e47fdd7bc 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableSummaryMetrics.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableSummaryMetrics.java
@@ -23,8 +23,8 @@ import static
org.apache.amoro.metrics.MetricDefine.defineGauge;
import org.apache.amoro.ServerTableIdentifier;
import org.apache.amoro.metrics.Gauge;
import org.apache.amoro.metrics.MetricDefine;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.optimizing.plan.AbstractOptimizingEvaluator;
-import org.apache.amoro.server.metrics.MetricRegistry;
import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
import org.apache.amoro.table.MixedTable;
import org.apache.amoro.table.UnkeyedTable;
diff --git
a/amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.table.TableRuntimeFactory
b/amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.table.TableRuntimeFactory
new file mode 100644
index 000000000..a3b50433a
--- /dev/null
+++
b/amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.table.TableRuntimeFactory
@@ -0,0 +1,19 @@
+#
+# 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.
+#
+
+org.apache.amoro.server.table.DefaultTableRuntimeFactory
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/AMSServiceTestBase.java
b/amoro-ams/src/test/java/org/apache/amoro/server/AMSServiceTestBase.java
index 15f7009b0..aa0c442ca 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/AMSServiceTestBase.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/AMSServiceTestBase.java
@@ -23,10 +23,14 @@ import org.apache.amoro.resource.ResourceGroup;
import org.apache.amoro.server.manager.EventsManager;
import org.apache.amoro.server.manager.MetricManager;
import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.apache.amoro.server.table.DefaultTableRuntimeFactory;
import org.apache.amoro.server.table.DefaultTableService;
+import org.apache.amoro.server.table.TableRuntimeFactoryManager;
+import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
+import org.mockito.Mockito;
import java.time.Duration;
@@ -34,14 +38,22 @@ public abstract class AMSServiceTestBase extends
AMSManagerTestBase {
private static DefaultTableService TABLE_SERVICE = null;
private static DefaultOptimizingService OPTIMIZING_SERVICE = null;
+ private static TableRuntimeFactoryManager tableRuntimeFactoryManager = null;
+
@BeforeClass
public static void initTableService() {
+ DefaultTableRuntimeFactory runtimeFactory = new
DefaultTableRuntimeFactory();
+ tableRuntimeFactoryManager =
Mockito.mock(TableRuntimeFactoryManager.class);
+ Mockito.when(tableRuntimeFactoryManager.installedPlugins())
+ .thenReturn(Lists.newArrayList(runtimeFactory));
try {
Configurations configurations = new Configurations();
configurations.set(AmoroManagementConf.OPTIMIZER_HB_TIMEOUT,
Duration.ofMillis(800L));
configurations.set(
AmoroManagementConf.OPTIMIZER_TASK_EXECUTE_TIMEOUT,
Duration.ofMillis(30000L));
- TABLE_SERVICE = new DefaultTableService(new Configurations(),
CATALOG_MANAGER);
+ TABLE_SERVICE =
+ new DefaultTableService(
+ new Configurations(), CATALOG_MANAGER,
tableRuntimeFactoryManager);
OPTIMIZING_SERVICE =
new DefaultOptimizingService(
configurations, CATALOG_MANAGER, OPTIMIZER_MANAGER,
TABLE_SERVICE);
@@ -63,6 +75,7 @@ public abstract class AMSServiceTestBase extends
AMSManagerTestBase {
TABLE_SERVICE.dispose();
MetricManager.dispose();
EventsManager.dispose();
+ tableRuntimeFactoryManager = null;
}
protected DefaultTableService tableService() {
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java
b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java
index bf0464453..7ce1cab33 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java
@@ -117,6 +117,10 @@ public class AmsEnvironment {
Objects.requireNonNull(this.getClass().getClassLoader().getResource("")).getPath();
FileUtils.writeStringToFile(
new File(rootPath + "/conf/config.yaml"), getAmsConfig(),
Charset.defaultCharset());
+ FileUtils.writeStringToFile(
+ new File(rootPath + "/conf/plugins/table-runtime-factories.yaml"),
+ getTableRuntimeFactoriesConfig(),
+ Charset.defaultCharset());
System.setProperty(Environments.AMORO_HOME, rootPath);
System.setProperty("derby.init.sql.dir", path + "../classes/sql/derby/");
serviceContainer = new AmoroServiceContainer();
@@ -415,4 +419,11 @@ public class AmsEnvironment {
+ " - name: mocked-reporter\n"
+ " properties:\n";
}
+
+ private String getTableRuntimeFactoriesConfig() {
+ return "table-runtime-factories:\n"
+ + " - name: default\n"
+ + " enabled: true\n"
+ + " priority: 100\n";
+ }
}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestAmsServiceMetrics.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmsServiceMetrics.java
index b62e0d81d..9b2c5436e 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/TestAmsServiceMetrics.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmsServiceMetrics.java
@@ -30,8 +30,8 @@ import static
org.apache.amoro.server.AmsServiceMetrics.AMS_JVM_THREADS_COUNT;
import org.apache.amoro.metrics.Gauge;
import org.apache.amoro.metrics.Metric;
import org.apache.amoro.metrics.MetricKey;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.server.manager.MetricManager;
-import org.apache.amoro.server.metrics.MetricRegistry;
import org.junit.Assert;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestAbstractPluginManager.java
b/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestAbstractPluginManager.java
index 19a4002e8..4994d22b6 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestAbstractPluginManager.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestAbstractPluginManager.java
@@ -99,18 +99,18 @@ public class TestAbstractPluginManager {
return Stream.of(
// enable 1 plugin
Arguments.of(
- ImmutableList.of(new PluginConfiguration("test-A", true,
Maps.newHashMap())), 1),
+ ImmutableList.of(new PluginConfiguration("test-A", true, 0,
Maps.newHashMap())), 1),
// enable 2 plugin
Arguments.of(
ImmutableList.of(
- new PluginConfiguration("test-A", true, Maps.newHashMap()),
- new PluginConfiguration("test-B", true, Maps.newHashMap())),
+ new PluginConfiguration("test-A", true, 0, Maps.newHashMap()),
+ new PluginConfiguration("test-B", true, 0, Maps.newHashMap())),
2),
// provider 2 configs, 1 enabled
Arguments.of(
ImmutableList.of(
- new PluginConfiguration("test-A", false, Maps.newHashMap()),
- new PluginConfiguration("test-B", true, Maps.newHashMap())),
+ new PluginConfiguration("test-A", false, 0, Maps.newHashMap()),
+ new PluginConfiguration("test-B", true, 0, Maps.newHashMap())),
1));
}
@@ -132,9 +132,9 @@ public class TestAbstractPluginManager {
public void testPluginNotInSpiProvider() {
List<PluginConfiguration> configs =
ImmutableList.of(
- new PluginConfiguration("test-A", true, Maps.newHashMap()),
- new PluginConfiguration("test-B", true, Maps.newHashMap()),
- new PluginConfiguration("test-C", true, Maps.newHashMap()));
+ new PluginConfiguration("test-A", true, 0, Maps.newHashMap()),
+ new PluginConfiguration("test-B", true, 0, Maps.newHashMap()),
+ new PluginConfiguration("test-C", true, 0, Maps.newHashMap()));
TestPluginManager manager = new TestPluginManager();
manager.setConfigs(configs);
@@ -145,9 +145,9 @@ public class TestAbstractPluginManager {
public void testPluginDuplicateConfigs() {
List<PluginConfiguration> configs =
ImmutableList.of(
- new PluginConfiguration("test-A", true, Maps.newHashMap()),
- new PluginConfiguration("test-B", true, Maps.newHashMap()),
- new PluginConfiguration("test-B", true, Maps.newHashMap()));
+ new PluginConfiguration("test-A", true, 0, Maps.newHashMap()),
+ new PluginConfiguration("test-B", true, 0, Maps.newHashMap()),
+ new PluginConfiguration("test-B", true, 0, Maps.newHashMap()));
TestPluginManager manager = new TestPluginManager();
manager.setConfigs(configs);
@@ -158,8 +158,8 @@ public class TestAbstractPluginManager {
public void testPluginDuplicateInstall() {
List<PluginConfiguration> configs =
ImmutableList.of(
- new PluginConfiguration("test-A", true, Maps.newHashMap()),
- new PluginConfiguration("test-B", true, Maps.newHashMap()));
+ new PluginConfiguration("test-A", true, 0, Maps.newHashMap()),
+ new PluginConfiguration("test-B", true, 0, Maps.newHashMap()));
TestPluginManager manager = new TestPluginManager();
manager.setConfigs(configs);
manager.initialize();
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
index 46fdee532..91ccf2875 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
@@ -42,12 +42,12 @@ import org.apache.amoro.catalog.CatalogTestHelper;
import org.apache.amoro.io.MixedDataTestHelpers;
import org.apache.amoro.metrics.Gauge;
import org.apache.amoro.metrics.MetricKey;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.optimizing.RewriteFilesOutput;
import org.apache.amoro.optimizing.TableOptimizing;
import org.apache.amoro.process.ProcessStatus;
import org.apache.amoro.resource.ResourceGroup;
import org.apache.amoro.server.manager.MetricManager;
-import org.apache.amoro.server.metrics.MetricRegistry;
import org.apache.amoro.server.resource.OptimizerInstance;
import org.apache.amoro.server.resource.OptimizerThread;
import org.apache.amoro.server.resource.QuotaProvider;
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestDefaultTableRuntimeHandler.java
b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestDefaultTableRuntimeHandler.java
index 71b984e0a..8b6b9c315 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestDefaultTableRuntimeHandler.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestDefaultTableRuntimeHandler.java
@@ -41,6 +41,7 @@ import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import org.mockito.Mockito;
import java.util.List;
@@ -48,6 +49,7 @@ import java.util.List;
public class TestDefaultTableRuntimeHandler extends AMSTableTestBase {
private DefaultTableService tableService;
+ private final TableRuntimeFactoryManager runtimeFactoryManager;
@Parameterized.Parameters(name = "{0}, {1}")
public static Object[] parameters() {
@@ -64,11 +66,16 @@ public class TestDefaultTableRuntimeHandler extends
AMSTableTestBase {
public TestDefaultTableRuntimeHandler(
CatalogTestHelper catalogTestHelper, TableTestHelper tableTestHelper) {
super(catalogTestHelper, tableTestHelper, false);
+ DefaultTableRuntimeFactory runtimeFactory = new
DefaultTableRuntimeFactory();
+ runtimeFactoryManager = Mockito.mock(TableRuntimeFactoryManager.class);
+ Mockito.when(runtimeFactoryManager.installedPlugins())
+ .thenReturn(Lists.newArrayList(runtimeFactory));
}
@Test
public void testInitialize() throws Exception {
- tableService = new DefaultTableService(new Configurations(),
CATALOG_MANAGER);
+ tableService =
+ new DefaultTableService(new Configurations(), CATALOG_MANAGER,
runtimeFactoryManager);
TestHandler handler = new TestHandler();
tableService.addHandlerChain(handler);
tableService.initialize();
@@ -87,7 +94,8 @@ public class TestDefaultTableRuntimeHandler extends
AMSTableTestBase {
Assert.assertTrue(handler.isDisposed());
// initialize with a history table
- tableService = new DefaultTableService(new Configurations(),
CATALOG_MANAGER);
+ tableService =
+ new DefaultTableService(new Configurations(), CATALOG_MANAGER,
runtimeFactoryManager);
handler = new TestHandler();
tableService.addHandlerChain(handler);
tableService.initialize();
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestSyncTableOfExternalCatalog.java
b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestSyncTableOfExternalCatalog.java
index 1d536c8b2..9c0ff5623 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestSyncTableOfExternalCatalog.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestSyncTableOfExternalCatalog.java
@@ -39,11 +39,11 @@ import org.apache.amoro.catalog.CatalogTestHelper;
import org.apache.amoro.catalog.CatalogTestHelpers;
import org.apache.amoro.hive.catalog.HiveCatalogTestHelper;
import org.apache.amoro.hive.catalog.HiveTableTestHelper;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.properties.CatalogMetaProperties;
import org.apache.amoro.server.catalog.ExternalCatalog;
import org.apache.amoro.server.catalog.ServerCatalog;
import org.apache.amoro.server.manager.MetricManager;
-import org.apache.amoro.server.metrics.MetricRegistry;
import org.apache.amoro.server.persistence.PersistentBase;
import org.apache.amoro.server.persistence.TableRuntimeMeta;
import org.apache.amoro.server.persistence.mapper.TableMetaMapper;
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableManagerQuery.java
b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableManagerQuery.java
index 103cbc0ce..d5d5ff741 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableManagerQuery.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableManagerQuery.java
@@ -318,7 +318,7 @@ public class TestTableManagerQuery extends AMSTableTestBase
{
DefaultTableRuntimeStore store =
new DefaultTableRuntimeStore(
identifier, meta, DefaultTableRuntime.REQUIRED_STATES,
Collections.emptyList());
- return new DefaultTableRuntime(store, tableService());
+ return new DefaultTableRuntime(store);
}
}
}
diff --git a/amoro-common/src/main/java/org/apache/amoro/TableRuntime.java
b/amoro-common/src/main/java/org/apache/amoro/TableRuntime.java
index 1f238a27b..097d900f6 100644
--- a/amoro-common/src/main/java/org/apache/amoro/TableRuntime.java
+++ b/amoro-common/src/main/java/org/apache/amoro/TableRuntime.java
@@ -19,6 +19,7 @@
package org.apache.amoro;
import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.process.ProcessFactory;
import org.apache.amoro.process.TableProcessState;
@@ -64,6 +65,13 @@ public interface TableRuntime {
*/
TableConfiguration getTableConfiguration();
+ /**
+ * Register the metric of the table runtime.
+ *
+ * @param metricRegistry the metric registry
+ */
+ void registerMetric(MetricRegistry metricRegistry);
+
/** Get table format */
default TableFormat getFormat() {
return getTableIdentifier().getFormat();
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/metrics/MetricRegistry.java
b/amoro-common/src/main/java/org/apache/amoro/metrics/MetricRegistry.java
similarity index 94%
rename from
amoro-ams/src/main/java/org/apache/amoro/server/metrics/MetricRegistry.java
rename to
amoro-common/src/main/java/org/apache/amoro/metrics/MetricRegistry.java
index 8150e1b34..4845bea0d 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/metrics/MetricRegistry.java
+++ b/amoro-common/src/main/java/org/apache/amoro/metrics/MetricRegistry.java
@@ -16,13 +16,8 @@
* limitations under the License.
*/
-package org.apache.amoro.server.metrics;
+package org.apache.amoro.metrics;
-import org.apache.amoro.metrics.Metric;
-import org.apache.amoro.metrics.MetricDefine;
-import org.apache.amoro.metrics.MetricKey;
-import org.apache.amoro.metrics.MetricRegisterListener;
-import org.apache.amoro.metrics.MetricSet;
import
org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting;
import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeHandler.java
b/amoro-common/src/main/java/org/apache/amoro/table/TableRuntimeFactory.java
similarity index 58%
copy from
amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeHandler.java
copy to
amoro-common/src/main/java/org/apache/amoro/table/TableRuntimeFactory.java
index 45a032a91..7654af710 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntimeHandler.java
+++ b/amoro-common/src/main/java/org/apache/amoro/table/TableRuntimeFactory.java
@@ -16,16 +16,25 @@
* limitations under the License.
*/
-package org.apache.amoro.server.table;
+package org.apache.amoro.table;
-import org.apache.amoro.config.TableConfiguration;
-import org.apache.amoro.server.optimizing.OptimizingStatus;
+import org.apache.amoro.ActivePlugin;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.TableRuntime;
-public interface TableRuntimeHandler {
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
- void addHandlerChain(RuntimeHandlerChain handler);
+/** Table runtime factory. */
+public interface TableRuntimeFactory extends ActivePlugin {
- void handleTableChanged(DefaultTableRuntime tableRuntime, OptimizingStatus
originalStatus);
+ Optional<TableRuntimeCreator> accept(
+ ServerTableIdentifier tableIdentifier, Map<String, String>
tableProperties);
- void handleTableChanged(DefaultTableRuntime tableRuntime, TableConfiguration
originalConfig);
+ interface TableRuntimeCreator {
+ List<StateKey<?>> requiredStateKeys();
+
+ TableRuntime create(TableRuntimeStore store);
+ }
}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/metrics/TestMetricRegistry.java
b/amoro-common/src/test/java/org/apache/amoro/metrics/TestMetricRegistry.java
similarity index 93%
rename from
amoro-ams/src/test/java/org/apache/amoro/server/metrics/TestMetricRegistry.java
rename to
amoro-common/src/test/java/org/apache/amoro/metrics/TestMetricRegistry.java
index ebafdfff9..c0f0c0d22 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/metrics/TestMetricRegistry.java
+++
b/amoro-common/src/test/java/org/apache/amoro/metrics/TestMetricRegistry.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.amoro.server.metrics;
+package org.apache.amoro.metrics;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -25,12 +25,6 @@ import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
-import org.apache.amoro.metrics.Counter;
-import org.apache.amoro.metrics.Gauge;
-import org.apache.amoro.metrics.Metric;
-import org.apache.amoro.metrics.MetricDefine;
-import org.apache.amoro.metrics.MetricKey;
-import org.apache.amoro.metrics.MetricRegisterListener;
import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
import org.junit.jupiter.api.BeforeEach;
diff --git a/charts/amoro/templates/_pod.tpl b/charts/amoro/templates/_pod.tpl
index 80213171b..66d581f97 100644
--- a/charts/amoro/templates/_pod.tpl
+++ b/charts/amoro/templates/_pod.tpl
@@ -73,6 +73,13 @@
readOnly: true
subPath: "metric-reporters.yaml"
{{- end -}}
+{{- /* table-runtime-factories.yaml from config-map*/ -}}
+{{- if or .Values.plugin.metricReporters }}
+- name: conf
+ mountPath: {{ include "amoro.home" .
}}/conf/plugins/table-runtime-factories.yaml
+ readOnly: true
+ subPath: "table-runtime-factories.yaml"
+{{- end -}}
{{- /* flink install dir. if flink optimizer container enabled.
flink distribution package will be installed to here*/ -}}
{{- if .Values.optimizer.flink.enabled }}
diff --git a/charts/amoro/templates/amoro-configmap.yaml
b/charts/amoro/templates/amoro-configmap.yaml
index 6dbb7ec29..5448a6f30 100644
--- a/charts/amoro/templates/amoro-configmap.yaml
+++ b/charts/amoro/templates/amoro-configmap.yaml
@@ -35,6 +35,11 @@ data:
properties:
{{ .prometheusExporter.properties | toYaml | nindent 10 | trim }}
{{- end }}
+ table-runtime-factories.yaml: |
+ table-runtime-factories:
+ - name: default
+ enabled: true
+ priority: 100
## Helm chart provided Amoro configurations
config.yaml: |
ams:
diff --git a/dist/src/main/amoro-bin/conf/config.yaml
b/dist/src/main/amoro-bin/conf/config.yaml
index f628433b7..94ef3cf78 100644
--- a/dist/src/main/amoro-bin/conf/config.yaml
+++ b/dist/src/main/amoro-bin/conf/config.yaml
@@ -63,6 +63,12 @@ ams:
blocker:
timeout: 1min # 60000
+ # config table runtime factories
+ table-runtime-factories:
+ - name: default
+ properties:
+ ~
+
# optional features
expire-snapshots:
enabled: true
diff --git a/dist/src/main/amoro-bin/conf/plugins/table-runtime-factories.yaml
b/dist/src/main/amoro-bin/conf/plugins/table-runtime-factories.yaml
new file mode 100644
index 000000000..34cf7c94d
--- /dev/null
+++ b/dist/src/main/amoro-bin/conf/plugins/table-runtime-factories.yaml
@@ -0,0 +1,25 @@
+#
+# 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.
+#
+
+# This file is config file for plugin metric reporter
+# configurations of metric reporters
+
+table-runtime-factories:
+ - name: default # configs for table runtime factory
+ enabled: true
+ priority: 100
+
