You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "lowka (via GitHub)" <gi...@apache.org> on 2024/08/21 14:11:35 UTC
[PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
lowka opened a new pull request, #4256:
URL: https://github.com/apache/ignite-3/pull/4256
---
Thank you for submitting the pull request.
To streamline the review process of the patch and ensure better code quality
we ask both an author and a reviewer to verify the following:
### The Review Checklist
- [ ] **Formal criteria:** TC status, codestyle, mandatory documentation. Also make sure to complete the following:
\- There is a single JIRA ticket related to the pull request.
\- The web-link to the pull request is attached to the JIRA ticket.
\- The JIRA ticket has the Patch Available state.
\- The description of the JIRA ticket explains WHAT was made, WHY and HOW.
\- The pull request title is treated as the final commit message. The following pattern must be used: IGNITE-XXXX Change summary where XXXX - number of JIRA issue.
- [ ] **Design:** new code conforms with the design principles of the components it is added to.
- [ ] **Patch quality:** patch cannot be split into smaller pieces, its size must be reasonable.
- [ ] **Code quality:** code is clean and readable, necessary developer documentation is added if needed.
- [ ] **Tests code quality:** test set covers positive/negative scenarios, happy/edge cases. Tests are effective in terms of execution time and resources.
### Notes
- [Apache Ignite Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Java+Code+Style+Guide)
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734818539
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -143,6 +143,9 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
*/
private volatile long minActiveTxBeginTime = UNDEFINED_MIN_TX_TIME;
+ /** {@link #minActiveTxBeginTime} stored for the latest data storage flush. */
+ private volatile long snapshottedMinActiveTxBeginTime = UNDEFINED_MIN_TX_TIME;
Review Comment:
Thanks. Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1761250184
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -856,4 +860,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
Review Comment:
> we need to collect timestamps from ALL replication groups (primary and non-primary), can you clarify is this possible to do with proposed approach?
Unfortunately no. And I've missed that, so please ignore my comment.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1774543917
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogManagerCompactionFacade.java:
##########
@@ -84,6 +86,22 @@ Int2IntMap collectTablesWithPartitionsBetween(long minTsInclusive, long maxTsInc
}
}
+ /**
+ * Returns the catalog version that is active at the given timestamp.
+ *
+ * @param timestamp Timestamp.
+ * @return Catalog revision or {@code null} if such version of the catalog doesn't exist.
+ */
+ @Nullable Catalog catalogAtTsNullable(long timestamp) {
Review Comment:
This method should be called `catalogAtTs()` it should never return null, because its invocations always follow not null checks.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734730578
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
@lowka
The idea of the test is to check that 'minActiveTxBeginTime` on all partition listeners updated, not **snapshotted** minActiveTxTime.
And the original `minActiveTxBeginTime` (not `snapshottedMinActiveTxBeginTime`) was required ONLY for this test.
Since you removed this field (renamed `snapshottedMinActiveTxBeginTime` to `minActiveTxBeginTime` ) and since now we have full integration test for catalog compaction runner, I suggest to remove `testRaftGroupsUpdate` because it will not work properly until fully reworked, and, again seems we already have full it-test for this .
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732267948
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
As I see the behavior on coordinator must be consistent with remotes.
Remote nodes report MIN_VALUE to prevent compaction. Coordinator should do the same.
Otherwise, this may lead to issue with single node cluster, for example.
May be it's better to move this to `getMinLocalTime`
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1764413261
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -856,4 +860,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
+ Class<G> groupType,
+ Class<L> groupListenerType
+ ) {
+ return new StateProviderImpl<>(groupType, groupListenerType, nodes);
+ }
+
+ private static class StateProviderImpl<G extends ReplicationGroupId, L> implements RaftGroupStateProvider<G, L> {
Review Comment:
@sanpwc Sounds reasonable. But could you please clarify who is going to cleanup this map and when?
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1761388828
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -714,10 +713,18 @@ private void handleUpdateMinimalActiveTxTimeCommand(UpdateMinimumActiveTxBeginTi
}
long minActiveTxBeginTime0 = minActiveTxBeginTime;
+ long timestamp = cmd.timestamp();
- assert minActiveTxBeginTime0 <= cmd.timestamp() : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + cmd.timestamp();
+ assert minActiveTxBeginTime0 <= timestamp : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + timestamp;
- minActiveTxBeginTime = cmd.timestamp();
+ storage.flush(false).whenComplete((r, t) -> {
Review Comment:
The fact that within command processing in raft state machine we
- may switch threads
- call long running operations
- skip exception handling
smells.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1777992667
##########
modules/table/src/test/java/org/apache/ignite/internal/table/MinimumRequiredTimeCollectorServiceSelfTest.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.ignite.internal.table;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorService;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorServiceImpl;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link MinimumRequiredTimeCollectorServiceImpl}.
+ */
+public class MinimumRequiredTimeCollectorServiceSelfTest extends BaseIgniteAbstractTest {
+
+ private static final long UNDEFINED_MIN_TIME = MinimumRequiredTimeCollectorService.UNDEFINED_MIN_TIME;
+
+ @Test
+ public void test() {
+ MinimumRequiredTimeCollectorServiceImpl collectorService = new MinimumRequiredTimeCollectorServiceImpl();
+
+ // No partitions
+ assertEquals(Map.of(), collectorService.minTimestampPerPartition());
Review Comment:
Fixed.
##########
modules/table/src/test/java/org/apache/ignite/internal/table/MinimumRequiredTimeCollectorServiceSelfTest.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.ignite.internal.table;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorService;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorServiceImpl;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link MinimumRequiredTimeCollectorServiceImpl}.
+ */
+public class MinimumRequiredTimeCollectorServiceSelfTest extends BaseIgniteAbstractTest {
+
+ private static final long UNDEFINED_MIN_TIME = MinimumRequiredTimeCollectorService.UNDEFINED_MIN_TIME;
Review Comment:
Done.
##########
modules/table/src/test/java/org/apache/ignite/internal/table/MinimumRequiredTimeCollectorServiceSelfTest.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.ignite.internal.table;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorService;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorServiceImpl;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link MinimumRequiredTimeCollectorServiceImpl}.
+ */
+public class MinimumRequiredTimeCollectorServiceSelfTest extends BaseIgniteAbstractTest {
+
Review Comment:
Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1774543917
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogManagerCompactionFacade.java:
##########
@@ -84,6 +86,22 @@ Int2IntMap collectTablesWithPartitionsBetween(long minTsInclusive, long maxTsInc
}
}
+ /**
+ * Returns the catalog version that is active at the given timestamp.
+ *
+ * @param timestamp Timestamp.
+ * @return Catalog revision or {@code null} if such version of the catalog doesn't exist.
+ */
+ @Nullable Catalog catalogAtTsNullable(long timestamp) {
Review Comment:
This method should be called `catalogAtTs()` it should never return null, because its invocations are always followed by not null checks.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732746216
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,62 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
+ }
+
+ return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinRequiredTime)
.thenComposeAsync(timeHolder -> {
long minRequiredTime = timeHolder.minRequiredTime;
long minActiveTxBeginTime = timeHolder.minActiveTxBeginTime;
+
Catalog catalog = catalogManagerFacade.catalogByTsNullable(minRequiredTime);
CompletableFuture<Boolean> catalogCompactionFut;
if (catalog == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", minRequiredTime);
+ LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}]. No catalog at minRequiredTime",
Review Comment:
Fixed.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,62 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
Review Comment:
Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732717034
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,62 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
+ }
+
+ return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinRequiredTime)
.thenComposeAsync(timeHolder -> {
long minRequiredTime = timeHolder.minRequiredTime;
long minActiveTxBeginTime = timeHolder.minActiveTxBeginTime;
+
Catalog catalog = catalogManagerFacade.catalogByTsNullable(minRequiredTime);
CompletableFuture<Boolean> catalogCompactionFut;
if (catalog == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", minRequiredTime);
+ LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}]. No catalog at minRequiredTime",
Review Comment:
pay attention to the @korlov42 comment
> Besides, every log message [must end with a period](https://cwiki.apache.org/confluence/display/IGNITE/Java+Code+Style+Guide#JavaCodeStyleGuide-1.4Punctuation).
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,62 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
Review Comment:
Also if this map is huge enough toString() may take some time even if debug is not enabled.
Also you missing period at the end of the sentence.
I suggest not to log ALL partitions and add period to the end of message.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,62 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
Review Comment:
I'm not sure we need dump ALL partitions even on debug level :thinking:
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -299,7 +334,11 @@ private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topology
catalogCompactionFut,
propagateToReplicasFut
);
- }, executor);
+ }, executor).whenComplete((r, t) -> {
Review Comment:
My main point was that this handler should be removed. Because we already have such logging in other places.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
How this was fixed?
Why not to return MIN_VALUE instead of null from `getMinLocalTime`?
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734745239
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
@lowka
The idea of the test testRaftGroupsUpdate is to check that 'minActiveTxBeginTime` on all partition listeners updated, not **snapshotted** minActiveTxTime.
And the original minActiveTxBeginTime (not snapshottedMinActiveTxBeginTime) was required ONLY for this test.
Since now we have full integration test for catalog compaction runner, I suggest to remove testRaftGroupsUpdate because it will not work properly.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732475312
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
This is a mere placeholder that is going to be replaced in determineGlobalMinimumRequiredTime, this value is not sent anywhere.
There is check below that deal with that
```
assert minRequiredTime != Long.MAX_VALUE : "Unexpected minRequiredTime is returned from determineGlobalMin call";
```
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1741472938
##########
modules/catalog-compaction/src/test/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunnerSelfTest.java:
##########
@@ -163,9 +185,12 @@ public void routineSucceedOnCoordinator() throws InterruptedException {
// Nothing should be changed if previous catalog doesn't exists.
Catalog earliestCatalog = Objects.requireNonNull(catalogManager.catalog(catalogManager.earliestCatalogVersion()));
compactionRunner = createRunner(NODE1, NODE1, (n) -> earliestCatalog.time());
- compactionRunner.triggerCompaction(clockService.now());
+
+ HybridTimestamp now = clockService.now();
+ compactionRunner.onLowWatermarkChanged(now);
+ compactionRunner.triggerCompaction(now);
Review Comment:
Done.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1758317160
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -856,4 +860,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
Review Comment:
@sanpwc we need to collect timestamps from ALL replication groups (primary and non-primary), can you clarify is this possible to do with proposed approach?
Can you briefly describe how to do this?
Currently we send a request to PartitionReplicaListener and apply the raft command, and as far as I remember we don't even know when the command will be applied to ALL replicas in the group (we only know when it will be applied to the primary replica) :thinking: .
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1761417556
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -856,4 +860,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
+ Class<G> groupType,
+ Class<L> groupListenerType
+ ) {
+ return new StateProviderImpl<>(groupType, groupListenerType, nodes);
+ }
+
+ private static class StateProviderImpl<G extends ReplicationGroupId, L> implements RaftGroupStateProvider<G, L> {
+
+ private final Class<G> groupType;
+
+ private final Class<L> listenerType;
+
+ private final Map<RaftNodeId, RaftGroupService> nodes;
+
+ StateProviderImpl(
+ Class<G> groupType,
+ Class<L> listenerType,
+ Map<RaftNodeId, RaftGroupService> nodes
+ ) {
+ this.groupType = groupType;
+ this.listenerType = listenerType;
+ this.nodes = nodes;
+ }
+
+ @Override
+ public <V> Map<G, V> getState(Function<L, V> stateFunc) {
+ Map<G, V> result = new HashMap<>();
+
+ for (Map.Entry<RaftNodeId, RaftGroupService> entry : nodes.entrySet()) {
Review Comment:
You are going to match retrieved result with full partitions set in order to check whether all partitions are taken into consideration, right?
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1773597789
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -335,9 +358,11 @@ CompletableFuture<TimeHolder> determineGlobalMinimumRequiredTime(
if (response.minimumActiveTxTime() < globalMinimumActiveTxTime) {
globalMinimumActiveTxTime = response.minimumActiveTxTime();
}
+
+ remotePartitions.put(nodeId, availablePartitionListToMap(response.partitions()));
}
- return new TimeHolder(globalMinimumRequiredTime, globalMinimumActiveTxTime);
+ return new TimeHolder(globalMinimumRequiredTime, globalMinimumActiveTxTime, remotePartitions);
Review Comment:
Fixed.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -714,10 +713,18 @@ private void handleUpdateMinimalActiveTxTimeCommand(UpdateMinimumActiveTxBeginTi
}
long minActiveTxBeginTime0 = minActiveTxBeginTime;
+ long timestamp = cmd.timestamp();
- assert minActiveTxBeginTime0 <= cmd.timestamp() : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + cmd.timestamp();
+ assert minActiveTxBeginTime0 <= timestamp : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + timestamp;
- minActiveTxBeginTime = cmd.timestamp();
+ storage.flush(false).whenComplete((r, t) -> {
+ synchronized (this) {
+ long minActiveTxBeginTime1 = minActiveTxBeginTime;
Review Comment:
Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "korlov42 (via GitHub)" <gi...@apache.org>.
korlov42 commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1752082305
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -714,10 +713,18 @@ private void handleUpdateMinimalActiveTxTimeCommand(UpdateMinimumActiveTxBeginTi
}
long minActiveTxBeginTime0 = minActiveTxBeginTime;
+ long timestamp = cmd.timestamp();
- assert minActiveTxBeginTime0 <= cmd.timestamp() : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + cmd.timestamp();
+ assert minActiveTxBeginTime0 <= timestamp : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + timestamp;
- minActiveTxBeginTime = cmd.timestamp();
+ storage.flush(false).whenComplete((r, t) -> {
+ synchronized (this) {
+ long minActiveTxBeginTime1 = minActiveTxBeginTime;
Review Comment:
why do we need `minActiveTxBeginTime1`? It used only once, so I believe we may just use `minActiveTxBeginTime`
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -335,9 +358,11 @@ CompletableFuture<TimeHolder> determineGlobalMinimumRequiredTime(
if (response.minimumActiveTxTime() < globalMinimumActiveTxTime) {
globalMinimumActiveTxTime = response.minimumActiveTxTime();
}
+
+ remotePartitions.put(nodeId, availablePartitionListToMap(response.partitions()));
}
- return new TimeHolder(globalMinimumRequiredTime, globalMinimumActiveTxTime);
+ return new TimeHolder(globalMinimumRequiredTime, globalMinimumActiveTxTime, remotePartitions);
Review Comment:
at this point we have _global_ minimum for required time and active tx time, but for some reason only remote partitions... let's add state of local partitions to complete the collection
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -387,24 +435,61 @@ private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTop
}
return catalogManagerFacade.compactCatalog(catalog.version());
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ LOG.warn("Catalog compaction has failed [timestamp={}].", ex, minRequiredTime);
+ } else {
+ if (res) {
+ LOG.info("Catalog compaction completed successfully [timestamp={}].", minRequiredTime);
+ } else {
+ LOG.info("Catalog compaction skipped [timestamp={}].", minRequiredTime);
+ }
+ }
});
}
- private CompletableFuture<Set<String>> requiredNodes(Catalog catalog) {
+ private CompletableFuture<Pair<Boolean, Set<String>>> validatePartitions(
+ Catalog catalog,
+ Map<Integer, BitSet> localPartitions,
+ Map<String, Map<Integer, BitSet>> remotePartitions
+ ) {
HybridTimestamp nowTs = clockService.now();
- Set<String> required = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ ConcurrentHashMap<String, CurrentPartitions> currentPartitionsPerNode = new ConcurrentHashMap<>();
return CompletableFutures.allOf(catalog.tables().stream()
- .map(table -> collectRequiredNodes(catalog, table, required, nowTs))
+ .map(table -> collectRequiredNodes(catalog, table, nowTs, currentPartitionsPerNode))
.collect(Collectors.toList())
- ).thenApply(ignore -> required);
+ ).thenApply(ignore -> {
+
+ Map<String, Map<Integer, BitSet>> all = new HashMap<>(remotePartitions);
+ all.put(localNodeName, localPartitions);
+
+ Set<String> required = currentPartitionsPerNode.keySet();
+
+ for (Map.Entry<String, Map<Integer, BitSet>> node : all.entrySet()) {
+ String nodeId = node.getKey();
+ Map<Integer, BitSet> tables = node.getValue();
+ CurrentPartitions partitionsPerNode = currentPartitionsPerNode.get(nodeId);
Review Comment:
looks like case when `currentPartitionsPerNode` has more nodes than we collected is missing (we need test for this case)
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -387,24 +435,61 @@ private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTop
}
return catalogManagerFacade.compactCatalog(catalog.version());
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ LOG.warn("Catalog compaction has failed [timestamp={}].", ex, minRequiredTime);
+ } else {
+ if (res) {
+ LOG.info("Catalog compaction completed successfully [timestamp={}].", minRequiredTime);
+ } else {
+ LOG.info("Catalog compaction skipped [timestamp={}].", minRequiredTime);
+ }
+ }
});
}
- private CompletableFuture<Set<String>> requiredNodes(Catalog catalog) {
+ private CompletableFuture<Pair<Boolean, Set<String>>> validatePartitions(
+ Catalog catalog,
+ Map<Integer, BitSet> localPartitions,
+ Map<String, Map<Integer, BitSet>> remotePartitions
+ ) {
HybridTimestamp nowTs = clockService.now();
- Set<String> required = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ ConcurrentHashMap<String, CurrentPartitions> currentPartitionsPerNode = new ConcurrentHashMap<>();
return CompletableFutures.allOf(catalog.tables().stream()
- .map(table -> collectRequiredNodes(catalog, table, required, nowTs))
+ .map(table -> collectRequiredNodes(catalog, table, nowTs, currentPartitionsPerNode))
.collect(Collectors.toList())
- ).thenApply(ignore -> required);
+ ).thenApply(ignore -> {
+
+ Map<String, Map<Integer, BitSet>> all = new HashMap<>(remotePartitions);
+ all.put(localNodeName, localPartitions);
+
+ Set<String> required = currentPartitionsPerNode.keySet();
Review Comment:
`currentPartitionsPerNode` is actually `requiredPartitionsPerNode `
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -387,24 +435,61 @@ private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTop
}
return catalogManagerFacade.compactCatalog(catalog.version());
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ LOG.warn("Catalog compaction has failed [timestamp={}].", ex, minRequiredTime);
+ } else {
+ if (res) {
+ LOG.info("Catalog compaction completed successfully [timestamp={}].", minRequiredTime);
+ } else {
+ LOG.info("Catalog compaction skipped [timestamp={}].", minRequiredTime);
+ }
+ }
});
}
- private CompletableFuture<Set<String>> requiredNodes(Catalog catalog) {
+ private CompletableFuture<Pair<Boolean, Set<String>>> validatePartitions(
+ Catalog catalog,
+ Map<Integer, BitSet> localPartitions,
+ Map<String, Map<Integer, BitSet>> remotePartitions
+ ) {
HybridTimestamp nowTs = clockService.now();
- Set<String> required = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ ConcurrentHashMap<String, CurrentPartitions> currentPartitionsPerNode = new ConcurrentHashMap<>();
return CompletableFutures.allOf(catalog.tables().stream()
- .map(table -> collectRequiredNodes(catalog, table, required, nowTs))
+ .map(table -> collectRequiredNodes(catalog, table, nowTs, currentPartitionsPerNode))
.collect(Collectors.toList())
- ).thenApply(ignore -> required);
+ ).thenApply(ignore -> {
+
+ Map<String, Map<Integer, BitSet>> all = new HashMap<>(remotePartitions);
+ all.put(localNodeName, localPartitions);
+
+ Set<String> required = currentPartitionsPerNode.keySet();
+
+ for (Map.Entry<String, Map<Integer, BitSet>> node : all.entrySet()) {
+ String nodeId = node.getKey();
+ Map<Integer, BitSet> tables = node.getValue();
+ CurrentPartitions partitionsPerNode = currentPartitionsPerNode.get(nodeId);
+
+ if (partitionsPerNode == null) {
+ return new Pair<>(false, required);
+ } else {
+ Map<Integer, BitSet> data = partitionsPerNode.data();
+ if (!data.equals(tables)) {
Review Comment:
perhaps, we can soften this check a little bit. We must ensure, that particular node contains all required partitions. However, it may contain more partitions that it was required (due to rebalance the same partition was counted twice on two different nodes)
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -387,24 +435,61 @@ private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTop
}
return catalogManagerFacade.compactCatalog(catalog.version());
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ LOG.warn("Catalog compaction has failed [timestamp={}].", ex, minRequiredTime);
+ } else {
+ if (res) {
+ LOG.info("Catalog compaction completed successfully [timestamp={}].", minRequiredTime);
+ } else {
+ LOG.info("Catalog compaction skipped [timestamp={}].", minRequiredTime);
+ }
+ }
});
}
- private CompletableFuture<Set<String>> requiredNodes(Catalog catalog) {
+ private CompletableFuture<Pair<Boolean, Set<String>>> validatePartitions(
+ Catalog catalog,
+ Map<Integer, BitSet> localPartitions,
+ Map<String, Map<Integer, BitSet>> remotePartitions
+ ) {
HybridTimestamp nowTs = clockService.now();
- Set<String> required = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ ConcurrentHashMap<String, CurrentPartitions> currentPartitionsPerNode = new ConcurrentHashMap<>();
return CompletableFutures.allOf(catalog.tables().stream()
- .map(table -> collectRequiredNodes(catalog, table, required, nowTs))
+ .map(table -> collectRequiredNodes(catalog, table, nowTs, currentPartitionsPerNode))
.collect(Collectors.toList())
- ).thenApply(ignore -> required);
+ ).thenApply(ignore -> {
+
+ Map<String, Map<Integer, BitSet>> all = new HashMap<>(remotePartitions);
+ all.put(localNodeName, localPartitions);
+
+ Set<String> required = currentPartitionsPerNode.keySet();
+
+ for (Map.Entry<String, Map<Integer, BitSet>> node : all.entrySet()) {
+ String nodeId = node.getKey();
+ Map<Integer, BitSet> tables = node.getValue();
+ CurrentPartitions partitionsPerNode = currentPartitionsPerNode.get(nodeId);
+
+ if (partitionsPerNode == null) {
Review Comment:
`allPartitions` will likely have results from more nodes than it was `required`, because not all nodes may own partitions (do we have test for this case?)
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732475312
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
This is a mere placeholder that is going to be replaced in determineGlobalMinimumRequiredTime.
There is check below that deal with that
```
assert minRequiredTime != Long.MAX_VALUE : "Unexpected minRequiredTime is returned from determineGlobalMin call";
```
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
This is a mere placeholder that is going to be replaced in determineGlobalMinimumRequiredTime.
There is check below that deal with that
```
assert minRequiredTime != Long.MAX_VALUE : "Unexpected minRequiredTime is returned from determineGlobalMin call";
```
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734730578
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
@lowka
The idea of the test is to check that 'minActiveTxBeginTime` on all partition listeners updated, not **snapshotted** minActiveTxTime.
And the original `minActiveTxBeginTime` (not `snapshottedMinActiveTxBeginTime`) was required ONLY for this test.
Since you removed this field (renamed `snapshottedMinActiveTxBeginTime` to `minActiveTxBeginTime` ) and since now we have full integration test for catalog compaction runner, I suggest to remove `testRaftGroupsUpdate` because it will not work properly until fully reworked, and, again we already have full it-test for this .
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734730578
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
@lowka
The idea of the test `testRaftGroupsUpdate` is to check that 'minActiveTxBeginTime` on all partition listeners updated, not **snapshotted** minActiveTxTime.
And the original `minActiveTxBeginTime` (not `snapshottedMinActiveTxBeginTime`) was required ONLY for this test.
Since you removed this field (renamed `snapshottedMinActiveTxBeginTime` to `minActiveTxBeginTime` ) and since now we have full integration test for catalog compaction runner, I suggest to remove `testRaftGroupsUpdate` because it will not work properly until fully reworked, and, again seems we already have test for this .
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734745239
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
@lowka
The idea of the test testRaftGroupsUpdate is to check that 'minActiveTxBeginTime` on all partition listeners updated, not **snapshotted** minActiveTxTime.
And the original minActiveTxBeginTime (not snapshottedMinActiveTxBeginTime) was required ONLY for this test.
Since now we have full integration test for catalog compaction runner, I suggest to remove testRaftGroupsUpdate because it will not work properly until reworked.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1773598537
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -714,10 +713,18 @@ private void handleUpdateMinimalActiveTxTimeCommand(UpdateMinimumActiveTxBeginTi
}
long minActiveTxBeginTime0 = minActiveTxBeginTime;
+ long timestamp = cmd.timestamp();
- assert minActiveTxBeginTime0 <= cmd.timestamp() : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + cmd.timestamp();
+ assert minActiveTxBeginTime0 <= timestamp : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + timestamp;
- minActiveTxBeginTime = cmd.timestamp();
+ storage.flush(false).whenComplete((r, t) -> {
Review Comment:
Fixed.
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -856,4 +860,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
+ Class<G> groupType,
+ Class<L> groupListenerType
+ ) {
+ return new StateProviderImpl<>(groupType, groupListenerType, nodes);
+ }
+
+ private static class StateProviderImpl<G extends ReplicationGroupId, L> implements RaftGroupStateProvider<G, L> {
+
+ private final Class<G> groupType;
+
+ private final Class<L> listenerType;
+
+ private final Map<RaftNodeId, RaftGroupService> nodes;
+
+ StateProviderImpl(
+ Class<G> groupType,
+ Class<L> listenerType,
+ Map<RaftNodeId, RaftGroupService> nodes
+ ) {
+ this.groupType = groupType;
+ this.listenerType = listenerType;
+ this.nodes = nodes;
+ }
+
+ @Override
+ public <V> Map<G, V> getState(Function<L, V> stateFunc) {
+ Map<G, V> result = new HashMap<>();
+
+ for (Map.Entry<RaftNodeId, RaftGroupService> entry : nodes.entrySet()) {
Review Comment:
Removed in favour of `MinimumRequiredTimeCollectorService`.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1773604514
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -856,4 +860,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
+ Class<G> groupType,
+ Class<L> groupListenerType
+ ) {
+ return new StateProviderImpl<>(groupType, groupListenerType, nodes);
+ }
+
+ private static class StateProviderImpl<G extends ReplicationGroupId, L> implements RaftGroupStateProvider<G, L> {
Review Comment:
Removed modifications of raft-code in favour of `MinimumRequiredTimeCollectorService `. that stores `TablePartitionId` -> `timestamp`
- When a partition is added it adds an entry to that map.
- its updates these entries when needed.
- When a partition is destroyed it removes its entry from the map.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1761389825
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -714,10 +713,18 @@ private void handleUpdateMinimalActiveTxTimeCommand(UpdateMinimumActiveTxBeginTi
}
long minActiveTxBeginTime0 = minActiveTxBeginTime;
+ long timestamp = cmd.timestamp();
- assert minActiveTxBeginTime0 <= cmd.timestamp() : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + cmd.timestamp();
+ assert minActiveTxBeginTime0 <= timestamp : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + timestamp;
- minActiveTxBeginTime = cmd.timestamp();
+ storage.flush(false).whenComplete((r, t) -> {
Review Comment:
Is that intended that only one storage is flushed?
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1777994608
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorService.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public interface MinimumRequiredTimeCollectorService {
+
+ /** Undefined value of a min timestamp. */
+ long UNDEFINED_MIN_TIME = 0;
+
+ /** Registers a partition. */
+ void addPartition(TablePartitionId tablePartitionId);
+
+ /** Records the given timestamp .*/
+ void recordMinActiveTxTimestamp(TablePartitionId tablePartitionId, long timestamp);
+
+ /** Remove timestamps associated with the given partition .*/
+ void removePartition(TablePartitionId tablePartitionId);
+
+ /** Returns a snapshot of collected timestamps. */
+ Map<TablePartitionId, Long> minTimestampPerPartition();
Review Comment:
This component only deals with `TablePartitionId` and callers of this method check their `TablePartitionId`s with results of this method.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorService.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public interface MinimumRequiredTimeCollectorService {
+
+ /** Undefined value of a min timestamp. */
+ long UNDEFINED_MIN_TIME = 0;
+
+ /** Registers a partition. */
+ void addPartition(TablePartitionId tablePartitionId);
+
+ /** Records the given timestamp .*/
+ void recordMinActiveTxTimestamp(TablePartitionId tablePartitionId, long timestamp);
+
+ /** Remove timestamps associated with the given partition .*/
+ void removePartition(TablePartitionId tablePartitionId);
+
+ /** Returns a snapshot of collected timestamps. */
+ Map<TablePartitionId, Long> minTimestampPerPartition();
+
+ /** Closes this service. */
+ void close();
Review Comment:
We should. Done.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorService.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public interface MinimumRequiredTimeCollectorService {
+
Review Comment:
Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1778233954
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorService.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public interface MinimumRequiredTimeCollectorService {
+
+ /** Undefined value of a min timestamp. */
+ long UNDEFINED_MIN_TIME = 0;
+
+ /** Registers a partition. */
+ void addPartition(TablePartitionId tablePartitionId);
+
+ /** Records the given timestamp .*/
+ void recordMinActiveTxTimestamp(TablePartitionId tablePartitionId, long timestamp);
+
+ /** Remove timestamps associated with the given partition .*/
+ void removePartition(TablePartitionId tablePartitionId);
+
+ /** Returns a snapshot of collected timestamps. */
+ Map<TablePartitionId, Long> minTimestampPerPartition();
Review Comment:
I mean that within collocation track zones instead of tables will manage paritions, thus ZonePartitionId will be used. So that if it's possible to use ReplicationGroupId in the component, it's better to do that.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1777349331
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorService.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public interface MinimumRequiredTimeCollectorService {
+
Review Comment:
Unnecessary blank line.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorServiceImpl.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public class MinimumRequiredTimeCollectorServiceImpl implements MinimumRequiredTimeCollectorService {
+
+ private final ConcurrentHashMap<TablePartitionId, Long> partitions = new ConcurrentHashMap<>();
+
+ /** {@inheritDoc} */
+ @Override
+ public synchronized void addPartition(TablePartitionId tablePartitionId) {
+ partitions.put(tablePartitionId, UNDEFINED_MIN_TIME);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public synchronized void recordMinActiveTxTimestamp(TablePartitionId tablePartitionId, long timestamp) {
+ Long time = partitions.get(tablePartitionId);
+ if (time == null) {
+ // Ignore removed partitions.
+ return;
+ }
+
+ if (time == UNDEFINED_MIN_TIME || timestamp > time) {
+ partitions.put(tablePartitionId, timestamp);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void removePartition(TablePartitionId tablePartitionId) {
Review Comment:
As mentioned above, I'd rather consider different approach. However, if you will stick with given on - should we mark removePartition as synchronized? Otherwise you'll have a race with recordMinActiveTxTimestamp.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorService.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public interface MinimumRequiredTimeCollectorService {
+
+ /** Undefined value of a min timestamp. */
+ long UNDEFINED_MIN_TIME = 0;
+
+ /** Registers a partition. */
+ void addPartition(TablePartitionId tablePartitionId);
+
+ /** Records the given timestamp .*/
+ void recordMinActiveTxTimestamp(TablePartitionId tablePartitionId, long timestamp);
+
+ /** Remove timestamps associated with the given partition .*/
+ void removePartition(TablePartitionId tablePartitionId);
+
+ /** Returns a snapshot of collected timestamps. */
+ Map<TablePartitionId, Long> minTimestampPerPartition();
+
+ /** Closes this service. */
+ void close();
Review Comment:
Should we mark the interface as ManuallyCloseable?
##########
modules/table/src/test/java/org/apache/ignite/internal/table/MinimumRequiredTimeCollectorServiceSelfTest.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.ignite.internal.table;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorService;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorServiceImpl;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link MinimumRequiredTimeCollectorServiceImpl}.
+ */
+public class MinimumRequiredTimeCollectorServiceSelfTest extends BaseIgniteAbstractTest {
+
Review Comment:
Same as above, unnecessary empty line.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorService.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public interface MinimumRequiredTimeCollectorService {
+
+ /** Undefined value of a min timestamp. */
+ long UNDEFINED_MIN_TIME = 0;
+
+ /** Registers a partition. */
+ void addPartition(TablePartitionId tablePartitionId);
+
+ /** Records the given timestamp .*/
+ void recordMinActiveTxTimestamp(TablePartitionId tablePartitionId, long timestamp);
+
+ /** Remove timestamps associated with the given partition .*/
+ void removePartition(TablePartitionId tablePartitionId);
Review Comment:
Don't you think that adjustWatermark(watermark) will be simper?
I mean that every partition may recordMinActiveTxTimestamp without previous partition registration (addPartition()). CleanupService (I don't remember real name) on successful cleanup will call adjustWatermark(cleanupTimestamp) that will remove all map entires with timestamp <=cleanupTimestamp. In that case removePartition is also no longer needed.
Or it's somehow related to awaiting all partitions readiness?
##########
modules/table/src/test/java/org/apache/ignite/internal/table/MinimumRequiredTimeCollectorServiceSelfTest.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.ignite.internal.table;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorService;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorServiceImpl;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link MinimumRequiredTimeCollectorServiceImpl}.
+ */
+public class MinimumRequiredTimeCollectorServiceSelfTest extends BaseIgniteAbstractTest {
+
+ private static final long UNDEFINED_MIN_TIME = MinimumRequiredTimeCollectorService.UNDEFINED_MIN_TIME;
+
+ @Test
+ public void test() {
+ MinimumRequiredTimeCollectorServiceImpl collectorService = new MinimumRequiredTimeCollectorServiceImpl();
+
+ // No partitions
+ assertEquals(Map.of(), collectorService.minTimestampPerPartition());
Review Comment:
It might be implementation specific. I'd rather assert that size == 0.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorServiceImpl.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public class MinimumRequiredTimeCollectorServiceImpl implements MinimumRequiredTimeCollectorService {
+
Review Comment:
Unnecessary blank line.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorService.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public interface MinimumRequiredTimeCollectorService {
+
+ /** Undefined value of a min timestamp. */
+ long UNDEFINED_MIN_TIME = 0;
+
+ /** Registers a partition. */
+ void addPartition(TablePartitionId tablePartitionId);
+
+ /** Records the given timestamp .*/
+ void recordMinActiveTxTimestamp(TablePartitionId tablePartitionId, long timestamp);
+
+ /** Remove timestamps associated with the given partition .*/
+ void removePartition(TablePartitionId tablePartitionId);
+
+ /** Returns a snapshot of collected timestamps. */
+ Map<TablePartitionId, Long> minTimestampPerPartition();
Review Comment:
Is that really important that it's TablePartitionId or we can use ReplicationGroupId here?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -714,10 +713,18 @@ private void handleUpdateMinimalActiveTxTimeCommand(UpdateMinimumActiveTxBeginTi
}
long minActiveTxBeginTime0 = minActiveTxBeginTime;
+ long timestamp = cmd.timestamp();
- assert minActiveTxBeginTime0 <= cmd.timestamp() : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + cmd.timestamp();
+ assert minActiveTxBeginTime0 <= timestamp : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + timestamp;
- minActiveTxBeginTime = cmd.timestamp();
+ storage.flush(false).whenComplete((r, t) -> {
Review Comment:
I'm talking about TxStateStorage and not other's table PartitionDataStorage here.
##########
modules/table/src/test/java/org/apache/ignite/internal/table/MinimumRequiredTimeCollectorServiceSelfTest.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.ignite.internal.table;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorService;
+import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorServiceImpl;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link MinimumRequiredTimeCollectorServiceImpl}.
+ */
+public class MinimumRequiredTimeCollectorServiceSelfTest extends BaseIgniteAbstractTest {
+
+ private static final long UNDEFINED_MIN_TIME = MinimumRequiredTimeCollectorService.UNDEFINED_MIN_TIME;
Review Comment:
Why not to use static import instead?
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1754932410
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -856,4 +860,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
Review Comment:
I'm not sure that fully get the idea of the provided code. I'd rather implement it in a bit different manner, precisely by adjusting UpdateMinimumActiveTxBeginTimeCommand handling in a way not only to update minActiveTxBeginTime but also to return last flushed one. WDYT?
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732278323
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -601,6 +603,20 @@ public MvPartitionStorage getMvStorage() {
return minActiveTxBeginTime0;
}
+ /**
+ * Returns minimum starting time among all active RW transactions in the cluster,
+ * or {@code null} if the value has not yet been set.
+ */
+ public @Nullable Long snapshottedMinActiveTxBeginTime() {
Review Comment:
Not sure but may be it worth somehow note the difference between `snapshottedMinActiveTxBeginTime` and `minimumActiveTxBeginTime` in javadocs.
Currently javadocs are identical :thinking:
Also I suggest to move non-test method upper then methods marked TestOnly.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -299,7 +334,11 @@ private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topology
catalogCompactionFut,
propagateToReplicasFut
);
- }, executor);
+ }, executor).whenComplete((r, t) -> {
Review Comment:
We have 2 separated processes:
1. catalogCompactionFut already has error handler with logging
2. propagateToReplicasFut - already has error handler with logging
Why we need this additional logging?
And what "startCompaction launched (time) has failed" means? :thinking:
This "combined" future was left for testing purposes
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
As I see the behavior on coordinator must be consistent with remotes.
Remotes report MIN_VALUE to prevent compaction. Coordinator should do the same.
Otherwise, this can lead to issue with single node cluster, for example.
May be it's better to move this to `getMinLocalTime`
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732728826
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
> How this was fixed?
I see now (localMinRequiredTime can be `null`), but then this `if condition` should be also removed
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
> How this was fixed?
I see now (localMinRequiredTime can be `null`), but this `if condition` should be also removed
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1761437347
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -856,4 +860,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
+ Class<G> groupType,
+ Class<L> groupListenerType
+ ) {
+ return new StateProviderImpl<>(groupType, groupListenerType, nodes);
+ }
+
+ private static class StateProviderImpl<G extends ReplicationGroupId, L> implements RaftGroupStateProvider<G, L> {
Review Comment:
Still seems a bit too complicated to me.
Why not to introduce new service MinimumRequiredTimeCollectorService, instantiate it in IgniteImpl and propagate to TableManager -> PartitionListener, similar to what you've did with MinimumRequiredTimeProvider? Internally MinimumRequiredTimeCollectorService will be a map partitionId -> minTxStartTime, where all partitionListeneners will put (localId, minTxStartTime) on handleUpdateMinimalActiveTxTimeCommand.
Seems simpler, though I can't say that it's beautiful.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732475312
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
This is a mere placeholder that is going to be replaced in determineGlobalMinimumRequiredTime, this value is not sent anywhere.
There is check below that deals with that
```
assert minRequiredTime != Long.MAX_VALUE : "Unexpected minRequiredTime is returned from determineGlobalMin call";
```
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734613284
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -370,12 +394,16 @@ CompletableFuture<Void> propagateTimeToLocalReplicas(long txBeginTime) {
}, executor);
}
- private long determineLocalMinimumRequiredTime() {
- // TODO https://issues.apache.org/jira/browse/IGNITE-22637 Provide actual minimum required time.
- return HybridTimestamp.MIN_VALUE.longValue();
- }
private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTopologySnapshot topologySnapshot) {
+ for (CatalogIndexDescriptor index : catalog.indexes()) {
+ if (index.status() == CatalogIndexStatus.BUILDING || index.status() == CatalogIndexStatus.REGISTERED) {
+ LOG.info("Catalog compaction aborted, index construction is taking place.");
Review Comment:
Moved code to the `tryCatalogCompaction`.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1761250184
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -856,4 +860,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
Review Comment:
> we need to collect timestamps from ALL replication groups (primary and non-primary), can you clarify is this possible to do with proposed approach?
Unfortunately no. And I've missed that, so please ignore my comment.
Can you please show a place where you evaluate lastSnapshotted(flushed) tx start time?
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1773604952
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -714,10 +713,18 @@ private void handleUpdateMinimalActiveTxTimeCommand(UpdateMinimumActiveTxBeginTi
}
long minActiveTxBeginTime0 = minActiveTxBeginTime;
+ long timestamp = cmd.timestamp();
- assert minActiveTxBeginTime0 <= cmd.timestamp() : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + cmd.timestamp();
+ assert minActiveTxBeginTime0 <= timestamp : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + timestamp;
- minActiveTxBeginTime = cmd.timestamp();
+ storage.flush(false).whenComplete((r, t) -> {
Review Comment:
Yes, it is. As different table may use different storages, each storage is checked (because this method is going to be called for all partitions).
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "korlov42 (via GitHub)" <gi...@apache.org>.
korlov42 commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734398680
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,57 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}].", e.getKey());
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}].",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}].", lwm);
Review Comment:
```suggestion
LOG.info("Catalog compaction started [lowWaterMark={}].", lwm);
```
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,57 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}].", e.getKey());
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}].",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}].", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinRequiredTime)
.thenComposeAsync(timeHolder -> {
long minRequiredTime = timeHolder.minRequiredTime;
long minActiveTxBeginTime = timeHolder.minActiveTxBeginTime;
+
Catalog catalog = catalogManagerFacade.catalogByTsNullable(minRequiredTime);
CompletableFuture<Boolean> catalogCompactionFut;
if (catalog == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", minRequiredTime);
+ LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}]. No catalog at minRequiredTime.",
Review Comment:
```suggestion
LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].",
```
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,57 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}].", e.getKey());
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}].",
Review Comment:
```suggestion
LOG.debug("Minimum required time was chosen [partitionMinTime={}, lowWatermark={}, chosen={}].",
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -143,6 +143,9 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
*/
private volatile long minActiveTxBeginTime = UNDEFINED_MIN_TX_TIME;
+ /** {@link #minActiveTxBeginTime} stored for the latest data storage flush. */
+ private volatile long snapshottedMinActiveTxBeginTime = UNDEFINED_MIN_TX_TIME;
Review Comment:
do we really need one more field?
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
why do you set update frequency to such a big interval? I tried to set it to 500ms and test passes 15 times in a row locally, duration is reduced from 1min to ~7sec
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -370,12 +394,16 @@ CompletableFuture<Void> propagateTimeToLocalReplicas(long txBeginTime) {
}, executor);
}
- private long determineLocalMinimumRequiredTime() {
- // TODO https://issues.apache.org/jira/browse/IGNITE-22637 Provide actual minimum required time.
- return HybridTimestamp.MIN_VALUE.longValue();
- }
private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTopologySnapshot topologySnapshot) {
+ for (CatalogIndexDescriptor index : catalog.indexes()) {
+ if (index.status() == CatalogIndexStatus.BUILDING || index.status() == CatalogIndexStatus.REGISTERED) {
+ LOG.info("Catalog compaction aborted, index construction is taking place.");
Review Comment:
I see that we log results both inside and outside of this method. Let's pick one place
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -538,9 +574,9 @@ private void handlePrepareToUpdateTimeOnReplicasMessage(NetworkMessage message)
/** Minimum required time provider. */
@FunctionalInterface
- interface MinimumRequiredTimeProvider {
- /** Returns minimum required timestamp. */
- long time();
+ public interface MinimumRequiredTimeProvider {
Review Comment:
let's move this interface to separate file
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734730578
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
@lowka
The idea of the test `testRaftGroupsUpdate` is to check that 'minActiveTxBeginTime` on all partition listeners updated, not **snapshotted** minActiveTxTime.
And the original `minActiveTxBeginTime` (not `snapshottedMinActiveTxBeginTime`) was required ONLY for this test.
Since you removed this field (renamed `snapshottedMinActiveTxBeginTime` to `minActiveTxBeginTime` ) and since now we have full integration test for catalog compaction runner, I suggest to remove `testRaftGroupsUpdate` because it will not work properly until fully reworked, and, again seems we already have full it-test for this .
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734745239
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
@lowka
The idea of the test testRaftGroupsUpdate is to check that 'minActiveTxBeginTime` on all partition listeners updated, not **snapshotted** minActiveTxTime.
And the original minActiveTxBeginTime (not snapshottedMinActiveTxBeginTime) was required ONLY for this test.
Since we removed this field (renamed snapshottedMinActiveTxBeginTime to minActiveTxBeginTime ) and since now we have full integration test for catalog compaction runner, I suggest to remove testRaftGroupsUpdate because it will not work properly until reworked, and, again seems we already have test for this .
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732728826
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
> How this was fixed?
I see now (localMinRequiredTime can be `null`), but this condition
```java
if (localMinRequiredTime == null) {
localMinRequiredTime = Long.MAX_VALUE;
```
should be also removed
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732763296
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1735968931
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -545,16 +566,29 @@ public void onReceived(NetworkMessage message, ClusterNode sender, @Nullable Lon
private void handleMinimumTimesRequest(ClusterNode sender, Long correlationId) {
HybridTimestamp lwm = lowWatermark;
- Long minRequiredTime = lwm != null ? getMinLocalTime(lwm) : null;
+ Long minRequiredTime;
+ if (lwm != null) {
+ LocalMinTime minLocalTime = getMinLocalTime(lwm);
+ if (minLocalTime != null) {
+ minRequiredTime = minLocalTime.time;
+ } else {
+ minRequiredTime = null;
+ }
+ } else {
+ minRequiredTime = null;
+ }
// We do not have local min time yet. Reply with the absolute min time.
if (minRequiredTime == null) {
minRequiredTime = HybridTimestamp.MIN_VALUE.longValue();
}
+ Map<Integer, BitSet> availablePartitions = buildTablePartitions(localMinTimeProvider.minTimePerPartition());
Review Comment:
This call is unnecessary. I later use `LocalMinTime::availablePartitions`.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1741571821
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -387,24 +437,55 @@ private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTop
}
return catalogManagerFacade.compactCatalog(catalog.version());
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ LOG.warn("Catalog compaction has failed [timestamp={}].", ex, minRequiredTime);
+ } else {
+ if (res) {
+ LOG.info("Catalog compaction completed successfully [timestamp={}].", minRequiredTime);
+ } else {
+ LOG.info("Catalog compaction skipped [timestamp={}].", minRequiredTime);
+ }
+ }
});
}
- private CompletableFuture<Set<String>> requiredNodes(Catalog catalog) {
+ private CompletableFuture<Pair<Boolean, Set<String>>> validatePartitions(
+ Catalog catalog,
+ Map<String, Map<Integer, BitSet>> remotePartitions
+ ) {
HybridTimestamp nowTs = clockService.now();
Set<String> required = Collections.newSetFromMap(new ConcurrentHashMap<>());
Review Comment:
Updated `validatePartitions`'s signature, now this set is being used.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1778351861
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorService.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public interface MinimumRequiredTimeCollectorService {
+
+ /** Undefined value of a min timestamp. */
+ long UNDEFINED_MIN_TIME = 0;
+
+ /** Registers a partition. */
+ void addPartition(TablePartitionId tablePartitionId);
+
+ /** Records the given timestamp .*/
+ void recordMinActiveTxTimestamp(TablePartitionId tablePartitionId, long timestamp);
+
+ /** Remove timestamps associated with the given partition .*/
+ void removePartition(TablePartitionId tablePartitionId);
+
+ /** Returns a snapshot of collected timestamps. */
+ Map<TablePartitionId, Long> minTimestampPerPartition();
Review Comment:
How can I obtain zoneId inside PartitionListener then?
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1761437347
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -856,4 +860,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
+ Class<G> groupType,
+ Class<L> groupListenerType
+ ) {
+ return new StateProviderImpl<>(groupType, groupListenerType, nodes);
+ }
+
+ private static class StateProviderImpl<G extends ReplicationGroupId, L> implements RaftGroupStateProvider<G, L> {
Review Comment:
Still seems a bit too complicated to me.
Why not to introduce new service MinimumRequiredTimeCollectorService, instantiate it in IgniteImpl and propagate to TableManager -> PartitionListener, similar to what you've did with MinimumRequiredTimeProvider.
Internally MinimumRequiredTimeCollectorService will be a map partitionId -> minTxStartTime, where all partitionListeneners will put (localId, minTxStartTime) on handleUpdateMinimalActiveTxTimeCommand.
Seems simpler, though I can't say that it's beautiful.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1731029488
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -704,6 +720,10 @@ private void handleUpdateMinimalActiveTxTimeCommand(UpdateMinimumActiveTxBeginTi
assert minActiveTxBeginTime0 <= cmd.timestamp() : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + cmd.timestamp();
minActiveTxBeginTime = cmd.timestamp();
+
+ storage.flush(false).whenComplete((r, t) -> {
+ snapshottedMinActiveTxBeginTime = minActiveTxBeginTime;
Review Comment:
Thanks. Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732460783
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -299,7 +334,11 @@ private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topology
catalogCompactionFut,
propagateToReplicasFut
);
- }, executor);
+ }, executor).whenComplete((r, t) -> {
Review Comment:
That was a typo. Replaced with `startCompaction launched has failed`.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1739640676
##########
modules/catalog-compaction/src/test/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunnerSelfTest.java:
##########
@@ -163,9 +185,12 @@ public void routineSucceedOnCoordinator() throws InterruptedException {
// Nothing should be changed if previous catalog doesn't exists.
Catalog earliestCatalog = Objects.requireNonNull(catalogManager.catalog(catalogManager.earliestCatalogVersion()));
compactionRunner = createRunner(NODE1, NODE1, (n) -> earliestCatalog.time());
- compactionRunner.triggerCompaction(clockService.now());
+
+ HybridTimestamp now = clockService.now();
+ compactionRunner.onLowWatermarkChanged(now);
+ compactionRunner.triggerCompaction(now);
Review Comment:
```suggestion
```
Why do we need to trigger the compaction twice?
Please, fix it in other places too.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -387,24 +437,55 @@ private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTop
}
return catalogManagerFacade.compactCatalog(catalog.version());
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ LOG.warn("Catalog compaction has failed [timestamp={}].", ex, minRequiredTime);
+ } else {
+ if (res) {
+ LOG.info("Catalog compaction completed successfully [timestamp={}].", minRequiredTime);
+ } else {
+ LOG.info("Catalog compaction skipped [timestamp={}].", minRequiredTime);
+ }
+ }
});
}
- private CompletableFuture<Set<String>> requiredNodes(Catalog catalog) {
+ private CompletableFuture<Pair<Boolean, Set<String>>> validatePartitions(
+ Catalog catalog,
+ Map<String, Map<Integer, BitSet>> remotePartitions
+ ) {
HybridTimestamp nowTs = clockService.now();
Set<String> required = Collections.newSetFromMap(new ConcurrentHashMap<>());
Review Comment:
We don't need this Set anymore, because it can be replaced with `actualPartitions.keySet()`
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "korlov42 (via GitHub)" <gi...@apache.org>.
korlov42 commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1741050516
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
I believe the issue is still here. In `determineGlobalMinimumRequiredTime`, `localMinimumRequiredTime` is used only once to initialize `globalMinimumRequiredTime`, and if there is at least one remote response, null-initialized `globalMinimumRequiredTime` will be re-initialized:
```
if (globalMinimumRequiredTime == null || response.minimumRequiredTime() < globalMinimumRequiredTime) {
globalMinimumRequiredTime = response.minimumRequiredTime();
}
```
Besides, partitions from coordinator are seemed to be completely ignored.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -714,10 +713,15 @@ private void handleUpdateMinimalActiveTxTimeCommand(UpdateMinimumActiveTxBeginTi
}
long minActiveTxBeginTime0 = minActiveTxBeginTime;
+ long timestamp = cmd.timestamp();
- assert minActiveTxBeginTime0 <= cmd.timestamp() : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + cmd.timestamp();
+ assert minActiveTxBeginTime0 <= timestamp : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + timestamp;
- minActiveTxBeginTime = cmd.timestamp();
+ storage.flush(false).whenComplete((r, t) -> {
+ if (timestamp > minActiveTxBeginTime) {
+ minActiveTxBeginTime = timestamp;
+ }
Review Comment:
this doesn't look solid... Do we have guarantee, that there is no other thread updating `minActiveTxBeginTime` at the same time? If no, we need to do check + update atomically
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734587438
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,57 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}].", e.getKey());
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}].",
Review Comment:
Fixed.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,57 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}].", e.getKey());
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}].",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}].", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinRequiredTime)
.thenComposeAsync(timeHolder -> {
long minRequiredTime = timeHolder.minRequiredTime;
long minActiveTxBeginTime = timeHolder.minActiveTxBeginTime;
+
Catalog catalog = catalogManagerFacade.catalogByTsNullable(minRequiredTime);
CompletableFuture<Boolean> catalogCompactionFut;
if (catalog == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", minRequiredTime);
+ LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}]. No catalog at minRequiredTime.",
Review Comment:
Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734818002
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
Done.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1766405572
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -856,4 +860,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
+ Class<G> groupType,
+ Class<L> groupListenerType
+ ) {
+ return new StateProviderImpl<>(groupType, groupListenerType, nodes);
+ }
+
+ private static class StateProviderImpl<G extends ReplicationGroupId, L> implements RaftGroupStateProvider<G, L> {
Review Comment:
I believe that a catalog compactor may clean the whole map on each successful compactisation.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1741547680
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
Fixed `globalMinimumRequiredTime` part.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1741578489
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
It now uses local partitions as well.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1741579631
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -714,10 +713,15 @@ private void handleUpdateMinimalActiveTxTimeCommand(UpdateMinimumActiveTxBeginTi
}
long minActiveTxBeginTime0 = minActiveTxBeginTime;
+ long timestamp = cmd.timestamp();
- assert minActiveTxBeginTime0 <= cmd.timestamp() : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + cmd.timestamp();
+ assert minActiveTxBeginTime0 <= timestamp : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + timestamp;
- minActiveTxBeginTime = cmd.timestamp();
+ storage.flush(false).whenComplete((r, t) -> {
+ if (timestamp > minActiveTxBeginTime) {
+ minActiveTxBeginTime = timestamp;
+ }
Review Comment:
Added `synchronized` block
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1730837941
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -704,6 +720,10 @@ private void handleUpdateMinimalActiveTxTimeCommand(UpdateMinimumActiveTxBeginTi
assert minActiveTxBeginTime0 <= cmd.timestamp() : "maxTime=" + minActiveTxBeginTime0 + ", cmdTime=" + cmd.timestamp();
minActiveTxBeginTime = cmd.timestamp();
+
+ storage.flush(false).whenComplete((r, t) -> {
+ snapshottedMinActiveTxBeginTime = minActiveTxBeginTime;
Review Comment:
flush(false) should guarantee us that if a checkpoint has already started at the moment, then we “subscribe” to the next one.
with this implementation
```
snapshottedMinActiveTxBeginTime = minActiveTxBeginTime;
```
the following situation may occur
1. we subscribed to a checkpoint (minActiveTxBeginTime=N)
2. the checkpoint started
3. a new value arrived minActiveTxBeginTime = N + M
4. we subscribe to the next checkpoint
5. the current checkpoint ends but we write not N but immediately N + M, which can be a problem
Thus, I suggest to rework this a bit.
1. whenComplete listener must use some local value instead of minActiveTxBeginTime (may be we can completely remove minActiveTxBeginTime field).
2. should update snapshottedMinActiveTxBeginTime only if local value is greater then stored in snapshottedMinActiveTxBeginTime (because completablefuture listeners are called in reverse order)
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732267948
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
As I see the behavior on coordinator must be consistent with remotes.
Remote nodes report MIN_VALUE to prevent compaction. Coordinator should do the same.
Otherwise, this can lead to issue with single node cluster, for example.
May be it's better to move this to `getMinLocalTime`
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732456302
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/UpdateLogImpl.java:
##########
@@ -225,6 +225,8 @@ public CompletableFuture<Boolean> saveSnapshot(SnapshotEntry update) {
Iif iif = iif(versionIsRecent, saveSnapshotAndDropOutdatedUpdates, ops().yield(false));
+ LOG.info("Snapshot version is updated. [Old version: {}, new version: {}]", oldSnapshotVersion, snapshotVersion);
Review Comment:
Removed this method.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
+ }
+
+ return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinRequiredTime)
.thenComposeAsync(timeHolder -> {
long minRequiredTime = timeHolder.minRequiredTime;
long minActiveTxBeginTime = timeHolder.minActiveTxBeginTime;
+
+ assert minRequiredTime != Long.MAX_VALUE : "Unexpected minRequiredTime is returned from determineGlobalMin call";
+
Catalog catalog = catalogManagerFacade.catalogByTsNullable(minRequiredTime);
CompletableFuture<Boolean> catalogCompactionFut;
if (catalog == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", minRequiredTime);
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Catalog compaction skipped, nothing to compact (timestamp={}). No catalog at minRequiredTime",
Review Comment:
Thanks. Fixed.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
+ }
+
+ return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinRequiredTime)
.thenComposeAsync(timeHolder -> {
long minRequiredTime = timeHolder.minRequiredTime;
long minActiveTxBeginTime = timeHolder.minActiveTxBeginTime;
+
+ assert minRequiredTime != Long.MAX_VALUE : "Unexpected minRequiredTime is returned from determineGlobalMin call";
+
Catalog catalog = catalogManagerFacade.catalogByTsNullable(minRequiredTime);
CompletableFuture<Boolean> catalogCompactionFut;
if (catalog == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", minRequiredTime);
+ if (LOG.isInfoEnabled()) {
Review Comment:
Thanks. Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "korlov42 (via GitHub)" <gi...@apache.org>.
korlov42 commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732263598
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/UpdateLogImpl.java:
##########
@@ -225,6 +225,8 @@ public CompletableFuture<Boolean> saveSnapshot(SnapshotEntry update) {
Iif iif = iif(versionIsRecent, saveSnapshotAndDropOutdatedUpdates, ops().yield(false));
+ LOG.info("Snapshot version is updated. [Old version: {}, new version: {}]", oldSnapshotVersion, snapshotVersion);
Review Comment:
```suggestion
LOG.info("Snapshot version is updated [oldVersion={}, newVersion={}].", oldSnapshotVersion, snapshotVersion);
```
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/UpdateLogImpl.java:
##########
@@ -225,6 +225,8 @@ public CompletableFuture<Boolean> saveSnapshot(SnapshotEntry update) {
Iif iif = iif(versionIsRecent, saveSnapshotAndDropOutdatedUpdates, ops().yield(false));
+ LOG.info("Snapshot version is updated. [Old version: {}, new version: {}]", oldSnapshotVersion, snapshotVersion);
Review Comment:
btw, this message is not quite correct. Next invoke is conditional, so there is chance that update won't be applied.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
+ }
+
+ return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinRequiredTime)
.thenComposeAsync(timeHolder -> {
long minRequiredTime = timeHolder.minRequiredTime;
long minActiveTxBeginTime = timeHolder.minActiveTxBeginTime;
+
+ assert minRequiredTime != Long.MAX_VALUE : "Unexpected minRequiredTime is returned from determineGlobalMin call";
+
Catalog catalog = catalogManagerFacade.catalogByTsNullable(minRequiredTime);
CompletableFuture<Boolean> catalogCompactionFut;
if (catalog == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", minRequiredTime);
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Catalog compaction skipped, nothing to compact (timestamp={}). No catalog at minRequiredTime",
Review Comment:
please format all messages according to [guideline](https://cwiki.apache.org/confluence/display/IGNITE/Java+Logging+Guidelines). Besides, every log message [must end with a period](https://cwiki.apache.org/confluence/display/IGNITE/Java+Code+Style+Guide#JavaCodeStyleGuide-1.4Punctuation).
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
+ }
+
+ return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinRequiredTime)
.thenComposeAsync(timeHolder -> {
long minRequiredTime = timeHolder.minRequiredTime;
long minActiveTxBeginTime = timeHolder.minActiveTxBeginTime;
+
+ assert minRequiredTime != Long.MAX_VALUE : "Unexpected minRequiredTime is returned from determineGlobalMin call";
+
Catalog catalog = catalogManagerFacade.catalogByTsNullable(minRequiredTime);
CompletableFuture<Boolean> catalogCompactionFut;
if (catalog == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", minRequiredTime);
+ if (LOG.isInfoEnabled()) {
Review Comment:
no need to make it conditional for already computed params
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732710895
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,62 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
Review Comment:
Also if this map is huge enough toString() may take some time even if debug is not enabled.
Also you missing period at the end of the sentence.
I suggest not to log ALL partitions and add period to the end of the message.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734730578
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
@lowka
The idea of the test `testRaftGroupsUpdate` is to check that 'minActiveTxBeginTime` on all partition listeners updated, not **snapshotted** minActiveTxTime.
And the original `minActiveTxBeginTime` (not `snapshottedMinActiveTxBeginTime`) was required ONLY for this test.
Since you removed this field (renamed `snapshottedMinActiveTxBeginTime` to `minActiveTxBeginTime` ) and since now we have full integration test for catalog compaction runner, I suggest to remove `testRaftGroupsUpdate` because it will not work properly until fully reworked, and, again seems we already have test for this .
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734745239
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
@lowka
The idea of the test testRaftGroupsUpdate is to check that 'minActiveTxBeginTime` on all partition listeners updated, not **snapshotted** minActiveTxTime.
And the original minActiveTxBeginTime (not snapshottedMinActiveTxBeginTime) was required ONLY for this test.
Since now we have full integration test for catalog compaction runner, I suggest to remove testRaftGroupsUpdate because it won't work properly and doesn't make much sense.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732475312
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
This is a mere placeholder that is going to be replaced in determineGlobalMinimumRequiredTime, this value is not sent anywhere.
There is check below that deals with that
```
assert minRequiredTime != Long.MAX_VALUE : "Unexpected minRequiredTime is returned from determineGlobalMin call";
```
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734638061
##########
modules/catalog-compaction/src/integrationTest/java/org/apache/ignite/internal/catalog/compaction/ItCatalogCompactionTest.java:
##########
@@ -67,11 +77,71 @@
class ItCatalogCompactionTest extends ClusterPerClassIntegrationTest {
private static final int CLUSTER_SIZE = 3;
+ /** How often we update the low water mark. */
+ private static final long LW_UPDATE_TIME_MS = TimeUnit.SECONDS.toMillis(10);
Review Comment:
Fixed. Disabled testRaftGroupsUpdate because it became more unstable under already created https://issues.apache.org/jira/browse/IGNITE-23052.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1735968931
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -545,16 +566,29 @@ public void onReceived(NetworkMessage message, ClusterNode sender, @Nullable Lon
private void handleMinimumTimesRequest(ClusterNode sender, Long correlationId) {
HybridTimestamp lwm = lowWatermark;
- Long minRequiredTime = lwm != null ? getMinLocalTime(lwm) : null;
+ Long minRequiredTime;
+ if (lwm != null) {
+ LocalMinTime minLocalTime = getMinLocalTime(lwm);
+ if (minLocalTime != null) {
+ minRequiredTime = minLocalTime.time;
+ } else {
+ minRequiredTime = null;
+ }
+ } else {
+ minRequiredTime = null;
+ }
// We do not have local min time yet. Reply with the absolute min time.
if (minRequiredTime == null) {
minRequiredTime = HybridTimestamp.MIN_VALUE.longValue();
}
+ Map<Integer, BitSet> availablePartitions = buildTablePartitions(localMinTimeProvider.minTimePerPartition());
Review Comment:
This call is unnecessary. I later use `LocalMinTime::availablePartitions` instead.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732508638
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,70 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition state is missing [partition={}, all={}]", e.getKey(), partitionStates);
+ }
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}]",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+ }
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}]", lwm);
+
+ Long localMinRequiredTime = getMinLocalTime(lwm);
+
+ if (localMinRequiredTime == null) {
+ // If do not have local time yet, Use a placeholder value that is going to be overwritten.
+ localMinRequiredTime = Long.MAX_VALUE;
Review Comment:
Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1734587193
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -538,9 +574,9 @@ private void handlePrepareToUpdateTimeOnReplicasMessage(NetworkMessage message)
/** Minimum required time provider. */
@FunctionalInterface
- interface MinimumRequiredTimeProvider {
- /** Returns minimum required timestamp. */
- long time();
+ public interface MinimumRequiredTimeProvider {
Review Comment:
Fixed.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -247,30 +238,57 @@ void triggerCompaction(@Nullable HybridTimestamp lwm) {
return;
}
- lastRunFuture = startCompaction(logicalTopologyService.localLogicalTopology());
+ lastRunFuture = startCompaction(lwm, logicalTopologyService.localLogicalTopology());
}
});
}
- private CompletableFuture<Void> startCompaction(LogicalTopologySnapshot topologySnapshot) {
- long localMinimum = localMinTimeProvider.time();
+ private @Nullable Long getMinLocalTime(HybridTimestamp lwm) {
+ Map<TablePartitionId, @Nullable Long> partitionStates = localMinTimeProvider.minTimePerPartition();
- if (catalogManagerFacade.catalogByTsNullable(localMinimum) == null) {
- LOG.info("Catalog compaction skipped, nothing to compact [timestamp={}].", localMinimum);
+ // Find the minimum time among all partitions.
+ long partitionMinTime = Long.MAX_VALUE;
- return CompletableFutures.nullCompletedFuture();
+ for (Map.Entry<TablePartitionId, Long> e : partitionStates.entrySet()) {
+ Long state = e.getValue();
+
+ if (state == null) {
+ LOG.debug("Partition state is missing [partition={}].", e.getKey());
+ return null;
+ }
+
+ partitionMinTime = Math.min(partitionMinTime, state);
}
- return determineGlobalMinimumRequiredTime(topologySnapshot.nodes(), localMinimum)
+ // Choose the minimum time between the low watermark and the minimum time among all partitions.
+ long chosenMinTime = Math.min(lwm.longValue(), partitionMinTime);
+
+ LOG.debug("Local minimum required time: [partitionMinTime={}, lowWatermark={}, chosen={}].",
+ partitionMinTime,
+ lwm,
+ chosenMinTime
+ );
+
+ return chosenMinTime;
+ }
+
+ private CompletableFuture<Void> startCompaction(HybridTimestamp lwm, LogicalTopologySnapshot topologySnapshot) {
+ LOG.info("Catalog compaction started at [lowWaterMark={}].", lwm);
Review Comment:
Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1732455568
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -601,6 +603,20 @@ public MvPartitionStorage getMvStorage() {
return minActiveTxBeginTime0;
}
+ /**
+ * Returns minimum starting time among all active RW transactions in the cluster,
+ * or {@code null} if the value has not yet been set.
+ */
+ public @Nullable Long snapshottedMinActiveTxBeginTime() {
Review Comment:
Thanks. Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1728565713
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -838,4 +842,54 @@ public void onLeaderStart(long term) {
listener.onLeaderStart();
}
}
+
+ /** Returns raft group state accessor. */
+ public <G extends ReplicationGroupId, L> RaftGroupStateProvider<G, L> getGroupStateProvider(
+ Class<G> groupType,
+ Class<L> groupListenerType
+ ) {
+ return new StateProviderImpl<>(groupType, groupListenerType, nodes);
+ }
+
+ private static class StateProviderImpl<G extends ReplicationGroupId, L> implements RaftGroupStateProvider<G, L> {
+
+ private final Class<G> groupType;
+
+ private final Class<L> listenerType;
+
+ private final Map<RaftNodeId, RaftGroupService> nodes;
+
+ StateProviderImpl(
+ Class<G> groupType,
+ Class<L> listenerType,
+ Map<RaftNodeId, RaftGroupService> nodes
+ ) {
+ this.groupType = groupType;
+ this.listenerType = listenerType;
+ this.nodes = nodes;
+ }
+
+ @Override
+ public <V> Map<G, V> getState(Function<L, V> stateFunc) {
+ Map<G, V> result = new HashMap<>();
+
+ for (Map.Entry<RaftNodeId, RaftGroupService> entry : nodes.entrySet()) {
+ RaftNodeId nodeId = entry.getKey();
+ ReplicationGroupId replicationGroupId = nodeId.groupId();
Review Comment:
All these `nodes` belong to table partitions, so I think have using a loop here is ok.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1777992800
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorServiceImpl.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public class MinimumRequiredTimeCollectorServiceImpl implements MinimumRequiredTimeCollectorService {
+
+ private final ConcurrentHashMap<TablePartitionId, Long> partitions = new ConcurrentHashMap<>();
+
+ /** {@inheritDoc} */
+ @Override
+ public synchronized void addPartition(TablePartitionId tablePartitionId) {
+ partitions.put(tablePartitionId, UNDEFINED_MIN_TIME);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public synchronized void recordMinActiveTxTimestamp(TablePartitionId tablePartitionId, long timestamp) {
+ Long time = partitions.get(tablePartitionId);
+ if (time == null) {
+ // Ignore removed partitions.
+ return;
+ }
+
+ if (time == UNDEFINED_MIN_TIME || timestamp > time) {
+ partitions.put(tablePartitionId, timestamp);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void removePartition(TablePartitionId tablePartitionId) {
Review Comment:
Added `synchronised`.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorServiceImpl.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public class MinimumRequiredTimeCollectorServiceImpl implements MinimumRequiredTimeCollectorService {
+
Review Comment:
Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "korlov42 (via GitHub)" <gi...@apache.org>.
korlov42 commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1778603431
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -211,12 +203,24 @@ public void updateCoordinator(ClusterNode newCoordinator) {
triggerCompaction(lowWatermark);
}
- /** Returns local view of the node on who is currently compaction coordinator. For test purposes only.*/
+ /** Returns local view of the node on who is currently compaction coordinator. For test purposes only. */
@TestOnly
public @Nullable String coordinator() {
return compactionCoordinatorNodeName;
}
+ /** Returns {@link MinimumRequiredTimeCollectorService}. For test purposes only. */
+ @TestOnly
+ public MinimumRequiredTimeCollectorService localMinTimeCollectorService() {
+ return localMinTimeCollectorService;
+ }
Review Comment:
so, you've introduced TestOnly method to expose object that was created outside and passed in constructor of CompactionRunner just one line before in the test... please remove this method
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorServiceImpl.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public class MinimumRequiredTimeCollectorServiceImpl implements MinimumRequiredTimeCollectorService {
+ private final ConcurrentHashMap<TablePartitionId, Long> partitions = new ConcurrentHashMap<>();
+
+ /** {@inheritDoc} */
+ @Override
+ public synchronized void addPartition(TablePartitionId tablePartitionId) {
+ partitions.put(tablePartitionId, UNDEFINED_MIN_TIME);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public synchronized void recordMinActiveTxTimestamp(TablePartitionId tablePartitionId, long timestamp) {
Review Comment:
why `synchronized`? I get that to make 2 call to ConcurrentHashMap atomic, but why don't just synchronize a single bucket rather that entire collection (e.g. use `computeIfPresent`)
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/MinimumRequiredTimeCollectorServiceImpl.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.ignite.internal.table.distributed.raft;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+
+/** Collects minimum required timestamp for each partition. */
+public class MinimumRequiredTimeCollectorServiceImpl implements MinimumRequiredTimeCollectorService {
+ private final ConcurrentHashMap<TablePartitionId, Long> partitions = new ConcurrentHashMap<>();
+
+ /** {@inheritDoc} */
+ @Override
+ public synchronized void addPartition(TablePartitionId tablePartitionId) {
+ partitions.put(tablePartitionId, UNDEFINED_MIN_TIME);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public synchronized void recordMinActiveTxTimestamp(TablePartitionId tablePartitionId, long timestamp) {
+ Long time = partitions.get(tablePartitionId);
+ if (time == null) {
+ // Ignore removed partitions.
+ return;
+ }
+
+ if (time == UNDEFINED_MIN_TIME || timestamp > time) {
+ partitions.put(tablePartitionId, timestamp);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public synchronized void removePartition(TablePartitionId tablePartitionId) {
+ partitions.remove(tablePartitionId);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public Map<TablePartitionId, Long> minTimestampPerPartition() {
+ return Map.copyOf(partitions);
+ }
+
+ @Override
+ public void close() {
Review Comment:
`close()` methods in its current form act more like `reset()`. After close, component must either throw an exception on every invocation or just ignore such invocation. But why do we have it in the first place? `MinimumRequiredTimeCollectorService` is created once per node and will garbage collected as soon as the last reference to the node instance will be lost. Is it a real use case when someone stops the node and keeps it somewhere (i.g. it's not collected by GC)?
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22637: Catalog compaction. Implement remote node part of compaction flow [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #4256:
URL: https://github.com/apache/ignite-3/pull/4256#discussion_r1773597316
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -387,24 +435,61 @@ private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTop
}
return catalogManagerFacade.compactCatalog(catalog.version());
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ LOG.warn("Catalog compaction has failed [timestamp={}].", ex, minRequiredTime);
+ } else {
+ if (res) {
+ LOG.info("Catalog compaction completed successfully [timestamp={}].", minRequiredTime);
+ } else {
+ LOG.info("Catalog compaction skipped [timestamp={}].", minRequiredTime);
+ }
+ }
});
}
- private CompletableFuture<Set<String>> requiredNodes(Catalog catalog) {
+ private CompletableFuture<Pair<Boolean, Set<String>>> validatePartitions(
+ Catalog catalog,
+ Map<Integer, BitSet> localPartitions,
+ Map<String, Map<Integer, BitSet>> remotePartitions
+ ) {
HybridTimestamp nowTs = clockService.now();
- Set<String> required = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ ConcurrentHashMap<String, CurrentPartitions> currentPartitionsPerNode = new ConcurrentHashMap<>();
return CompletableFutures.allOf(catalog.tables().stream()
- .map(table -> collectRequiredNodes(catalog, table, required, nowTs))
+ .map(table -> collectRequiredNodes(catalog, table, nowTs, currentPartitionsPerNode))
.collect(Collectors.toList())
- ).thenApply(ignore -> required);
+ ).thenApply(ignore -> {
+
+ Map<String, Map<Integer, BitSet>> all = new HashMap<>(remotePartitions);
+ all.put(localNodeName, localPartitions);
+
+ Set<String> required = currentPartitionsPerNode.keySet();
+
+ for (Map.Entry<String, Map<Integer, BitSet>> node : all.entrySet()) {
+ String nodeId = node.getKey();
+ Map<Integer, BitSet> tables = node.getValue();
+ CurrentPartitions partitionsPerNode = currentPartitionsPerNode.get(nodeId);
Review Comment:
Fixed.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -387,24 +435,61 @@ private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTop
}
return catalogManagerFacade.compactCatalog(catalog.version());
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ LOG.warn("Catalog compaction has failed [timestamp={}].", ex, minRequiredTime);
+ } else {
+ if (res) {
+ LOG.info("Catalog compaction completed successfully [timestamp={}].", minRequiredTime);
+ } else {
+ LOG.info("Catalog compaction skipped [timestamp={}].", minRequiredTime);
+ }
+ }
});
}
- private CompletableFuture<Set<String>> requiredNodes(Catalog catalog) {
+ private CompletableFuture<Pair<Boolean, Set<String>>> validatePartitions(
+ Catalog catalog,
+ Map<Integer, BitSet> localPartitions,
+ Map<String, Map<Integer, BitSet>> remotePartitions
+ ) {
HybridTimestamp nowTs = clockService.now();
- Set<String> required = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ ConcurrentHashMap<String, CurrentPartitions> currentPartitionsPerNode = new ConcurrentHashMap<>();
return CompletableFutures.allOf(catalog.tables().stream()
- .map(table -> collectRequiredNodes(catalog, table, required, nowTs))
+ .map(table -> collectRequiredNodes(catalog, table, nowTs, currentPartitionsPerNode))
.collect(Collectors.toList())
- ).thenApply(ignore -> required);
+ ).thenApply(ignore -> {
+
+ Map<String, Map<Integer, BitSet>> all = new HashMap<>(remotePartitions);
+ all.put(localNodeName, localPartitions);
+
+ Set<String> required = currentPartitionsPerNode.keySet();
+
+ for (Map.Entry<String, Map<Integer, BitSet>> node : all.entrySet()) {
+ String nodeId = node.getKey();
+ Map<Integer, BitSet> tables = node.getValue();
+ CurrentPartitions partitionsPerNode = currentPartitionsPerNode.get(nodeId);
+
+ if (partitionsPerNode == null) {
+ return new Pair<>(false, required);
+ } else {
+ Map<Integer, BitSet> data = partitionsPerNode.data();
+ if (!data.equals(tables)) {
Review Comment:
Done. It now checks for inclusion.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -387,24 +435,61 @@ private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTop
}
return catalogManagerFacade.compactCatalog(catalog.version());
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ LOG.warn("Catalog compaction has failed [timestamp={}].", ex, minRequiredTime);
+ } else {
+ if (res) {
+ LOG.info("Catalog compaction completed successfully [timestamp={}].", minRequiredTime);
+ } else {
+ LOG.info("Catalog compaction skipped [timestamp={}].", minRequiredTime);
+ }
+ }
});
}
- private CompletableFuture<Set<String>> requiredNodes(Catalog catalog) {
+ private CompletableFuture<Pair<Boolean, Set<String>>> validatePartitions(
+ Catalog catalog,
+ Map<Integer, BitSet> localPartitions,
+ Map<String, Map<Integer, BitSet>> remotePartitions
+ ) {
HybridTimestamp nowTs = clockService.now();
- Set<String> required = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ ConcurrentHashMap<String, CurrentPartitions> currentPartitionsPerNode = new ConcurrentHashMap<>();
return CompletableFutures.allOf(catalog.tables().stream()
- .map(table -> collectRequiredNodes(catalog, table, required, nowTs))
+ .map(table -> collectRequiredNodes(catalog, table, nowTs, currentPartitionsPerNode))
.collect(Collectors.toList())
- ).thenApply(ignore -> required);
+ ).thenApply(ignore -> {
+
+ Map<String, Map<Integer, BitSet>> all = new HashMap<>(remotePartitions);
+ all.put(localNodeName, localPartitions);
+
+ Set<String> required = currentPartitionsPerNode.keySet();
+
+ for (Map.Entry<String, Map<Integer, BitSet>> node : all.entrySet()) {
+ String nodeId = node.getKey();
+ Map<Integer, BitSet> tables = node.getValue();
+ CurrentPartitions partitionsPerNode = currentPartitionsPerNode.get(nodeId);
+
+ if (partitionsPerNode == null) {
Review Comment:
Fixed.
##########
modules/catalog-compaction/src/main/java/org/apache/ignite/internal/catalog/compaction/CatalogCompactionRunner.java:
##########
@@ -387,24 +435,61 @@ private CompletableFuture<Boolean> tryCompactCatalog(Catalog catalog, LogicalTop
}
return catalogManagerFacade.compactCatalog(catalog.version());
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ LOG.warn("Catalog compaction has failed [timestamp={}].", ex, minRequiredTime);
+ } else {
+ if (res) {
+ LOG.info("Catalog compaction completed successfully [timestamp={}].", minRequiredTime);
+ } else {
+ LOG.info("Catalog compaction skipped [timestamp={}].", minRequiredTime);
+ }
+ }
});
}
- private CompletableFuture<Set<String>> requiredNodes(Catalog catalog) {
+ private CompletableFuture<Pair<Boolean, Set<String>>> validatePartitions(
+ Catalog catalog,
+ Map<Integer, BitSet> localPartitions,
+ Map<String, Map<Integer, BitSet>> remotePartitions
+ ) {
HybridTimestamp nowTs = clockService.now();
- Set<String> required = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ ConcurrentHashMap<String, CurrentPartitions> currentPartitionsPerNode = new ConcurrentHashMap<>();
return CompletableFutures.allOf(catalog.tables().stream()
- .map(table -> collectRequiredNodes(catalog, table, required, nowTs))
+ .map(table -> collectRequiredNodes(catalog, table, nowTs, currentPartitionsPerNode))
.collect(Collectors.toList())
- ).thenApply(ignore -> required);
+ ).thenApply(ignore -> {
+
+ Map<String, Map<Integer, BitSet>> all = new HashMap<>(remotePartitions);
+ all.put(localNodeName, localPartitions);
+
+ Set<String> required = currentPartitionsPerNode.keySet();
Review Comment:
Renamed for clarity.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org