You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "tsreaper (via GitHub)" <gi...@apache.org> on 2023/07/28 08:44:04 UTC
[GitHub] [incubator-paimon] tsreaper commented on a diff in pull request #1638: [core] Support commit metrics
tsreaper commented on code in PR #1638:
URL: https://github.com/apache/incubator-paimon/pull/1638#discussion_r1277045531
##########
paimon-core/src/main/java/org/apache/paimon/metrics/commit/CommitMetrics.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.paimon.metrics.commit;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.metrics.Counter;
+import org.apache.paimon.metrics.DescriptiveStatisticsHistogram;
+import org.apache.paimon.metrics.Histogram;
+import org.apache.paimon.metrics.MetricGroup;
+import org.apache.paimon.metrics.groups.GenericMetricGroup;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+/** Metrics to measure a commit. */
+public class CommitMetrics {
+ private static final int HISTOGRAM_WINDOW_SIZE = 10_000;
+ protected static final String GROUP_NAME = "commit";
+
+ private final MetricGroup genericMetricGroup;
+ private final FileStorePathFactory pathFactory;
+
+ public CommitMetrics(FileStorePathFactory pathFactory) {
+ this.pathFactory = pathFactory;
+ this.genericMetricGroup =
+ GenericMetricGroup.createGenericMetricGroup(
+ pathFactory.root().getName(), GROUP_NAME);
+ registerGenericCommitMetrics();
+ }
+
+ public MetricGroup getGenericMetricGroup() {
+ return genericMetricGroup;
+ }
+
+ private final Histogram durationHistogram =
+ new DescriptiveStatisticsHistogram(HISTOGRAM_WINDOW_SIZE);
+
+ private CommitStats latestCommit;
+
+ @VisibleForTesting static final String LAST_COMMIT_DURATION = "lastCommitDuration";
+ @VisibleForTesting static final String COMMIT_DURATION = "commitDuration";
+ @VisibleForTesting static final String LAST_COMMIT_ATTEMPTS = "lastCommitAttempts";
+ @VisibleForTesting static final String LAST_TABLE_FILES_ADDED = "lastTableFilesAdded";
+ @VisibleForTesting static final String LAST_TABLE_FILES_DELETED = "lastTableFilesDeleted";
+ @VisibleForTesting static final String LAST_TABLE_FILES_APPENDED = "lastTableFilesAppended";
+
+ @VisibleForTesting
+ static final String LAST_TABLE_FILES_COMMIT_COMPACTED = "lastTableFilesCommitCompacted";
Review Comment:
Why not `lastTableFilesCompacted` (which corresponds to `lastTableFilesAppended`)?
##########
paimon-core/src/main/java/org/apache/paimon/metrics/commit/CommitStats.java:
##########
@@ -0,0 +1,213 @@
+/*
+ * 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.paimon.metrics.commit;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Statistics for a commit. */
+public class CommitStats {
+ private final long duration;
+ private final long attempts;
+ private final long tableFilesAdded;
+ private final long tableFilesAppended;
+ private final long tableFilesDeleted;
+ private final long changelogFilesAppended;
+ private final long changelogFilesCompacted;
+ private final long changelogRecordsCompacted;
+
+ private final long deltaRecordsCompacted;
+ private final long changelogRecordsAppended;
+ private final long deltaRecordsAppended;
+ private final long tableFilesCompacted;
+ private final long generatedSnapshots;
+ private final long numPartitionsWritten;
+ private final long numBucketsWritten;
+ private final Map<BinaryRow, Set<Integer>> bucketsWritten;
Review Comment:
Is it needed any more? Remove useless code.
##########
paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java:
##########
@@ -226,19 +234,33 @@ public void commit(ManifestCommittable committable, Map<String, String> properti
baseEntries.addAll(
readAllEntriesFromChangedPartitions(
latestSnapshot, appendTableFiles, compactTableFiles));
- noConflictsOrFail(latestSnapshot.commitUser(), baseEntries, appendTableFiles);
+ try {
+ noConflictsOrFail(latestSnapshot.commitUser(), baseEntries, appendTableFiles);
+ } catch (RuntimeException e) {
+ reportCommit(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ 0,
+ 0,
+ 1);
+ throw e;
+ }
Review Comment:
Avoid adding `try... catch...` every where. Use `try... finally...`.
##########
paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java:
##########
@@ -481,7 +547,7 @@ private ManifestEntry makeEntry(FileKind kind, CommitMessage commitMessage, Data
kind, commitMessage.partition(), commitMessage.bucket(), numBucket, file);
}
- private void tryCommit(
+ private long tryCommit(
Review Comment:
No need for `long`. It is impossible to perform `2137483648` or more attempts.
##########
paimon-core/src/main/java/org/apache/paimon/metrics/commit/CommitMetrics.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.paimon.metrics.commit;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.metrics.Counter;
+import org.apache.paimon.metrics.DescriptiveStatisticsHistogram;
+import org.apache.paimon.metrics.Histogram;
+import org.apache.paimon.metrics.MetricGroup;
+import org.apache.paimon.metrics.groups.GenericMetricGroup;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+/** Metrics to measure a commit. */
+public class CommitMetrics {
+ private static final int HISTOGRAM_WINDOW_SIZE = 10_000;
+ protected static final String GROUP_NAME = "commit";
+
+ private final MetricGroup genericMetricGroup;
+ private final FileStorePathFactory pathFactory;
+
+ public CommitMetrics(FileStorePathFactory pathFactory) {
+ this.pathFactory = pathFactory;
+ this.genericMetricGroup =
+ GenericMetricGroup.createGenericMetricGroup(
+ pathFactory.root().getName(), GROUP_NAME);
+ registerGenericCommitMetrics();
+ }
+
+ public MetricGroup getGenericMetricGroup() {
+ return genericMetricGroup;
+ }
+
+ private final Histogram durationHistogram =
+ new DescriptiveStatisticsHistogram(HISTOGRAM_WINDOW_SIZE);
+
+ private CommitStats latestCommit;
+
+ @VisibleForTesting static final String LAST_COMMIT_DURATION = "lastCommitDuration";
+ @VisibleForTesting static final String COMMIT_DURATION = "commitDuration";
+ @VisibleForTesting static final String LAST_COMMIT_ATTEMPTS = "lastCommitAttempts";
+ @VisibleForTesting static final String LAST_TABLE_FILES_ADDED = "lastTableFilesAdded";
+ @VisibleForTesting static final String LAST_TABLE_FILES_DELETED = "lastTableFilesDeleted";
+ @VisibleForTesting static final String LAST_TABLE_FILES_APPENDED = "lastTableFilesAppended";
+
+ @VisibleForTesting
+ static final String LAST_TABLE_FILES_COMMIT_COMPACTED = "lastTableFilesCommitCompacted";
+
+ @VisibleForTesting
+ static final String LAST_CHANGELOG_FILES_APPENDED = "lastChangelogFilesAppended";
+
+ @VisibleForTesting
+ static final String LAST_CHANGELOG_FILES_COMMIT_COMPACTED = "lastChangelogFileCommitCompacted";
+
+ @VisibleForTesting static final String TOTAL_TABLE_FILES = "totalTableFiles";
+ @VisibleForTesting static final String TOTAL_CHANGELOG_FILES = "totalChangelogFiles";
+ @VisibleForTesting static final String LAST_GENERATED_SNAPSHOTS = "lastGeneratedSnapshots";
+ @VisibleForTesting static final String LAST_DELTA_RECORDS_APPENDED = "lastDeltaRecordsAppended";
+
+ @VisibleForTesting
+ static final String LAST_CHANGELOG_RECORDS_APPENDED = "lastChangelogRecordsAppended";
+
+ @VisibleForTesting
+ static final String LAST_DELTA_RECORDS_COMMIT_COMPACTED = "lastDeltaRecordsCommitCompacted";
+
+ @VisibleForTesting
+ static final String LAST_CHANGELOG_RECORDS_COMMIT_COMPACTED =
+ "lastChangelogRecordsCommitCompacted";
+
+ @VisibleForTesting static final String LAST_PARTITIONS_WRITTEN = "lastPartitionsWritten";
+ @VisibleForTesting static final String LAST_BUCKETS_WRITTEN = "lastBucketsWritten";
+
+ transient Counter totalTableFilesCounter;
+ transient Counter totalChangelogFilesCounter;
+
+ private void registerGenericCommitMetrics() {
+ genericMetricGroup.gauge(
+ LAST_COMMIT_DURATION,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getDuration();
+ });
+ genericMetricGroup.gauge(
+ LAST_COMMIT_ATTEMPTS,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getAttempts();
+ });
+ genericMetricGroup.gauge(
+ LAST_GENERATED_SNAPSHOTS,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getGeneratedSnapshots();
+ });
+ genericMetricGroup.gauge(
+ LAST_PARTITIONS_WRITTEN,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getNumPartitionsWritten();
+ });
+ genericMetricGroup.gauge(
+ LAST_BUCKETS_WRITTEN,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getNumBucketsWritten();
+ });
+ genericMetricGroup.histogram(COMMIT_DURATION, durationHistogram);
+ genericMetricGroup.gauge(
+ LAST_TABLE_FILES_ADDED,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getTableFilesAdded();
+ });
+ genericMetricGroup.gauge(
+ LAST_TABLE_FILES_DELETED,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getTableFilesDeleted();
+ });
+ genericMetricGroup.gauge(
+ LAST_TABLE_FILES_APPENDED,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getTableFilesAppended();
+ });
+ genericMetricGroup.gauge(
+ LAST_TABLE_FILES_COMMIT_COMPACTED,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getTableFilesCompacted();
+ });
+ genericMetricGroup.gauge(
+ LAST_CHANGELOG_FILES_APPENDED,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getChangelogFilesAppended();
+ });
+ genericMetricGroup.gauge(
+ LAST_CHANGELOG_FILES_COMMIT_COMPACTED,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getChangelogFilesCompacted();
+ });
+ genericMetricGroup.gauge(
+ LAST_DELTA_RECORDS_APPENDED,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getDeltaRecordsAppended();
+ });
+ genericMetricGroup.gauge(
+ LAST_CHANGELOG_RECORDS_APPENDED,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getChangelogRecordsAppended();
+ });
+ genericMetricGroup.gauge(
+ LAST_DELTA_RECORDS_COMMIT_COMPACTED,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getDeltaRecordsCompacted();
+ });
+ genericMetricGroup.gauge(
+ LAST_CHANGELOG_RECORDS_COMMIT_COMPACTED,
+ () -> {
+ return latestCommit == null ? 0L : latestCommit.getChangelogRecordsCompacted();
+ });
+ totalTableFilesCounter = genericMetricGroup.counter(TOTAL_TABLE_FILES);
+ totalChangelogFilesCounter = genericMetricGroup.counter(TOTAL_CHANGELOG_FILES);
+ }
+
+ public void reportCommit(CommitStats commitStats) {
+ latestCommit = commitStats;
+ totalTableFilesCounter.inc(
+ commitStats.getTableFilesAdded() - commitStats.getTableFilesDeleted());
+ totalChangelogFilesCounter.inc(
+ commitStats.getChangelogFilesAppended() + commitStats.getChangelogFilesCompacted());
Review Comment:
In PIP-3 you say that `totalTableFiles` is the "umber of total data files currently maintained on storage". However this implementation only counts the number of table files created during this job.
##########
paimon-core/src/main/java/org/apache/paimon/metrics/commit/CommitStats.java:
##########
@@ -0,0 +1,213 @@
+/*
+ * 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.paimon.metrics.commit;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Statistics for a commit. */
+public class CommitStats {
+ private final long duration;
+ private final long attempts;
+ private final long tableFilesAdded;
+ private final long tableFilesAppended;
+ private final long tableFilesDeleted;
+ private final long changelogFilesAppended;
+ private final long changelogFilesCompacted;
+ private final long changelogRecordsCompacted;
+
+ private final long deltaRecordsCompacted;
+ private final long changelogRecordsAppended;
+ private final long deltaRecordsAppended;
+ private final long tableFilesCompacted;
+ private final long generatedSnapshots;
+ private final long numPartitionsWritten;
+ private final long numBucketsWritten;
+ private final Map<BinaryRow, Set<Integer>> bucketsWritten;
+
+ public CommitStats(
+ List<ManifestEntry> appendTableFiles,
+ List<ManifestEntry> appendChangelogFiles,
+ List<ManifestEntry> compactTableFiles,
+ List<ManifestEntry> compactChangelogFiles,
+ long commitDuration,
+ int generatedSnapshots,
+ long attempts) {
+ List<ManifestEntry> addedTableFiles = new ArrayList<>(appendTableFiles);
+ addedTableFiles.addAll(
+ compactTableFiles.stream()
+ .filter(f -> FileKind.ADD.equals(f.kind()))
+ .collect(Collectors.toList()));
+ List<ManifestEntry> deletedTableFiles =
+ new ArrayList<>(
+ compactTableFiles.stream()
+ .filter(f -> FileKind.DELETE.equals(f.kind()))
+ .collect(Collectors.toList()));
Review Comment:
`Collectors.toList` will create an `ArrayList`. No need to create another one.
##########
paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java:
##########
@@ -160,7 +161,8 @@ public FileStoreCommitImpl newCommit(String commitUser) {
options.manifestFullCompactionThresholdSize(),
options.manifestMergeMinCount(),
partitionType.getFieldCount() > 0 && options.dynamicPartitionOverwrite(),
- newKeyComparator());
+ newKeyComparator(),
+ new CommitMetrics(pathFactory()));
Review Comment:
Why create `CommitMetrics` here? `CommitMetrics` is only used inside `FileStoreCommitImpl`. Users do not need to know its existence.
##########
paimon-core/src/main/java/org/apache/paimon/metrics/commit/CommitStats.java:
##########
@@ -0,0 +1,213 @@
+/*
+ * 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.paimon.metrics.commit;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Statistics for a commit. */
+public class CommitStats {
+ private final long duration;
+ private final long attempts;
+ private final long tableFilesAdded;
+ private final long tableFilesAppended;
+ private final long tableFilesDeleted;
+ private final long changelogFilesAppended;
+ private final long changelogFilesCompacted;
+ private final long changelogRecordsCompacted;
+
+ private final long deltaRecordsCompacted;
+ private final long changelogRecordsAppended;
+ private final long deltaRecordsAppended;
+ private final long tableFilesCompacted;
+ private final long generatedSnapshots;
+ private final long numPartitionsWritten;
+ private final long numBucketsWritten;
+ private final Map<BinaryRow, Set<Integer>> bucketsWritten;
+
+ public CommitStats(
+ List<ManifestEntry> appendTableFiles,
+ List<ManifestEntry> appendChangelogFiles,
+ List<ManifestEntry> compactTableFiles,
+ List<ManifestEntry> compactChangelogFiles,
+ long commitDuration,
+ int generatedSnapshots,
+ long attempts) {
+ List<ManifestEntry> addedTableFiles = new ArrayList<>(appendTableFiles);
+ addedTableFiles.addAll(
+ compactTableFiles.stream()
+ .filter(f -> FileKind.ADD.equals(f.kind()))
+ .collect(Collectors.toList()));
+ List<ManifestEntry> deletedTableFiles =
+ new ArrayList<>(
+ compactTableFiles.stream()
+ .filter(f -> FileKind.DELETE.equals(f.kind()))
+ .collect(Collectors.toList()));
+
+ this.tableFilesAdded = addedTableFiles.size();
+ this.tableFilesAppended = appendTableFiles.size();
+ this.tableFilesDeleted = deletedTableFiles.size();
+ this.tableFilesCompacted = compactTableFiles.size();
+ this.changelogFilesAppended = appendChangelogFiles.size();
+ this.changelogFilesCompacted = compactChangelogFiles.size();
+ this.numPartitionsWritten = numChangedPartitions(appendTableFiles, compactTableFiles);
+ this.numBucketsWritten = numChangedBuckets(appendTableFiles, compactTableFiles);
+ this.bucketsWritten = changedPartBuckets(appendTableFiles, compactTableFiles);
+ this.changelogRecordsCompacted = getRowCounts(compactChangelogFiles);
+ this.deltaRecordsCompacted = getRowCounts(compactTableFiles);
+ this.changelogRecordsAppended = getRowCounts(appendChangelogFiles);
+ this.deltaRecordsAppended = getRowCounts(appendTableFiles);
+ this.duration = commitDuration;
+ this.generatedSnapshots = generatedSnapshots;
+ this.attempts = attempts;
+ }
+
+ @VisibleForTesting
+ protected static long numChangedPartitions(List<ManifestEntry>... changes) {
+ return Arrays.stream(changes)
+ .flatMap(Collection::stream)
+ .map(ManifestEntry::partition)
+ .distinct()
+ .collect(Collectors.toList())
+ .size();
Review Comment:
```suggestion
return Arrays.stream(changes)
.flatMap(Collection::stream)
.map(ManifestEntry::partition)
.distinct()
.count();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org