This is an automated email from the ASF dual-hosted git repository.
jinsongzhou 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 854f8d40c [AMORO-3104] Added table summary metrics (#3109)
854f8d40c is described below
commit 854f8d40c216a0bc4ba706742499d755fc54ce94
Author: Marig_Weizhi <[email protected]>
AuthorDate: Fri Aug 30 14:16:51 2024 +0800
[AMORO-3104] Added table summary metrics (#3109)
* add table summary metrics
* remove pendingInput
* update metrics doc
* add unit test
* update for table summary metrics
* add unit test
* update unit test for table summary metrics
* update unit test for table summary metrics
* update test for table summary metrics
* Update
amoro-ams/amoro-ams-server/src/main/resources/postgres/ams-postgres-init.sql
Co-authored-by: ZhouJinsong <[email protected]>
* Update amoro-ams/amoro-ams-server/src/main/resources/postgres/upgrade.sql
Co-authored-by: ZhouJinsong <[email protected]>
---------
Co-authored-by: ZhouJinsong <[email protected]>
---
.../optimizing/plan/AbstractPartitionPlan.java | 20 ++
.../optimizing/plan/CommonPartitionEvaluator.java | 39 ++++
.../optimizing/plan/OptimizingEvaluator.java | 39 +++-
.../server/optimizing/plan/PartitionEvaluator.java | 12 +
.../server/persistence/mapper/TableMetaMapper.java | 12 +-
.../apache/amoro/server/table/TableRuntime.java | 19 +-
.../amoro/server/table/TableRuntimeMeta.java | 13 ++
.../amoro/server/table/TableSummaryMetrics.java | 241 +++++++++++++++++++++
.../executor/TableRuntimeRefreshExecutor.java | 3 +-
.../src/main/resources/derby/ams-derby-init.sql | 1 +
.../src/main/resources/mysql/ams-mysql-init.sql | 1 +
.../src/main/resources/mysql/upgrade.sql | 3 +
.../main/resources/postgres/ams-postgres-init.sql | 2 +
.../src/main/resources/postgres/upgrade.sql | 4 +
.../optimizing/plan/TestOptimizingEvaluator.java | 6 +-
.../server/table/TestTableSummaryMetrics.java | 223 +++++++++++++++++++
docs/user-guides/metrics.md | 28 ++-
17 files changed, 651 insertions(+), 15 deletions(-)
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/AbstractPartitionPlan.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/AbstractPartitionPlan.java
index a97512b55..5a87cd9e2 100644
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/AbstractPartitionPlan.java
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/AbstractPartitionPlan.java
@@ -191,6 +191,11 @@ public abstract class AbstractPartitionPlan implements
PartitionEvaluator {
return evaluator().getFragmentFileSize();
}
+ @Override
+ public long getFragmentFileRecords() {
+ return evaluator().getFragmentFileRecords();
+ }
+
@Override
public int getSegmentFileCount() {
return evaluator().getSegmentFileCount();
@@ -201,6 +206,11 @@ public abstract class AbstractPartitionPlan implements
PartitionEvaluator {
return evaluator().getSegmentFileSize();
}
+ @Override
+ public long getSegmentFileRecords() {
+ return evaluator().getSegmentFileRecords();
+ }
+
@Override
public int getEqualityDeleteFileCount() {
return evaluator().getEqualityDeleteFileCount();
@@ -211,6 +221,11 @@ public abstract class AbstractPartitionPlan implements
PartitionEvaluator {
return evaluator().getEqualityDeleteFileSize();
}
+ @Override
+ public long getEqualityDeleteFileRecords() {
+ return evaluator().getEqualityDeleteFileRecords();
+ }
+
@Override
public int getPosDeleteFileCount() {
return evaluator().getPosDeleteFileCount();
@@ -221,6 +236,11 @@ public abstract class AbstractPartitionPlan implements
PartitionEvaluator {
return evaluator().getPosDeleteFileSize();
}
+ @Override
+ public long getPosDeleteFileRecords() {
+ return evaluator().getPosDeleteFileRecords();
+ }
+
@Override
public Weight getWeight() {
return evaluator().getWeight();
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/CommonPartitionEvaluator.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/CommonPartitionEvaluator.java
index cded0ca33..ffdb2660c 100644
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/CommonPartitionEvaluator.java
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/CommonPartitionEvaluator.java
@@ -52,23 +52,29 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
// fragment files
protected int fragmentFileCount = 0;
protected long fragmentFileSize = 0;
+ protected long fragmentFileRecords = 0;
// segment files
protected int rewriteSegmentFileCount = 0;
protected long rewriteSegmentFileSize = 0L;
+ protected long rewriteSegmentFileRecords = 0L;
protected int undersizedSegmentFileCount = 0;
protected long undersizedSegmentFileSize = 0;
+ protected long undersizedSegmentFileRecords = 0;
protected int rewritePosSegmentFileCount = 0;
protected int combinePosSegmentFileCount = 0;
protected long rewritePosSegmentFileSize = 0L;
+ protected long rewritePosSegmentFileRecords = 0L;
protected long min1SegmentFileSize = Integer.MAX_VALUE;
protected long min2SegmentFileSize = Integer.MAX_VALUE;
// delete files
protected int equalityDeleteFileCount = 0;
protected long equalityDeleteFileSize = 0L;
+ protected long equalityDeleteFileRecords = 0L;
protected int posDeleteFileCount = 0;
protected long posDeleteFileSize = 0L;
+ protected long posDeleteFileRecords = 0L;
private long cost = -1;
private Boolean necessary = null;
@@ -132,6 +138,7 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
private boolean addFragmentFile(DataFile dataFile, List<ContentFile<?>>
deletes) {
fragmentFileSize += dataFile.fileSizeInBytes();
fragmentFileCount++;
+ fragmentFileRecords += dataFile.recordCount();
for (ContentFile<?> delete : deletes) {
addDelete(delete);
@@ -149,6 +156,7 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
if (fileShouldRewrite(dataFile, deletes)) {
rewriteSegmentFileSize += dataFile.fileSizeInBytes();
rewriteSegmentFileCount++;
+ rewriteSegmentFileRecords += dataFile.recordCount();
return true;
}
@@ -162,6 +170,7 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
undersizedSegmentFileSize += dataFile.fileSizeInBytes();
undersizedSegmentFileCount++;
+ undersizedSegmentFileRecords += dataFile.recordCount();
return true;
}
@@ -169,6 +178,7 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
if (fileShouldRewrite(dataFile, deletes)) {
rewriteSegmentFileSize += dataFile.fileSizeInBytes();
rewriteSegmentFileCount++;
+ rewriteSegmentFileRecords += dataFile.recordCount();
for (ContentFile<?> delete : deletes) {
addDelete(delete);
}
@@ -178,6 +188,7 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
if (segmentShouldRewritePos(dataFile, deletes)) {
rewritePosSegmentFileSize += dataFile.fileSizeInBytes();
rewritePosSegmentFileCount++;
+ rewritePosSegmentFileRecords += dataFile.recordCount();
for (ContentFile<?> delete : deletes) {
addDelete(delete);
}
@@ -241,9 +252,11 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
if (delete.content() == FileContent.POSITION_DELETES) {
posDeleteFileCount++;
posDeleteFileSize += delete.fileSizeInBytes();
+ posDeleteFileRecords += delete.recordCount();
} else {
equalityDeleteFileCount++;
equalityDeleteFileSize += delete.fileSizeInBytes();
+ equalityDeleteFileRecords += delete.recordCount();
}
}
@@ -366,6 +379,11 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
return fragmentFileSize;
}
+ @Override
+ public long getFragmentFileRecords() {
+ return fragmentFileRecords;
+ }
+
@Override
public int getSegmentFileCount() {
return rewriteSegmentFileCount + undersizedSegmentFileCount +
rewritePosSegmentFileCount;
@@ -376,6 +394,11 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
return rewriteSegmentFileSize + undersizedSegmentFileSize +
rewritePosSegmentFileSize;
}
+ @Override
+ public long getSegmentFileRecords() {
+ return rewriteSegmentFileRecords + undersizedSegmentFileRecords +
rewritePosSegmentFileRecords;
+ }
+
@Override
public int getEqualityDeleteFileCount() {
return equalityDeleteFileCount;
@@ -386,6 +409,11 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
return equalityDeleteFileSize;
}
+ @Override
+ public long getEqualityDeleteFileRecords() {
+ return equalityDeleteFileRecords;
+ }
+
@Override
public int getPosDeleteFileCount() {
return posDeleteFileCount;
@@ -396,6 +424,11 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
return posDeleteFileSize;
}
+ @Override
+ public long getPosDeleteFileRecords() {
+ return posDeleteFileRecords;
+ }
+
public static class Weight implements PartitionEvaluator.Weight {
private final long cost;
@@ -423,18 +456,24 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
.add("lastFullOptimizeTime", tableRuntime.getLastFullOptimizingTime())
.add("fragmentFileCount", fragmentFileCount)
.add("fragmentFileSize", fragmentFileSize)
+ .add("fragmentFileRecords", fragmentFileRecords)
.add("rewriteSegmentFileCount", rewriteSegmentFileCount)
.add("rewriteSegmentFileSize", rewriteSegmentFileSize)
+ .add("rewriteSegmentFileRecords", rewriteSegmentFileRecords)
.add("undersizedSegmentFileCount", undersizedSegmentFileCount)
.add("undersizedSegmentFileSize", undersizedSegmentFileSize)
+ .add("undersizedSegmentFileRecords", undersizedSegmentFileRecords)
.add("rewritePosSegmentFileCount", rewritePosSegmentFileCount)
.add("rewritePosSegmentFileSize", rewritePosSegmentFileSize)
+ .add("rewritePosSegmentFileRecords", rewritePosSegmentFileRecords)
.add("min1SegmentFileSize", min1SegmentFileSize)
.add("min2SegmentFileSize", min2SegmentFileSize)
.add("equalityDeleteFileCount", equalityDeleteFileCount)
.add("equalityDeleteFileSize", equalityDeleteFileSize)
+ .add("equalityDeleteFileRecords", equalityDeleteFileRecords)
.add("posDeleteFileCount", posDeleteFileCount)
.add("posDeleteFileSize", posDeleteFileSize)
+ .add("posDeleteFileRecords", posDeleteFileRecords)
.toString();
}
}
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/OptimizingEvaluator.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/OptimizingEvaluator.java
index cadf5ae2a..7ce4c3e2d 100644
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/OptimizingEvaluator.java
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/OptimizingEvaluator.java
@@ -50,6 +50,7 @@ import java.io.UncheckedIOException;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
public class OptimizingEvaluator {
@@ -60,6 +61,7 @@ public class OptimizingEvaluator {
protected final TableSnapshot currentSnapshot;
protected boolean isInitialized = false;
+ protected Map<String, PartitionEvaluator> needOptimizingPlanMap =
Maps.newHashMap();
protected Map<String, PartitionEvaluator> partitionPlanMap =
Maps.newHashMap();
public OptimizingEvaluator(TableRuntime tableRuntime, MixedTable table) {
@@ -95,7 +97,7 @@ public class OptimizingEvaluator {
LOG.info(
"{} finished evaluating, found {} partitions that need optimizing in
{} ms",
mixedTable.id(),
- partitionPlanMap.size(),
+ needOptimizingPlanMap.size(),
System.currentTimeMillis() - startTime);
}
@@ -129,7 +131,10 @@ public class OptimizingEvaluator {
mixedTable.id(),
count,
System.currentTimeMillis() - startTime);
- partitionPlanMap.values().removeIf(plan -> !plan.isNecessary());
+ needOptimizingPlanMap.putAll(
+ partitionPlanMap.entrySet().stream()
+ .filter(entry -> entry.getValue().isNecessary())
+ .collect(Collectors.toMap(entry -> entry.getKey(), entry ->
entry.getValue())));
}
private Map<String, String> partitionProperties(Pair<Integer, StructLike>
partition) {
@@ -165,7 +170,7 @@ public class OptimizingEvaluator {
if (!isInitialized) {
initEvaluator();
}
- return !partitionPlanMap.isEmpty();
+ return !needOptimizingPlanMap.isEmpty();
}
public PendingInput getPendingInput() {
@@ -175,16 +180,26 @@ public class OptimizingEvaluator {
return new PendingInput(partitionPlanMap.values());
}
+ public PendingInput getOptimizingPendingInput() {
+ if (!isInitialized) {
+ initEvaluator();
+ }
+ return new PendingInput(needOptimizingPlanMap.values());
+ }
+
public static class PendingInput {
@JsonIgnore private final Map<Integer, Set<StructLike>> partitions =
Maps.newHashMap();
private int dataFileCount = 0;
private long dataFileSize = 0;
+ private long dataFileRecords = 0;
private int equalityDeleteFileCount = 0;
private int positionalDeleteFileCount = 0;
private long positionalDeleteBytes = 0L;
private long equalityDeleteBytes = 0L;
+ private long equalityDeleteFileRecords = 0L;
+ private long positionalDeleteFileRecords = 0L;
public PendingInput() {}
@@ -195,9 +210,12 @@ public class OptimizingEvaluator {
.add(evaluator.getPartition().second());
dataFileCount += evaluator.getFragmentFileCount() +
evaluator.getSegmentFileCount();
dataFileSize += evaluator.getFragmentFileSize() +
evaluator.getSegmentFileSize();
+ dataFileRecords += evaluator.getFragmentFileRecords() +
evaluator.getSegmentFileRecords();
positionalDeleteBytes += evaluator.getPosDeleteFileSize();
+ positionalDeleteFileRecords += evaluator.getPosDeleteFileRecords();
positionalDeleteFileCount += evaluator.getPosDeleteFileCount();
equalityDeleteBytes += evaluator.getEqualityDeleteFileSize();
+ equalityDeleteFileRecords += evaluator.getEqualityDeleteFileRecords();
equalityDeleteFileCount += evaluator.getEqualityDeleteFileCount();
}
}
@@ -214,6 +232,10 @@ public class OptimizingEvaluator {
return dataFileSize;
}
+ public long getDataFileRecords() {
+ return dataFileRecords;
+ }
+
public int getEqualityDeleteFileCount() {
return equalityDeleteFileCount;
}
@@ -230,16 +252,27 @@ public class OptimizingEvaluator {
return equalityDeleteBytes;
}
+ public long getEqualityDeleteFileRecords() {
+ return equalityDeleteFileRecords;
+ }
+
+ public long getPositionalDeleteFileRecords() {
+ return positionalDeleteFileRecords;
+ }
+
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("partitions", partitions)
.add("dataFileCount", dataFileCount)
.add("dataFileSize", dataFileSize)
+ .add("dataFileRecords", dataFileRecords)
.add("equalityDeleteFileCount", equalityDeleteFileCount)
.add("positionalDeleteFileCount", positionalDeleteFileCount)
.add("positionalDeleteBytes", positionalDeleteBytes)
.add("equalityDeleteBytes", equalityDeleteBytes)
+ .add("equalityDeleteFileRecords", equalityDeleteFileRecords)
+ .add("positionalDeleteFileRecords", positionalDeleteFileRecords)
.toString();
}
}
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/PartitionEvaluator.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/PartitionEvaluator.java
index 02d59b9a9..974703bf0 100644
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/PartitionEvaluator.java
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/plan/PartitionEvaluator.java
@@ -87,21 +87,33 @@ public interface PartitionEvaluator {
/** Get the total size of fragment files involved in optimizing. */
long getFragmentFileSize();
+ /** Get the total records of fragment files involved in optimizing. */
+ long getFragmentFileRecords();
+
/** Get the count of segment files involved in optimizing. */
int getSegmentFileCount();
/** Get the total size of segment files involved in optimizing. */
long getSegmentFileSize();
+ /** Get the total records of segment files involved in optimizing. */
+ long getSegmentFileRecords();
+
/** Get the count of equality delete files involved in optimizing. */
int getEqualityDeleteFileCount();
/** Get the total size of equality delete files involved in optimizing. */
long getEqualityDeleteFileSize();
+ /** Get the total records of equality delete files involved in optimizing. */
+ long getEqualityDeleteFileRecords();
+
/** Get the count of positional delete files involved in optimizing. */
int getPosDeleteFileCount();
/** Get the total size of positional delete files involved in optimizing. */
long getPosDeleteFileSize();
+
+ /** Get the total records of positional delete files involved in optimizing.
*/
+ long getPosDeleteFileRecords();
}
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/persistence/mapper/TableMetaMapper.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/persistence/mapper/TableMetaMapper.java
index 69cb2303c..f056dcb5e 100644
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/persistence/mapper/TableMetaMapper.java
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/persistence/mapper/TableMetaMapper.java
@@ -329,6 +329,8 @@ public interface TableMetaMapper {
+ " table_config = #{runtime.tableConfiguration,"
+ "
typeHandler=org.apache.amoro.server.persistence.converter.JsonObjectConverter},"
+ " pending_input = #{runtime.pendingInput, jdbcType=VARCHAR,"
+ + "
typeHandler=org.apache.amoro.server.persistence.converter.JsonObjectConverter},"
+ + " table_summary = #{runtime.tableSummary, jdbcType=VARCHAR,"
+ "
typeHandler=org.apache.amoro.server.persistence.converter.JsonObjectConverter}"
+ " WHERE table_id = #{runtime.tableIdentifier.id}")
void updateTableRuntime(@Param("runtime") TableRuntime runtime);
@@ -341,7 +343,7 @@ public interface TableMetaMapper {
+ " current_change_snapshotId, last_optimized_snapshotId,
last_optimized_change_snapshotId,"
+ " last_major_optimizing_time, last_minor_optimizing_time,"
+ " last_full_optimizing_time, optimizing_status,
optimizing_status_start_time, optimizing_process_id,"
- + " optimizer_group, table_config, pending_input) VALUES"
+ + " optimizer_group, table_config, pending_input, table_summary)
VALUES"
+ " (#{runtime.tableIdentifier.id},
#{runtime.tableIdentifier.catalog},"
+ " #{runtime.tableIdentifier.database},
#{runtime.tableIdentifier.tableName}, #{runtime"
+ ".currentSnapshotId},"
@@ -359,6 +361,8 @@ public interface TableMetaMapper {
+ " #{runtime.tableConfiguration,"
+ "
typeHandler=org.apache.amoro.server.persistence.converter.JsonObjectConverter},"
+ " #{runtime.pendingInput, jdbcType=VARCHAR,"
+ + "
typeHandler=org.apache.amoro.server.persistence.converter.JsonObjectConverter},"
+ + " #{runtime.tableSummary, jdbcType=VARCHAR,"
+ "
typeHandler=org.apache.amoro.server.persistence.converter.JsonObjectConverter})")
void insertTableRuntime(@Param("runtime") TableRuntime runtime);
@@ -367,7 +371,7 @@ public interface TableMetaMapper {
+ ".current_change_snapshotId, a.last_optimized_snapshotId,
a.last_optimized_change_snapshotId,"
+ " a.last_major_optimizing_time, a.last_minor_optimizing_time,
a.last_full_optimizing_time, a.optimizing_status,"
+ " a.optimizing_status_start_time, a.optimizing_process_id,"
- + " a.optimizer_group, a.table_config, a.pending_input,
b.optimizing_type, b.target_snapshot_id,"
+ + " a.optimizer_group, a.table_config, a.pending_input,
a.table_summary, b.optimizing_type, b.target_snapshot_id,"
+ " b.target_change_snapshot_id, b.plan_time, b.from_sequence,
b.to_sequence FROM table_runtime a"
+ " INNER JOIN table_identifier i ON a.table_id = i.table_id "
+ " LEFT JOIN table_optimizing_process b ON a.optimizing_process_id
= b.process_id")
@@ -410,6 +414,10 @@ public interface TableMetaMapper {
property = "pendingInput",
column = "pending_input",
typeHandler = JsonObjectConverter.class),
+ @Result(
+ property = "tableSummary",
+ column = "table_summary",
+ typeHandler = JsonObjectConverter.class),
@Result(property = "optimizingType", column = "optimizing_type"),
@Result(property = "targetSnapshotId", column = "target_snapshot_id"),
@Result(property = "targetChangeSnapshotId", column =
"target_change_napshot_id"),
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java
index dbe245f3a..9884d2e78 100644
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java
@@ -88,9 +88,11 @@ public class TableRuntime extends StatedPersistentBase {
@StateField private volatile TableConfiguration tableConfiguration;
@StateField private volatile long processId;
@StateField private volatile OptimizingEvaluator.PendingInput pendingInput;
+ @StateField private volatile OptimizingEvaluator.PendingInput tableSummary;
private volatile long lastPlanTime;
private final TableOptimizingMetrics optimizingMetrics;
private final TableOrphanFilesCleaningMetrics orphanFilesCleaningMetrics;
+ private final TableSummaryMetrics tableSummaryMetrics;
protected TableRuntime(
ServerTableIdentifier tableIdentifier,
@@ -105,6 +107,7 @@ public class TableRuntime extends StatedPersistentBase {
persistTableRuntime();
optimizingMetrics = new TableOptimizingMetrics(tableIdentifier);
orphanFilesCleaningMetrics = new
TableOrphanFilesCleaningMetrics(tableIdentifier);
+ tableSummaryMetrics = new TableSummaryMetrics(tableIdentifier);
}
protected TableRuntime(TableRuntimeMeta tableRuntimeMeta,
TableRuntimeHandler tableHandler) {
@@ -134,12 +137,15 @@ public class TableRuntime extends StatedPersistentBase {
? OptimizingStatus.PENDING
: tableRuntimeMeta.getTableStatus();
this.pendingInput = tableRuntimeMeta.getPendingInput();
+ this.tableSummary = tableRuntimeMeta.getTableSummary();
optimizingMetrics = new TableOptimizingMetrics(tableIdentifier);
optimizingMetrics.statusChanged(optimizingStatus,
this.currentStatusStartTime);
optimizingMetrics.lastOptimizingTime(OptimizingType.MINOR,
this.lastMinorOptimizingTime);
optimizingMetrics.lastOptimizingTime(OptimizingType.MAJOR,
this.lastMajorOptimizingTime);
optimizingMetrics.lastOptimizingTime(OptimizingType.FULL,
this.lastFullOptimizingTime);
orphanFilesCleaningMetrics = new
TableOrphanFilesCleaningMetrics(tableIdentifier);
+ tableSummaryMetrics = new TableSummaryMetrics(tableIdentifier);
+ tableSummaryMetrics.refresh(tableSummary);
}
public void recover(OptimizingProcess optimizingProcess) {
@@ -153,6 +159,7 @@ public class TableRuntime extends StatedPersistentBase {
public void registerMetric(MetricRegistry metricRegistry) {
this.optimizingMetrics.register(metricRegistry);
this.orphanFilesCleaningMetrics.register(metricRegistry);
+ this.tableSummaryMetrics.register(metricRegistry);
}
public void dispose() {
@@ -167,6 +174,7 @@ public class TableRuntime extends StatedPersistentBase {
});
optimizingMetrics.unregister();
orphanFilesCleaningMetrics.unregister();
+ tableSummaryMetrics.unregister();
}
public void beginPlanning() {
@@ -243,6 +251,15 @@ public class TableRuntime extends StatedPersistentBase {
});
}
+ public void setTableSummary(OptimizingEvaluator.PendingInput tableSummary) {
+ invokeConsistency(
+ () -> {
+ this.tableSummary = tableSummary;
+ tableSummaryMetrics.refresh(tableSummary);
+ persistUpdatingRuntime();
+ });
+ }
+
/**
* When there is no task to be optimized, clean pendingInput and update
`lastOptimizedSnapshotId`
* to `currentSnapshotId`.
@@ -322,7 +339,7 @@ public class TableRuntime extends StatedPersistentBase {
private boolean refreshSnapshots(AmoroTable<?> amoroTable) {
MixedTable table = (MixedTable) amoroTable.originalTable();
-
+ tableSummaryMetrics.refreshSnapshots(table);
long lastSnapshotId = currentSnapshotId;
if (table.isKeyedTable()) {
long changeSnapshotId = currentChangeSnapshotId;
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntimeMeta.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntimeMeta.java
index e34bf5363..9fd8c05ea 100644
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntimeMeta.java
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntimeMeta.java
@@ -45,6 +45,7 @@ public class TableRuntimeMeta {
private String optimizerGroup;
private TableConfiguration tableConfig;
private OptimizingEvaluator.PendingInput pendingInput;
+ private OptimizingEvaluator.PendingInput tableSummary;
private long optimizingProcessId = 0;
private OptimizingProcess.Status processStatus;
private OptimizingType optimizingType;
@@ -203,6 +204,18 @@ public class TableRuntimeMeta {
this.lastOptimizedSnapshotId = lastOptimizedSnapshotId;
}
+ public OptimizingEvaluator.PendingInput getTableSummary() {
+ return tableSummary;
+ }
+
+ public void setTableSummary(OptimizingEvaluator.PendingInput tableSummary) {
+ this.tableSummary = tableSummary;
+ }
+
+ public void setTableRuntime(TableRuntime tableRuntime) {
+ this.tableRuntime = tableRuntime;
+ }
+
public void setLastOptimizedChangeSnapshotId(long
lastOptimizedChangeSnapshotId) {
this.lastOptimizedChangeSnapshotId = lastOptimizedChangeSnapshotId;
}
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableSummaryMetrics.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableSummaryMetrics.java
new file mode 100644
index 000000000..cf476c10c
--- /dev/null
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableSummaryMetrics.java
@@ -0,0 +1,241 @@
+/*
+ * 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 static org.apache.amoro.api.metrics.MetricDefine.defineGauge;
+
+import org.apache.amoro.api.ServerTableIdentifier;
+import org.apache.amoro.api.metrics.Gauge;
+import org.apache.amoro.api.metrics.Metric;
+import org.apache.amoro.api.metrics.MetricDefine;
+import org.apache.amoro.api.metrics.MetricKey;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.plan.OptimizingEvaluator;
+import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap;
+import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
+import org.apache.amoro.table.MixedTable;
+import org.apache.amoro.table.UnkeyedTable;
+
+import java.util.List;
+
+/** Table Summary metrics. */
+public class TableSummaryMetrics {
+
+ // table summary files number metrics
+ public static final MetricDefine TABLE_SUMMARY_TOTAL_FILES =
+ defineGauge("table_summary_total_files")
+ .withDescription("Total number of files in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_SUMMARY_DATA_FILES =
+ defineGauge("table_summary_data_files")
+ .withDescription("Number of data files in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_SUMMARY_EQUALITY_DELETE_FILES =
+ defineGauge("table_summary_equality_delete_files")
+ .withDescription("Number of equality delete files in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_SUMMARY_POSITION_DELETE_FILES =
+ defineGauge("table_summary_position_delete_files")
+ .withDescription("Number of position delete files in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ // table summary files size metrics
+ public static final MetricDefine TABLE_SUMMARY_TOTAL_FILES_SIZE =
+ defineGauge("table_summary_total_files_size")
+ .withDescription("Total size of files in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_SUMMARY_DATA_FILES_SIZE =
+ defineGauge("table_summary_data_files_size")
+ .withDescription("Size of data files in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_SUMMARY_EQUALITY_DELETE_FILES_SIZE =
+ defineGauge("table_summary_equality_delete_files_size")
+ .withDescription("Size of equality delete files in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_SUMMARY_POSITION_DELETE_FILES_SIZE =
+ defineGauge("table_summary_position_delete_files_size")
+ .withDescription("Size of position delete files in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ // table summary files records metrics
+ public static final MetricDefine TABLE_SUMMARY_TOTAL_RECORDS =
+ defineGauge("table_summary_total_records")
+ .withDescription("Total records in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_SUMMARY_DATA_FILES_RECORDS =
+ defineGauge("table_summary_data_files_records")
+ .withDescription("Records of data files in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_SUMMARY_EQUALITY_DELETE_FILES_RECORDS
=
+ defineGauge("table_summary_equality_delete_files_records")
+ .withDescription("Records of equality delete files in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_SUMMARY_POSITION_DELETE_FILES_RECORDS
=
+ defineGauge("table_summary_position_delete_files_records")
+ .withDescription("Records of position delete files in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ // table summary snapshots number metrics
+ public static final MetricDefine TABLE_SUMMARY_SNAPSHOTS =
+ defineGauge("table_summary_snapshots")
+ .withDescription("Number of snapshots in the table")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ private final ServerTableIdentifier identifier;
+ private final List<MetricKey> registeredMetricKeys = Lists.newArrayList();
+ private MetricRegistry globalRegistry;
+
+ private long totalFiles = 0L;
+ private long dataFiles = 0L;
+ private long positionDeleteFiles = 0L;
+ private long equalityDeleteFiles = 0L;
+ private long totalFilesSize = 0L;
+ private long positionDeleteFilesSize = 0L;
+ private long dataFilesSize = 0L;
+ private long equalityDeleteFilesSize = 0L;
+ private long positionDeleteFilesRecords = 0L;
+ private long totalRecords = 0L;
+ private long dataFilesRecords = 0L;
+ private long equalityDeleteFilesRecords = 0L;
+ private long snapshots = 0L;
+
+ public TableSummaryMetrics(ServerTableIdentifier identifier) {
+ this.identifier = identifier;
+ }
+
+ private void registerMetric(MetricRegistry registry, MetricDefine define,
Metric metric) {
+ MetricKey key =
+ registry.register(
+ define,
+ ImmutableMap.of(
+ "catalog",
+ identifier.getCatalog(),
+ "database",
+ identifier.getDatabase(),
+ "table",
+ identifier.getTableName()),
+ metric);
+ registeredMetricKeys.add(key);
+ }
+
+ public void register(MetricRegistry registry) {
+ if (globalRegistry == null) {
+ // register files number metrics
+ registerMetric(registry, TABLE_SUMMARY_TOTAL_FILES, (Gauge<Long>) () ->
totalFiles);
+ registerMetric(registry, TABLE_SUMMARY_DATA_FILES, (Gauge<Long>) () ->
dataFiles);
+ registerMetric(
+ registry, TABLE_SUMMARY_POSITION_DELETE_FILES, (Gauge<Long>) () ->
positionDeleteFiles);
+ registerMetric(
+ registry, TABLE_SUMMARY_EQUALITY_DELETE_FILES, (Gauge<Long>) () ->
equalityDeleteFiles);
+
+ // register files size metrics
+ registerMetric(registry, TABLE_SUMMARY_TOTAL_FILES_SIZE, (Gauge<Long>)
() -> totalFilesSize);
+ registerMetric(registry, TABLE_SUMMARY_DATA_FILES_SIZE, (Gauge<Long>) ()
-> dataFilesSize);
+ registerMetric(
+ registry,
+ TABLE_SUMMARY_POSITION_DELETE_FILES_SIZE,
+ (Gauge<Long>) () -> positionDeleteFilesSize);
+ registerMetric(
+ registry,
+ TABLE_SUMMARY_EQUALITY_DELETE_FILES_SIZE,
+ (Gauge<Long>) () -> equalityDeleteFilesSize);
+
+ // register files records metrics
+ registerMetric(registry, TABLE_SUMMARY_TOTAL_RECORDS, (Gauge<Long>) ()
-> totalRecords);
+ registerMetric(
+ registry, TABLE_SUMMARY_DATA_FILES_RECORDS, (Gauge<Long>) () ->
dataFilesRecords);
+ registerMetric(
+ registry,
+ TABLE_SUMMARY_POSITION_DELETE_FILES_RECORDS,
+ (Gauge<Long>) () -> positionDeleteFilesRecords);
+ registerMetric(
+ registry,
+ TABLE_SUMMARY_EQUALITY_DELETE_FILES_RECORDS,
+ (Gauge<Long>) () -> equalityDeleteFilesRecords);
+
+ // register snapshots number metrics
+ registerMetric(registry, TABLE_SUMMARY_SNAPSHOTS, (Gauge<Long>) () ->
snapshots);
+
+ globalRegistry = registry;
+ }
+ }
+
+ public void unregister() {
+ registeredMetricKeys.forEach(globalRegistry::unregister);
+ registeredMetricKeys.clear();
+ globalRegistry = null;
+ }
+
+ public void refresh(OptimizingEvaluator.PendingInput tableSummary) {
+ if (tableSummary == null) {
+ return;
+ }
+ totalFiles =
+ tableSummary.getDataFileCount()
+ + tableSummary.getEqualityDeleteFileCount()
+ + tableSummary.getPositionalDeleteFileCount();
+ dataFiles = tableSummary.getDataFileCount();
+ positionDeleteFiles = tableSummary.getPositionalDeleteFileCount();
+ equalityDeleteFiles = tableSummary.getEqualityDeleteFileCount();
+
+ totalFilesSize =
+ tableSummary.getDataFileSize()
+ + tableSummary.getEqualityDeleteBytes()
+ + tableSummary.getPositionalDeleteBytes();
+ positionDeleteFilesSize = tableSummary.getPositionalDeleteBytes();
+ dataFilesSize = tableSummary.getDataFileSize();
+ equalityDeleteFilesSize = tableSummary.getEqualityDeleteBytes();
+
+ totalRecords =
+ tableSummary.getDataFileRecords()
+ + tableSummary.getEqualityDeleteFileRecords()
+ + tableSummary.getPositionalDeleteFileRecords();
+ positionDeleteFilesRecords = tableSummary.getPositionalDeleteFileRecords();
+ dataFilesRecords = tableSummary.getDataFileRecords();
+ equalityDeleteFilesRecords = tableSummary.getEqualityDeleteFileRecords();
+ }
+
+ public void refreshSnapshots(MixedTable table) {
+ UnkeyedTable unkeyedTable =
+ table.isKeyedTable() ? table.asKeyedTable().baseTable() :
table.asUnkeyedTable();
+ snapshots = Lists.newArrayList(unkeyedTable.snapshots().iterator()).size();
+ }
+}
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/executor/TableRuntimeRefreshExecutor.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/executor/TableRuntimeRefreshExecutor.java
index e36c7413e..4fa3598a8 100644
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/executor/TableRuntimeRefreshExecutor.java
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/executor/TableRuntimeRefreshExecutor.java
@@ -50,7 +50,7 @@ public class TableRuntimeRefreshExecutor extends
BaseTableExecutor {
if (tableRuntime.isOptimizingEnabled() &&
!tableRuntime.getOptimizingStatus().isProcessing()) {
OptimizingEvaluator evaluator = new OptimizingEvaluator(tableRuntime,
table);
if (evaluator.isNecessary()) {
- OptimizingEvaluator.PendingInput pendingInput =
evaluator.getPendingInput();
+ OptimizingEvaluator.PendingInput pendingInput =
evaluator.getOptimizingPendingInput();
logger.debug(
"{} optimizing is necessary and get pending input {}",
tableRuntime.getTableIdentifier(),
@@ -59,6 +59,7 @@ public class TableRuntimeRefreshExecutor extends
BaseTableExecutor {
} else {
tableRuntime.optimizingNotNecessary();
}
+ tableRuntime.setTableSummary(evaluator.getPendingInput());
}
}
diff --git
a/amoro-ams/amoro-ams-server/src/main/resources/derby/ams-derby-init.sql
b/amoro-ams/amoro-ams-server/src/main/resources/derby/ams-derby-init.sql
index a7dc2ecf1..f41cdaf27 100644
--- a/amoro-ams/amoro-ams-server/src/main/resources/derby/ams-derby-init.sql
+++ b/amoro-ams/amoro-ams-server/src/main/resources/derby/ams-derby-init.sql
@@ -118,6 +118,7 @@ CREATE TABLE table_runtime (
table_config CLOB(64m),
optimizing_config CLOB(64m),
pending_input CLOB(64m),
+ table_summary CLOB(64m),
CONSTRAINT table_runtime_pk PRIMARY KEY (table_id),
CONSTRAINT table_runtime_table_name_idx UNIQUE (catalog_name, db_name,
table_name)
);
diff --git
a/amoro-ams/amoro-ams-server/src/main/resources/mysql/ams-mysql-init.sql
b/amoro-ams/amoro-ams-server/src/main/resources/mysql/ams-mysql-init.sql
index 45bc86092..76e16147a 100644
--- a/amoro-ams/amoro-ams-server/src/main/resources/mysql/ams-mysql-init.sql
+++ b/amoro-ams/amoro-ams-server/src/main/resources/mysql/ams-mysql-init.sql
@@ -129,6 +129,7 @@ CREATE TABLE `table_runtime`
`table_config` mediumtext,
`optimizing_config` mediumtext,
`pending_input` mediumtext,
+ `table_summary` mediumtext,
PRIMARY KEY (`table_id`),
UNIQUE KEY `table_index` (`catalog_name`,`db_name`,`table_name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT 'Optimize running information
of each table' ROW_FORMAT=DYNAMIC;
diff --git a/amoro-ams/amoro-ams-server/src/main/resources/mysql/upgrade.sql
b/amoro-ams/amoro-ams-server/src/main/resources/mysql/upgrade.sql
index a457d458d..31ab7604d 100644
--- a/amoro-ams/amoro-ams-server/src/main/resources/mysql/upgrade.sql
+++ b/amoro-ams/amoro-ams-server/src/main/resources/mysql/upgrade.sql
@@ -21,3 +21,6 @@ TRUNCATE TABLE `table_blocker`;
ALTER TABLE `table_blocker` DROP INDEX `table_index`;
ALTER TABLE `table_blocker` ADD COLUMN `prev_blocker_id` bigint(20) NOT NULL
DEFAULT -1 COMMENT 'prev blocker id when created';
ALTER TABLE `table_blocker` ADD UNIQUE KEY `uq_prev`
(`catalog_name`,`db_name`,`table_name`, `prev_blocker_id`);
+
+-- ADD COLUMN table_summary FOR TABLE_RUNTIME
+ALTER TABLE `table_runtime` ADD COLUMN `table_summary` mediumtext AFTER
`pending_input`;
\ No newline at end of file
diff --git
a/amoro-ams/amoro-ams-server/src/main/resources/postgres/ams-postgres-init.sql
b/amoro-ams/amoro-ams-server/src/main/resources/postgres/ams-postgres-init.sql
index 40688a5ed..83c3f6d0f 100644
---
a/amoro-ams/amoro-ams-server/src/main/resources/postgres/ams-postgres-init.sql
+++
b/amoro-ams/amoro-ams-server/src/main/resources/postgres/ams-postgres-init.sql
@@ -189,6 +189,7 @@ CREATE TABLE table_runtime
table_config TEXT,
optimizing_config TEXT,
pending_input TEXT,
+ table_summary TEXT,
PRIMARY KEY (table_id),
UNIQUE (catalog_name, db_name, table_name)
);
@@ -211,6 +212,7 @@ COMMENT ON COLUMN table_runtime.optimizer_group IS
'Optimizer group';
COMMENT ON COLUMN table_runtime.table_config IS 'Table-specific configuration';
COMMENT ON COLUMN table_runtime.optimizing_config IS 'Optimizing
configuration';
COMMENT ON COLUMN table_runtime.pending_input IS 'Pending input data';
+COMMENT ON COLUMN table_runtime.table_summary IS 'Table summary data';
CREATE TABLE table_optimizing_process
(
diff --git a/amoro-ams/amoro-ams-server/src/main/resources/postgres/upgrade.sql
b/amoro-ams/amoro-ams-server/src/main/resources/postgres/upgrade.sql
index 0b07e8805..0267468c2 100644
--- a/amoro-ams/amoro-ams-server/src/main/resources/postgres/upgrade.sql
+++ b/amoro-ams/amoro-ams-server/src/main/resources/postgres/upgrade.sql
@@ -22,3 +22,7 @@ ALTER TABLE `table_blocker` DROP INDEX `table_index`;
ALTER TABLE `table_blocker` ADD COLUMN `prev_blocker_id` bigint(20) NOT NULL
DEFAULT -1;
COMMENT ON COLUMN table_blocker.prev_blocker_id IS 'prev blocker id when
created';
ALTER TABLE `table_blocker` ADD UNIQUE KEY `uq_prev`
(`catalog_name`,`db_name`,`table_name`, `prev_blocker_id`);
+
+-- ADD COLUMN table_summary FOR TABLE_RUNTIME
+ALTER TABLE table_runtime ADD COLUMN table_summary TEXT;
+COMMENT ON COLUMN table_runtime.table_summary IS 'Table summary data';
diff --git
a/amoro-ams/amoro-ams-server/src/test/java/org/apache/amoro/server/optimizing/plan/TestOptimizingEvaluator.java
b/amoro-ams/amoro-ams-server/src/test/java/org/apache/amoro/server/optimizing/plan/TestOptimizingEvaluator.java
index 07c77179b..63a853fe1 100644
---
a/amoro-ams/amoro-ams-server/src/test/java/org/apache/amoro/server/optimizing/plan/TestOptimizingEvaluator.java
+++
b/amoro-ams/amoro-ams-server/src/test/java/org/apache/amoro/server/optimizing/plan/TestOptimizingEvaluator.java
@@ -71,7 +71,7 @@ public class TestOptimizingEvaluator extends
MixedTablePlanTestBase {
public void testEmpty() {
OptimizingEvaluator optimizingEvaluator = buildOptimizingEvaluator();
Assert.assertFalse(optimizingEvaluator.isNecessary());
- OptimizingEvaluator.PendingInput pendingInput =
optimizingEvaluator.getPendingInput();
+ OptimizingEvaluator.PendingInput pendingInput =
optimizingEvaluator.getOptimizingPendingInput();
assertEmptyInput(pendingInput);
}
@@ -90,7 +90,7 @@ public class TestOptimizingEvaluator extends
MixedTablePlanTestBase {
OptimizingEvaluator optimizingEvaluator = buildOptimizingEvaluator();
Assert.assertFalse(optimizingEvaluator.isNecessary());
- OptimizingEvaluator.PendingInput pendingInput =
optimizingEvaluator.getPendingInput();
+ OptimizingEvaluator.PendingInput pendingInput =
optimizingEvaluator.getOptimizingPendingInput();
assertEmptyInput(pendingInput);
// add more files
@@ -104,7 +104,7 @@ public class TestOptimizingEvaluator extends
MixedTablePlanTestBase {
optimizingEvaluator = buildOptimizingEvaluator();
Assert.assertTrue(optimizingEvaluator.isNecessary());
- pendingInput = optimizingEvaluator.getPendingInput();
+ pendingInput = optimizingEvaluator.getOptimizingPendingInput();
assertInput(pendingInput, FileInfo.buildFileInfo(dataFiles));
}
diff --git
a/amoro-ams/amoro-ams-server/src/test/java/org/apache/amoro/server/table/TestTableSummaryMetrics.java
b/amoro-ams/amoro-ams-server/src/test/java/org/apache/amoro/server/table/TestTableSummaryMetrics.java
new file mode 100644
index 000000000..f3b4f20dc
--- /dev/null
+++
b/amoro-ams/amoro-ams-server/src/test/java/org/apache/amoro/server/table/TestTableSummaryMetrics.java
@@ -0,0 +1,223 @@
+/*
+ * 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 static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_DATA_FILES;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_DATA_FILES_RECORDS;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_DATA_FILES_SIZE;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_EQUALITY_DELETE_FILES;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_EQUALITY_DELETE_FILES_RECORDS;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_EQUALITY_DELETE_FILES_SIZE;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_POSITION_DELETE_FILES;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_POSITION_DELETE_FILES_RECORDS;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_POSITION_DELETE_FILES_SIZE;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_SNAPSHOTS;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_TOTAL_FILES;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_TOTAL_FILES_SIZE;
+import static
org.apache.amoro.server.table.TableSummaryMetrics.TABLE_SUMMARY_TOTAL_RECORDS;
+
+import org.apache.amoro.BasicTableTestHelper;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.TableTestHelper;
+import org.apache.amoro.api.ServerTableIdentifier;
+import org.apache.amoro.api.metrics.Gauge;
+import org.apache.amoro.api.metrics.Metric;
+import org.apache.amoro.api.metrics.MetricDefine;
+import org.apache.amoro.api.metrics.MetricKey;
+import org.apache.amoro.catalog.BasicCatalogTestHelper;
+import org.apache.amoro.catalog.CatalogTestHelper;
+import org.apache.amoro.io.MixedDataTestHelpers;
+import org.apache.amoro.server.manager.MetricManager;
+import org.apache.amoro.server.optimizing.OptimizingTestHelpers;
+import org.apache.amoro.server.table.executor.TableRuntimeRefreshExecutor;
+import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap;
+import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
+import org.apache.amoro.table.MixedTable;
+import org.apache.amoro.table.UnkeyedTable;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.data.Record;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+@RunWith(Parameterized.class)
+public class TestTableSummaryMetrics extends AMSTableTestBase {
+
+ @Parameterized.Parameters(name = "{0}, {1}")
+ public static Object[] parameters() {
+ return new Object[][] {
+ {new BasicCatalogTestHelper(TableFormat.ICEBERG), new
BasicTableTestHelper(true, false)}
+ };
+ }
+
+ public TestTableSummaryMetrics(
+ CatalogTestHelper catalogTestHelper, TableTestHelper tableTestHelper) {
+ super(catalogTestHelper, tableTestHelper, false);
+ }
+
+ @Before
+ public void prepare() {
+ createDatabase();
+ createTable();
+ }
+
+ @After
+ public void clear() {
+ try {
+ dropTable();
+ dropDatabase();
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+
+ private void initTableWithFiles() {
+ UnkeyedTable table =
+ ((MixedTable)
tableService().loadTable(serverTableIdentifier()).originalTable())
+ .asUnkeyedTable();
+ appendData(table);
+ appendPosDelete(table);
+ TableRuntime runtime = tableService().getRuntime(serverTableIdentifier());
+ runtime.refresh(tableService().loadTable(serverTableIdentifier()));
+ }
+
+ private void appendData(UnkeyedTable table) {
+ ArrayList<Record> newRc =
+ Lists.newArrayList(
+ tableTestHelper().generateTestRecord(1, "111", 0,
"2022-01-01T12:00:00"),
+ tableTestHelper().generateTestRecord(2, "222", 0,
"2022-01-01T12:00:00"));
+ List<DataFile> dataFiles =
+ OptimizingTestHelpers.appendBase(
+ table, tableTestHelper().writeBaseStore(table, 0L, newRc, false));
+
+ AppendFiles appendFiles = table.newAppend();
+ dataFiles.forEach(appendFiles::appendFile);
+ appendFiles.commit();
+ }
+
+ private void appendPosDelete(UnkeyedTable table) {
+ ArrayList<Record> newRc =
+ Lists.newArrayList(
+ tableTestHelper().generateTestRecord(3, "333", 0,
"2022-01-01T12:00:00"),
+ tableTestHelper().generateTestRecord(4, "444", 0,
"2022-01-01T12:00:00"));
+ List<DataFile> dataFiles =
+ OptimizingTestHelpers.appendBase(
+ table, tableTestHelper().writeBaseStore(table, 0L, newRc, false));
+ List<DeleteFile> posDeleteFiles = Lists.newArrayList();
+ for (DataFile dataFile : dataFiles) {
+ posDeleteFiles.addAll(
+ MixedDataTestHelpers.writeBaseStorePosDelete(
+ table, 0L, dataFile, Collections.singletonList(0L)));
+ }
+ OptimizingTestHelpers.appendBasePosDelete(table, posDeleteFiles);
+ }
+
+ void refreshPending() {
+ TableRuntimeRefreshExecutor refresher =
+ new TableRuntimeRefreshExecutor(tableService(), 1, Integer.MAX_VALUE);
+ refresher.execute(tableService().getRuntime(serverTableIdentifier()));
+ refresher.dispose();
+ }
+
+ @Test
+ public void testTableSummaryMetrics() {
+ ServerTableIdentifier identifier = serverTableIdentifier();
+ Map<MetricKey, Metric> metrics =
MetricManager.getInstance().getGlobalRegistry().getMetrics();
+ Gauge<Long> totalFiles = getMetric(metrics, identifier,
TABLE_SUMMARY_TOTAL_FILES);
+ Gauge<Long> dataFiles = getMetric(metrics, identifier,
TABLE_SUMMARY_DATA_FILES);
+ Gauge<Long> posDelFiles = getMetric(metrics, identifier,
TABLE_SUMMARY_POSITION_DELETE_FILES);
+ Gauge<Long> eqDelFiles = getMetric(metrics, identifier,
TABLE_SUMMARY_EQUALITY_DELETE_FILES);
+
+ Gauge<Long> totalSize = getMetric(metrics, identifier,
TABLE_SUMMARY_TOTAL_FILES_SIZE);
+ Gauge<Long> dataSize = getMetric(metrics, identifier,
TABLE_SUMMARY_DATA_FILES_SIZE);
+ Gauge<Long> posDelSize =
+ getMetric(metrics, identifier,
TABLE_SUMMARY_POSITION_DELETE_FILES_SIZE);
+ Gauge<Long> eqDelSize =
+ getMetric(metrics, identifier,
TABLE_SUMMARY_EQUALITY_DELETE_FILES_SIZE);
+
+ Gauge<Long> totalRecords = getMetric(metrics, identifier,
TABLE_SUMMARY_TOTAL_RECORDS);
+ Gauge<Long> dataRecords = getMetric(metrics, identifier,
TABLE_SUMMARY_DATA_FILES_RECORDS);
+ Gauge<Long> posDelRecords =
+ getMetric(metrics, identifier,
TABLE_SUMMARY_POSITION_DELETE_FILES_RECORDS);
+ Gauge<Long> eqDelRecords =
+ getMetric(metrics, identifier,
TABLE_SUMMARY_EQUALITY_DELETE_FILES_RECORDS);
+
+ Gauge<Long> snapshots = getMetric(metrics, identifier,
TABLE_SUMMARY_SNAPSHOTS);
+
+ Assertions.assertEquals(0, totalFiles.getValue());
+ Assertions.assertEquals(0, dataFiles.getValue());
+ Assertions.assertEquals(0, posDelFiles.getValue());
+ Assertions.assertEquals(0, eqDelFiles.getValue());
+
+ Assertions.assertEquals(0, totalSize.getValue());
+ Assertions.assertEquals(0, dataSize.getValue());
+ Assertions.assertEquals(0, posDelSize.getValue());
+ Assertions.assertEquals(0, eqDelSize.getValue());
+
+ Assertions.assertEquals(0, totalRecords.getValue());
+ Assertions.assertEquals(0, dataRecords.getValue());
+ Assertions.assertEquals(0, posDelRecords.getValue());
+ Assertions.assertEquals(0, eqDelRecords.getValue());
+
+ // refresh metrics
+ initTableWithFiles();
+ refreshPending();
+
+ Assertions.assertTrue(totalFiles.getValue() > 0);
+ Assertions.assertTrue(dataFiles.getValue() > 0);
+ Assertions.assertTrue(posDelFiles.getValue() > 0);
+
+ Assertions.assertTrue(totalSize.getValue() > 0);
+ Assertions.assertTrue(dataSize.getValue() > 0);
+ Assertions.assertTrue(posDelSize.getValue() > 0);
+
+ Assertions.assertTrue(totalRecords.getValue() > 0);
+ Assertions.assertTrue(dataRecords.getValue() > 0);
+ Assertions.assertTrue(posDelRecords.getValue() > 0);
+
+ Assertions.assertTrue(snapshots.getValue() > 0);
+ }
+
+ private Gauge<Long> getMetric(
+ Map<MetricKey, Metric> metrics, ServerTableIdentifier identifier,
MetricDefine metricDefine) {
+ Gauge<Long> metric =
+ (Gauge<Long>)
+ metrics.get(
+ new MetricKey(
+ metricDefine,
+ ImmutableMap.of(
+ "catalog",
+ identifier.getCatalog(),
+ "database",
+ identifier.getDatabase(),
+ "table",
+ identifier.getTableName())));
+ return metric;
+ }
+}
diff --git a/docs/user-guides/metrics.md b/docs/user-guides/metrics.md
index c84d565d8..cc86bcb0d 100644
--- a/docs/user-guides/metrics.md
+++ b/docs/user-guides/metrics.md
@@ -78,11 +78,11 @@ Amoro has supported built-in metrics to measure status of
table self-optimizing
## Orphan Files Cleaning metrics
-| Metric Name | Type | Tags
| Description
|
-|-------------------------------------------|---------|--------------------------|--------------------------------------------------------------------------------|
-| table_orphan_content_file_cleaning_count | Counter | catalog, database,
table | Count of orphan content files cleaned in the table since ams started
|
-| table_orphan_metadata_file_cleaning_count | Counter | catalog, database,
table | Count of orphan metadata files cleaned in the table since ams started
|
-| table_expected_orphan_content_file_cleaning_count | Counter | catalog,
database, table | Expected Count of orphan content files cleaned in the table
since ams started |
+| Metric Name | Type | Tags
| Description
|
+|----------------------------------------------------|---------|--------------------------|--------------------------------------------------------------------------------|
+| table_orphan_content_file_cleaning_count | Counter | catalog,
database, table | Count of orphan content files cleaned in the table since ams
started |
+| table_orphan_metadata_file_cleaning_count | Counter | catalog,
database, table | Count of orphan metadata files cleaned in the table since ams
started |
+| table_expected_orphan_content_file_cleaning_count | Counter | catalog,
database, table | Expected Count of orphan content files cleaned in the table
since ams started |
| table_expected_orphan_metadata_file_cleaning_count | Counter | catalog,
database, table | Expected Count of orphan metadata files cleaned in the table
since ams started |
@@ -97,3 +97,21 @@ Amoro has supported built-in metrics to measure status of
table self-optimizing
| ams_jvm_threads_count | Gauge |
| The total number of live threads used by the AMS |
| ams_jvm_garbage_collector_count | Gauge
|garbage_collector| The count of the JVM's Garbage Collector, such as G1 Young
|
| ams_jvm_garbage_collector_time | Gauge
|garbage_collector| The time spent by the JVM's Garbage Collector, such as G1
Young |
+
+## table summary metrics
+
+| Metric Name | Type | Tags
| Description |
+|---------------------------------------------|---------|--------------------------|-----------------------------------------------|
+| table_summary_total_files | Gauge | catalog, database,
table | Total number of files in the table |
+| table_summary_data_files | Gauge | catalog, database,
table | Number of data files in the table |
+| table_summary_equality_delete_files | Gauge | catalog, database,
table | Number of equality delete files in the table |
+| table_summary_position_delete_files | Gauge | catalog, database,
table | Number of position delete files in the table |
+| table_summary_total_files_size | Gauge | catalog, database,
table | Total size of files in the table |
+| table_summary_data_files_size | Gauge | catalog, database,
table | Size of data files in the table |
+| table_summary_equality_delete_files_size | Gauge | catalog, database,
table | Size of equality delete files in the table |
+| table_summary_position_delete_files_size | Gauge | catalog, database,
table | Size of position delete files in the table |
+| table_summary_total_records | Gauge | catalog, database,
table | Total records in the table |
+| table_summary_data_files_records | Gauge | catalog, database,
table | Records of data files in the table |
+| table_summary_equality_delete_files_records | Gauge | catalog, database,
table | Records of equality delete files in the table |
+| table_summary_position_delete_files_records | Gauge | catalog, database,
table | Records of position delete files in the table |
+| table_summary_snapshots | Gauge | catalog, database,
table | Number of snapshots in the table |
\ No newline at end of file
