dsmiley commented on code in PR #3903: URL: https://github.com/apache/solr/pull/3903#discussion_r2678936722
########## solr/core/src/java/org/apache/solr/handler/admin/api/UpgradeCoreIndex.java: ########## @@ -65,6 +65,34 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +/** + * Implements the UPGRADECOREINDEX CoreAdmin action, which upgrades an existing core's index Review Comment: I appreciate your time writing this but a user isn't going to read this, only someone maintaing this code will. ref-guide documentation is for our users. ########## solr/core/src/java/org/apache/solr/handler/admin/api/UpgradeCoreIndex.java: ########## @@ -0,0 +1,473 @@ +/* + * 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.solr.handler.admin.api; + +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.nio.file.Path; +import java.util.List; +import java.util.Set; +import org.apache.lucene.document.Document; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexableField; +import org.apache.lucene.index.LeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.MergePolicy; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.index.StoredFields; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.Version; +import org.apache.solr.client.api.model.UpgradeCoreIndexRequestBody; +import org.apache.solr.client.api.model.UpgradeCoreIndexResponse; +import org.apache.solr.common.SolrException; +import org.apache.solr.common.SolrInputDocument; +import org.apache.solr.common.params.ModifiableSolrParams; +import org.apache.solr.common.params.UpdateParams; +import org.apache.solr.common.util.NamedList; +import org.apache.solr.core.CoreContainer; +import org.apache.solr.core.SolrCore; +import org.apache.solr.handler.RequestHandlerBase; +import org.apache.solr.handler.admin.CoreAdminHandler; +import org.apache.solr.index.LatestVersionMergePolicy; +import org.apache.solr.request.LocalSolrQueryRequest; +import org.apache.solr.request.SolrQueryRequest; +import org.apache.solr.request.SolrRequestHandler; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.schema.IndexSchema; +import org.apache.solr.schema.SchemaField; +import org.apache.solr.search.DocValuesIteratorCache; +import org.apache.solr.search.SolrDocumentFetcher; +import org.apache.solr.search.SolrIndexSearcher; +import org.apache.solr.update.AddUpdateCommand; +import org.apache.solr.update.CommitUpdateCommand; +import org.apache.solr.update.processor.UpdateRequestProcessor; +import org.apache.solr.update.processor.UpdateRequestProcessorChain; +import org.apache.solr.util.RefCounted; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Implements the UPGRADECOREINDEX CoreAdmin action, which upgrades an existing core's index + * in-place by reindexing documents from segments created by older Lucene versions. + * + * <p>This action is intended for user-managed or standalone installations when upgrading Solr + * across major versions. + * + * <p>The upgrade process: + * + * <ol> + * <li>Temporarily installs {@link LatestVersionMergePolicy} to prevent older-version segments + * from participating in merges during reindexing. + * <li>Iterates each segment whose {@code minVersion} is older than the current Lucene major + * version. For each live document, rebuilds a {@link SolrInputDocument} from stored fields, + * decorates it with non-stored DocValues fields (excluding copyField targets), and re-adds it + * through Solr's update pipeline. + * <li>Commits the changes and validates that no older-format segments remain. + * <li>Restores the original merge policy. + * </ol> + * + * <p><strong>Important:</strong> Only fields that are stored or have DocValues enabled can be + * preserved during upgrade. Fields that are neither stored, nor docValues-enabled or copyField + * targets, will lose their data. + * + * @see LatestVersionMergePolicy + * @see UpgradeCoreIndexRequestBody + * @see UpgradeCoreIndexResponse + */ +public class UpgradeCoreIndex extends CoreAdminAPIBase { + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + public enum CoreIndexUpgradeStatus { + UPGRADE_SUCCESSFUL, + ERROR, + NO_UPGRADE_NEEDED; + } + + private static final int RETRY_COUNT_FOR_SEGMENT_DELETION = 5; + + public UpgradeCoreIndex( + CoreContainer coreContainer, + CoreAdminHandler.CoreAdminAsyncTracker coreAdminAsyncTracker, + SolrQueryRequest req, + SolrQueryResponse rsp) { + super(coreContainer, coreAdminAsyncTracker, req, rsp); + } + + @Override + public boolean isExpensive() { + return true; + } + + public UpgradeCoreIndexResponse upgradeCoreIndex( + String coreName, UpgradeCoreIndexRequestBody requestBody) throws Exception { + ensureRequiredParameterProvided("coreName", coreName); + + final UpgradeCoreIndexResponse response = + instantiateJerseyResponse(UpgradeCoreIndexResponse.class); + + return handlePotentiallyAsynchronousTask( + response, + coreName, + requestBody.async, + "upgrade-index", + () -> { + try (SolrCore core = coreContainer.getCore(coreName)) { + + // Set LatestVersionMergePolicy to prevent older segments from + // participating in merges while we reindex. This is to prevent any older version + // segments from + // merging with any newly formed segments created due to reindexing and undoing the work + // we are doing. + RefCounted<IndexWriter> iwRef = null; + MergePolicy originalMergePolicy = null; + int numSegmentsEligibleForUpgrade = 0, numSegmentsUpgraded = 0; + try { + iwRef = core.getSolrCoreState().getIndexWriter(core); + + IndexWriter iw = iwRef.get(); + + originalMergePolicy = iw.getConfig().getMergePolicy(); + iw.getConfig() + .setMergePolicy(new LatestVersionMergePolicy(iw.getConfig().getMergePolicy())); + + RefCounted<SolrIndexSearcher> ssearcherRef = core.getSearcher(); + try { + List<LeafReaderContext> leafContexts = + ssearcherRef.get().getTopReaderContext().leaves(); + DocValuesIteratorCache dvICache = new DocValuesIteratorCache(ssearcherRef.get()); + + UpdateRequestProcessorChain updateProcessorChain = + getUpdateProcessorChain(core, requestBody.updateChain); + + for (LeafReaderContext lrc : leafContexts) { + if (shouldUpgradeSegment(lrc)) { + numSegmentsEligibleForUpgrade++; + } + } + if (numSegmentsEligibleForUpgrade == 0) { + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.NO_UPGRADE_NEEDED.toString(); + response.numSegmentsEligibleForUpgrade = 0; + return response; + } + + for (LeafReaderContext lrc : leafContexts) { + if (!shouldUpgradeSegment(lrc)) { + continue; + } + processSegment(lrc, updateProcessorChain, core, ssearcherRef.get(), dvICache); + numSegmentsUpgraded++; + } + } catch (Exception e) { + log.error("Error while processing core: [{}}]", coreName, e); + throw new CoreAdminAPIBaseException(e); + } finally { + // important to decrement searcher ref count after use since we obtained it via the + // SolrCore.getSearcher() method + ssearcherRef.decref(); + } + + try { + doCommit(core); + } catch (IOException e) { + throw new CoreAdminAPIBaseException(e); + } + + boolean indexUpgraded = isIndexUpgraded(core); + + if (!indexUpgraded) { + log.error( + "Validation failed for core '{}'. Some data is still present in the older (<{}.x) Lucene index format.", + coreName, + Version.LATEST.major); + throw new CoreAdminAPIBaseException( + new SolrException( + SolrException.ErrorCode.SERVER_ERROR, + "Validation failed for core '" + + coreName + + "'. Some data is still present in the older (<" + + Version.LATEST.major + + ".x) Lucene index format.")); + } + + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.UPGRADE_SUCCESSFUL.toString(); + response.numSegmentsEligibleForUpgrade = numSegmentsEligibleForUpgrade; + response.numSegmentsUpgraded = numSegmentsUpgraded; + } catch (Exception ioEx) { + throw new CoreAdminAPIBaseException(ioEx); + + } finally { + // Restore original merge policy + if (iwRef != null) { + IndexWriter iw = iwRef.get(); + if (originalMergePolicy != null) { + iw.getConfig().setMergePolicy(originalMergePolicy); + } + iwRef.decref(); + } + } + } + + return response; + }); + } + + private boolean shouldUpgradeSegment(LeafReaderContext lrc) { + Version segmentMinVersion = null; + + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + + SegmentCommitInfo si = ((SegmentReader) leafReader).getSegmentInfo(); + segmentMinVersion = si.info.getMinVersion(); + + return (segmentMinVersion == null || segmentMinVersion.major < Version.LATEST.major); + } + + @SuppressWarnings({"rawtypes"}) + private UpdateRequestProcessorChain getUpdateProcessorChain( + SolrCore core, String requestedUpdateChain) { + + // Try explicitly requested chain first + if (requestedUpdateChain != null) { + UpdateRequestProcessorChain resolvedChain = + core.getUpdateProcessingChain(requestedUpdateChain); + if (resolvedChain != null) { + return resolvedChain; + } else { + log.error( + "UPGRADECOREINDEX:: Requested update chain {} not found for core {}", + requestedUpdateChain, + core.getName()); + } + } + + // Try to find chain configured in /update handler + String updateChainName = null; + SolrRequestHandler reqHandler = core.getRequestHandler("/update"); + + NamedList initArgs = ((RequestHandlerBase) reqHandler).getInitArgs(); + + if (initArgs != null) { + // Check invariants first + Object invariants = initArgs.get("invariants"); + if (invariants instanceof NamedList) { + updateChainName = (String) ((NamedList) invariants).get(UpdateParams.UPDATE_CHAIN); + } + + // Check defaults if not found in invariants + if (updateChainName == null) { + Object defaults = initArgs.get("defaults"); + if (defaults instanceof NamedList) { + updateChainName = (String) ((NamedList) defaults).get(UpdateParams.UPDATE_CHAIN); + } + } + } + + // default chain is returned if updateChainName is null + return core.getUpdateProcessingChain(updateChainName); + } + + private boolean isIndexUpgraded(SolrCore core) throws IOException { + + try (FSDirectory dir = FSDirectory.open(Path.of(core.getIndexDir())); + IndexReader reader = DirectoryReader.open(dir)) { + + List<LeafReaderContext> leaves = reader.leaves(); + if (leaves == null || leaves.isEmpty()) { + // no segments to process/validate + return true; + } + + for (LeafReaderContext lrc : leaves) { + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + if (leafReader instanceof SegmentReader) { + SegmentReader segmentReader = (SegmentReader) leafReader; + SegmentCommitInfo si = segmentReader.getSegmentInfo(); + Version segMinVersion = si.info.getMinVersion(); + if (segMinVersion == null || segMinVersion.major != Version.LATEST.major) { + log.warn( + "isIndexUpgraded(): Core: {}, Segment [{}] is still at minVersion [{}] and is not updated to the latest version [{}]; numLiveDocs: [{}]", + core.getName(), + si.info.name, + (segMinVersion == null ? 6 : segMinVersion.major), + Version.LATEST.major, + segmentReader.numDocs()); + return false; + } + } + } + return true; + } catch (Exception e) { + log.error("Error while opening segmentInfos for core [{}]", core.getName(), e); + throw e; + } + } + + private void doCommit(SolrCore core) throws IOException { + try (LocalSolrQueryRequest req = new LocalSolrQueryRequest(core, new ModifiableSolrParams())) { + CommitUpdateCommand cmd = new CommitUpdateCommand(req, false); + core.getUpdateHandler().commit(cmd); + } catch (IOException ioEx) { + log.warn("Error committing on core [{}] during index upgrade", core.getName(), ioEx); + throw ioEx; + } + } + + private void processSegment( + LeafReaderContext leafReaderContext, + UpdateRequestProcessorChain processorChain, + SolrCore core, + SolrIndexSearcher solrIndexSearcher, + DocValuesIteratorCache dvICache) + throws Exception { + + Exception exceptionToThrow = null; + int numDocsProcessed = 0; + + String coreName = core.getName(); + IndexSchema indexSchema = core.getLatestSchema(); + + LeafReader leafReader = FilterLeafReader.unwrap(leafReaderContext.reader()); + SegmentReader segmentReader = (SegmentReader) leafReader; + final String segmentName = segmentReader.getSegmentName(); + Bits bits = segmentReader.getLiveDocs(); Review Comment: the name of this variable is terrible; it's like calling a `String` as `string`. The only name for this is `liveDocs` ########## solr/core/src/test/org/apache/solr/handler/admin/UpgradeCoreIndexActionTest.java: ########## @@ -0,0 +1,333 @@ +/* + * 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.solr.handler.admin; + +import java.lang.invoke.MethodHandles; +import java.lang.invoke.VarHandle; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.SegmentInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.util.Version; +import org.apache.solr.SolrTestCaseJ4; +import org.apache.solr.common.params.CommonAdminParams; +import org.apache.solr.common.params.CoreAdminParams; +import org.apache.solr.core.SolrCore; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.util.RefCounted; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +public class UpgradeCoreIndexActionTest extends SolrTestCaseJ4 { + private static final int DOCS_PER_SEGMENT = 3; + private static final String DV_FIELD = "dvonly_i_dvo"; + + private static VarHandle segmentInfoMinVersionHandle; + private static String savedDirectoryFactory; + + @BeforeClass + public static void beforeClass() throws Exception { + savedDirectoryFactory = System.getProperty("solr.directoryFactory"); + // UpgradeCoreIndex currently validates via FSDirectory; use a filesystem-backed factory. + System.setProperty("solr.directoryFactory", "solr.StandardDirectoryFactory"); + initCore("solrconfig-nomergepolicyfactory.xml", "schema.xml"); + segmentInfoMinVersionHandle = + MethodHandles.privateLookupIn(SegmentInfo.class, MethodHandles.lookup()) + .findVarHandle(SegmentInfo.class, "minVersion", Version.class); + } + + @AfterClass + public static void afterClass() { + if (savedDirectoryFactory == null) { + System.clearProperty("solr.directoryFactory"); + } else { + System.setProperty("solr.directoryFactory", savedDirectoryFactory); + } + } + + @Before + public void resetIndex() { + assertU(delQ("*:*")); + assertU(commit("openSearcher", "true")); + } + + @Test + public void testUpgradeCoreIndexSelectiveReindexDeletesOldSegments() throws Exception { + final SolrCore core = h.getCore(); + final String coreName = core.getName(); + + final SegmentLayout layout = buildThreeSegments(coreName); + final Version simulatedOldMinVersion = Version.fromBits(Version.LATEST.major - 1, 0, 0); + + // Simulate: + // - seg1: "pure 9x" (minVersion=9) + // - seg2: "pure 10x" (minVersion=10) + // - seg3: "minVersion 9x, version 10x" (merged segment; minVersion=9) + setMinVersionForSegments(core, Set.of(layout.seg1, layout.seg3), simulatedOldMinVersion); + + final Set<String> segmentsBeforeUpgrade = listSegmentNames(core); + + CoreAdminHandler admin = new CoreAdminHandler(h.getCoreContainer()); + try { + final SolrQueryResponse resp = new SolrQueryResponse(); + admin.handleRequestBody( + req( + CoreAdminParams.ACTION, + CoreAdminParams.CoreAdminAction.UPGRADECOREINDEX.toString(), + CoreAdminParams.CORE, + coreName), + resp); + + assertNull("Unexpected exception: " + resp.getException(), resp.getException()); + assertEquals(coreName, resp.getValues().get("core")); + assertEquals(2, resp.getValues().get("numSegmentsEligibleForUpgrade")); + assertEquals(2, resp.getValues().get("numSegmentsUpgraded")); + assertEquals("UPGRADE_SUCCESSFUL", resp.getValues().get("upgradeStatus")); + } finally { + admin.shutdown(); + admin.close(); + } + + // The action commits internally and reopens the searcher; verify segments on disk. + final Set<String> segmentsAfter = listSegmentNames(core); + final Set<String> newSegments = new HashSet<>(segmentsAfter); + newSegments.removeAll(segmentsBeforeUpgrade); + assertFalse( + "Expected at least one new segment to be created by reindexing", newSegments.isEmpty()); + assertTrue("Expected seg2 to remain", segmentsAfter.contains(layout.seg2)); + assertFalse("Expected seg1 to be dropped", segmentsAfter.contains(layout.seg1)); + assertFalse("Expected seg3 to be dropped", segmentsAfter.contains(layout.seg3)); + + // Searcher was reopened by the action's commit; verify document count and field values. + assertQ(req("q", "*:*"), "//result[@numFound='" + (3 * DOCS_PER_SEGMENT) + "']"); + + // Validate docValues-only (non-stored) fields were preserved for reindexed documents. + // seg1 and seg3 were reindexed; seg2 was not. + assertDocValuesOnlyFieldPreserved(); + } + + @Test + @SuppressWarnings({"unchecked"}) + public void testUpgradeCoreIndexAsyncRequestStatusContainsOperationResponse() throws Exception { + final SolrCore core = h.getCore(); + final String coreName = core.getName(); + + final SegmentLayout layout = buildThreeSegments(coreName); + final Version simulatedOldMinVersion = Version.fromBits(Version.LATEST.major - 1, 0, 0); + setMinVersionForSegments(core, Set.of(layout.seg1, layout.seg3), simulatedOldMinVersion); + + final Set<String> segmentsBeforeUpgrade = listSegmentNames(core); + + final String requestId = "upgradecoreindex_async_1"; + CoreAdminHandler admin = new CoreAdminHandler(h.getCoreContainer()); + try { + SolrQueryResponse submitResp = new SolrQueryResponse(); + admin.handleRequestBody( + req( + CoreAdminParams.ACTION, + CoreAdminParams.CoreAdminAction.UPGRADECOREINDEX.toString(), + CoreAdminParams.CORE, + coreName, + CommonAdminParams.ASYNC, + requestId), + submitResp); + assertNull(submitResp.getException()); + + SolrQueryResponse statusResp = new SolrQueryResponse(); + int maxRetries = 60; + while (maxRetries-- > 0) { + statusResp = new SolrQueryResponse(); + admin.handleRequestBody( + req( + CoreAdminParams.ACTION, + CoreAdminParams.CoreAdminAction.REQUESTSTATUS.toString(), + CoreAdminParams.REQUESTID, + requestId), + statusResp); + + if ("completed".equals(statusResp.getValues().get("STATUS"))) { + break; + } + Thread.sleep(250); + } + + assertEquals("completed", statusResp.getValues().get("STATUS")); + Object opResponse = statusResp.getValues().get("response"); + assertNotNull(opResponse); + assertTrue("Expected map response, got: " + opResponse.getClass(), opResponse instanceof Map); + + Map<String, Object> opResponseMap = (Map<String, Object>) opResponse; + assertEquals(coreName, opResponseMap.get("core")); + assertEquals(2, ((Number) opResponseMap.get("numSegmentsEligibleForUpgrade")).intValue()); + assertEquals(2, ((Number) opResponseMap.get("numSegmentsUpgraded")).intValue()); + assertEquals("UPGRADE_SUCCESSFUL", opResponseMap.get("upgradeStatus")); + } finally { + admin.shutdown(); + admin.close(); + } + + final Set<String> segmentsAfter = listSegmentNames(core); + final Set<String> newSegments = new HashSet<>(segmentsAfter); + newSegments.removeAll(segmentsBeforeUpgrade); + assertFalse( + "Expected at least one new segment to be created by reindexing", newSegments.isEmpty()); + assertTrue("Expected seg2 to remain", segmentsAfter.contains(layout.seg2)); + assertFalse("Expected seg1 to be dropped", segmentsAfter.contains(layout.seg1)); + assertFalse("Expected seg3 to be dropped", segmentsAfter.contains(layout.seg3)); + + // Validate docValues-only (non-stored) fields were preserved for reindexed documents. + assertDocValuesOnlyFieldPreserved(); + } + + @Test + public void testNoUpgradeNeededWhenAllSegmentsCurrent() throws Exception { + final SolrCore core = h.getCore(); + final String coreName = core.getName(); + + // Index documents and commit - all segments will be at the current Lucene version + for (int i = 0; i < DOCS_PER_SEGMENT; i++) { + assertU(adoc("id", Integer.toString(i))); + } + assertU(commit("openSearcher", "true")); + + final Set<String> segmentsBefore = listSegmentNames(core); + assertFalse("Expected at least one segment", segmentsBefore.isEmpty()); + + CoreAdminHandler admin = new CoreAdminHandler(h.getCoreContainer()); + try { + final SolrQueryResponse resp = new SolrQueryResponse(); + admin.handleRequestBody( + req( + CoreAdminParams.ACTION, + CoreAdminParams.CoreAdminAction.UPGRADECOREINDEX.toString(), + CoreAdminParams.CORE, + coreName), + resp); + + assertNull("Unexpected exception: " + resp.getException(), resp.getException()); + assertEquals(coreName, resp.getValues().get("core")); + assertEquals(0, resp.getValues().get("numSegmentsEligibleForUpgrade")); + assertEquals("NO_UPGRADE_NEEDED", resp.getValues().get("upgradeStatus")); + } finally { + admin.shutdown(); + admin.close(); + } + + // Verify no segments were modified + final Set<String> segmentsAfter = listSegmentNames(core); + assertEquals("Segments should remain unchanged", segmentsBefore, segmentsAfter); + + // Verify documents are still queryable + assertQ(req("q", "*:*"), "//result[@numFound='" + DOCS_PER_SEGMENT + "']"); + } + + private SegmentLayout buildThreeSegments(String coreName) throws Exception { + final SolrCore core = h.getCore(); + + Set<String> segmentsBefore = listSegmentNames(core); + indexDocs(0); + final String seg1 = commitAndGetNewSegment(core, segmentsBefore); + segmentsBefore = listSegmentNames(core); + + indexDocs(1000); + final String seg2 = commitAndGetNewSegment(core, segmentsBefore); + segmentsBefore = listSegmentNames(core); + + indexDocs(2000); + final String seg3 = commitAndGetNewSegment(core, segmentsBefore); + + Set<String> allSegments = listSegmentNames(core); + assertTrue(allSegments.contains(seg1)); + assertTrue(allSegments.contains(seg2)); + assertTrue(allSegments.contains(seg3)); + + return new SegmentLayout(coreName, seg1, seg2, seg3); + } + + private void indexDocs(int baseId) { + for (int i = 0; i < DOCS_PER_SEGMENT; i++) { + // schema.xml copies id into numeric fields; use numeric IDs to avoid parsing errors + final String id = Integer.toString(baseId + i); + assertU(adoc("id", id, DV_FIELD, Integer.toString(baseId + i + 10_000), "title", "t" + id)); + } + } + + private void assertDocValuesOnlyFieldPreserved() { + // Assert one doc that must have been reindexed (seg1) and one from seg3. + assertDocHasDvFieldValue(0, 10_000); + assertDocHasDvFieldValue(2000, 12_000); + + // Also sanity-check a doc from the untouched segment (seg2) still has its value. + assertDocHasDvFieldValue(1000, 11_000); + } + + private void assertDocHasDvFieldValue(int id, int expected) { + assertQ( + req("q", "id:" + id, "fl", "id," + DV_FIELD), + "//result[@numFound='1']", + "//result/doc/int[@name='" + DV_FIELD + "'][.='" + expected + "']"); + } + + private String commitAndGetNewSegment(SolrCore core, Set<String> segmentsBefore) + throws Exception { + assertU(commit("openSearcher", "true")); + Set<String> segmentsAfter = new HashSet<>(listSegmentNames(core)); + segmentsAfter.removeAll(new HashSet<>(segmentsBefore)); + assertEquals("Expected exactly one new segment", 1, segmentsAfter.size()); + return segmentsAfter.iterator().next(); + } + + private Set<String> listSegmentNames(SolrCore core) { + RefCounted<org.apache.solr.search.SolrIndexSearcher> searcherRef = core.getSearcher(); Review Comment: have you seen `core.withSearcher`? More elegant. ########## solr/core/src/java/org/apache/solr/handler/admin/api/UpgradeCoreIndex.java: ########## @@ -0,0 +1,473 @@ +/* + * 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.solr.handler.admin.api; + +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.nio.file.Path; +import java.util.List; +import java.util.Set; +import org.apache.lucene.document.Document; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexableField; +import org.apache.lucene.index.LeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.MergePolicy; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.index.StoredFields; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.Version; +import org.apache.solr.client.api.model.UpgradeCoreIndexRequestBody; +import org.apache.solr.client.api.model.UpgradeCoreIndexResponse; +import org.apache.solr.common.SolrException; +import org.apache.solr.common.SolrInputDocument; +import org.apache.solr.common.params.ModifiableSolrParams; +import org.apache.solr.common.params.UpdateParams; +import org.apache.solr.common.util.NamedList; +import org.apache.solr.core.CoreContainer; +import org.apache.solr.core.SolrCore; +import org.apache.solr.handler.RequestHandlerBase; +import org.apache.solr.handler.admin.CoreAdminHandler; +import org.apache.solr.index.LatestVersionMergePolicy; +import org.apache.solr.request.LocalSolrQueryRequest; +import org.apache.solr.request.SolrQueryRequest; +import org.apache.solr.request.SolrRequestHandler; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.schema.IndexSchema; +import org.apache.solr.schema.SchemaField; +import org.apache.solr.search.DocValuesIteratorCache; +import org.apache.solr.search.SolrDocumentFetcher; +import org.apache.solr.search.SolrIndexSearcher; +import org.apache.solr.update.AddUpdateCommand; +import org.apache.solr.update.CommitUpdateCommand; +import org.apache.solr.update.processor.UpdateRequestProcessor; +import org.apache.solr.update.processor.UpdateRequestProcessorChain; +import org.apache.solr.util.RefCounted; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Implements the UPGRADECOREINDEX CoreAdmin action, which upgrades an existing core's index + * in-place by reindexing documents from segments created by older Lucene versions. + * + * <p>This action is intended for user-managed or standalone installations when upgrading Solr + * across major versions. + * + * <p>The upgrade process: + * + * <ol> + * <li>Temporarily installs {@link LatestVersionMergePolicy} to prevent older-version segments + * from participating in merges during reindexing. + * <li>Iterates each segment whose {@code minVersion} is older than the current Lucene major + * version. For each live document, rebuilds a {@link SolrInputDocument} from stored fields, + * decorates it with non-stored DocValues fields (excluding copyField targets), and re-adds it + * through Solr's update pipeline. + * <li>Commits the changes and validates that no older-format segments remain. + * <li>Restores the original merge policy. + * </ol> + * + * <p><strong>Important:</strong> Only fields that are stored or have DocValues enabled can be + * preserved during upgrade. Fields that are neither stored, nor docValues-enabled or copyField + * targets, will lose their data. + * + * @see LatestVersionMergePolicy + * @see UpgradeCoreIndexRequestBody + * @see UpgradeCoreIndexResponse + */ +public class UpgradeCoreIndex extends CoreAdminAPIBase { + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + public enum CoreIndexUpgradeStatus { + UPGRADE_SUCCESSFUL, + ERROR, + NO_UPGRADE_NEEDED; + } + + private static final int RETRY_COUNT_FOR_SEGMENT_DELETION = 5; + + public UpgradeCoreIndex( + CoreContainer coreContainer, + CoreAdminHandler.CoreAdminAsyncTracker coreAdminAsyncTracker, + SolrQueryRequest req, + SolrQueryResponse rsp) { + super(coreContainer, coreAdminAsyncTracker, req, rsp); + } + + @Override + public boolean isExpensive() { + return true; + } + + public UpgradeCoreIndexResponse upgradeCoreIndex( + String coreName, UpgradeCoreIndexRequestBody requestBody) throws Exception { + ensureRequiredParameterProvided("coreName", coreName); + + final UpgradeCoreIndexResponse response = + instantiateJerseyResponse(UpgradeCoreIndexResponse.class); + + return handlePotentiallyAsynchronousTask( + response, + coreName, + requestBody.async, + "upgrade-index", + () -> { + try (SolrCore core = coreContainer.getCore(coreName)) { + + // Set LatestVersionMergePolicy to prevent older segments from + // participating in merges while we reindex. This is to prevent any older version + // segments from + // merging with any newly formed segments created due to reindexing and undoing the work + // we are doing. + RefCounted<IndexWriter> iwRef = null; + MergePolicy originalMergePolicy = null; + int numSegmentsEligibleForUpgrade = 0, numSegmentsUpgraded = 0; + try { + iwRef = core.getSolrCoreState().getIndexWriter(core); + + IndexWriter iw = iwRef.get(); + + originalMergePolicy = iw.getConfig().getMergePolicy(); + iw.getConfig() + .setMergePolicy(new LatestVersionMergePolicy(iw.getConfig().getMergePolicy())); + + RefCounted<SolrIndexSearcher> ssearcherRef = core.getSearcher(); + try { + List<LeafReaderContext> leafContexts = + ssearcherRef.get().getTopReaderContext().leaves(); + DocValuesIteratorCache dvICache = new DocValuesIteratorCache(ssearcherRef.get()); + + UpdateRequestProcessorChain updateProcessorChain = + getUpdateProcessorChain(core, requestBody.updateChain); + + for (LeafReaderContext lrc : leafContexts) { + if (shouldUpgradeSegment(lrc)) { + numSegmentsEligibleForUpgrade++; + } + } + if (numSegmentsEligibleForUpgrade == 0) { + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.NO_UPGRADE_NEEDED.toString(); + response.numSegmentsEligibleForUpgrade = 0; + return response; + } + + for (LeafReaderContext lrc : leafContexts) { + if (!shouldUpgradeSegment(lrc)) { + continue; + } + processSegment(lrc, updateProcessorChain, core, ssearcherRef.get(), dvICache); + numSegmentsUpgraded++; + } + } catch (Exception e) { + log.error("Error while processing core: [{}}]", coreName, e); + throw new CoreAdminAPIBaseException(e); + } finally { + // important to decrement searcher ref count after use since we obtained it via the + // SolrCore.getSearcher() method + ssearcherRef.decref(); + } + + try { + doCommit(core); + } catch (IOException e) { + throw new CoreAdminAPIBaseException(e); + } + + boolean indexUpgraded = isIndexUpgraded(core); + + if (!indexUpgraded) { + log.error( + "Validation failed for core '{}'. Some data is still present in the older (<{}.x) Lucene index format.", + coreName, + Version.LATEST.major); + throw new CoreAdminAPIBaseException( + new SolrException( + SolrException.ErrorCode.SERVER_ERROR, + "Validation failed for core '" + + coreName + + "'. Some data is still present in the older (<" + + Version.LATEST.major + + ".x) Lucene index format.")); + } + + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.UPGRADE_SUCCESSFUL.toString(); + response.numSegmentsEligibleForUpgrade = numSegmentsEligibleForUpgrade; + response.numSegmentsUpgraded = numSegmentsUpgraded; + } catch (Exception ioEx) { + throw new CoreAdminAPIBaseException(ioEx); + + } finally { + // Restore original merge policy + if (iwRef != null) { + IndexWriter iw = iwRef.get(); + if (originalMergePolicy != null) { + iw.getConfig().setMergePolicy(originalMergePolicy); + } + iwRef.decref(); + } + } + } + + return response; + }); + } + + private boolean shouldUpgradeSegment(LeafReaderContext lrc) { + Version segmentMinVersion = null; + + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + + SegmentCommitInfo si = ((SegmentReader) leafReader).getSegmentInfo(); + segmentMinVersion = si.info.getMinVersion(); + + return (segmentMinVersion == null || segmentMinVersion.major < Version.LATEST.major); + } + + @SuppressWarnings({"rawtypes"}) + private UpdateRequestProcessorChain getUpdateProcessorChain( + SolrCore core, String requestedUpdateChain) { + + // Try explicitly requested chain first + if (requestedUpdateChain != null) { + UpdateRequestProcessorChain resolvedChain = + core.getUpdateProcessingChain(requestedUpdateChain); + if (resolvedChain != null) { + return resolvedChain; + } else { + log.error( + "UPGRADECOREINDEX:: Requested update chain {} not found for core {}", + requestedUpdateChain, + core.getName()); + } + } + + // Try to find chain configured in /update handler + String updateChainName = null; + SolrRequestHandler reqHandler = core.getRequestHandler("/update"); + + NamedList initArgs = ((RequestHandlerBase) reqHandler).getInitArgs(); + + if (initArgs != null) { + // Check invariants first + Object invariants = initArgs.get("invariants"); + if (invariants instanceof NamedList) { + updateChainName = (String) ((NamedList) invariants).get(UpdateParams.UPDATE_CHAIN); + } + + // Check defaults if not found in invariants + if (updateChainName == null) { + Object defaults = initArgs.get("defaults"); + if (defaults instanceof NamedList) { + updateChainName = (String) ((NamedList) defaults).get(UpdateParams.UPDATE_CHAIN); + } + } + } + + // default chain is returned if updateChainName is null + return core.getUpdateProcessingChain(updateChainName); + } + + private boolean isIndexUpgraded(SolrCore core) throws IOException { + + try (FSDirectory dir = FSDirectory.open(Path.of(core.getIndexDir())); + IndexReader reader = DirectoryReader.open(dir)) { + + List<LeafReaderContext> leaves = reader.leaves(); + if (leaves == null || leaves.isEmpty()) { + // no segments to process/validate + return true; + } + + for (LeafReaderContext lrc : leaves) { + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + if (leafReader instanceof SegmentReader) { + SegmentReader segmentReader = (SegmentReader) leafReader; + SegmentCommitInfo si = segmentReader.getSegmentInfo(); + Version segMinVersion = si.info.getMinVersion(); + if (segMinVersion == null || segMinVersion.major != Version.LATEST.major) { + log.warn( + "isIndexUpgraded(): Core: {}, Segment [{}] is still at minVersion [{}] and is not updated to the latest version [{}]; numLiveDocs: [{}]", + core.getName(), + si.info.name, + (segMinVersion == null ? 6 : segMinVersion.major), + Version.LATEST.major, + segmentReader.numDocs()); + return false; + } + } + } + return true; + } catch (Exception e) { + log.error("Error while opening segmentInfos for core [{}]", core.getName(), e); + throw e; + } + } + + private void doCommit(SolrCore core) throws IOException { + try (LocalSolrQueryRequest req = new LocalSolrQueryRequest(core, new ModifiableSolrParams())) { + CommitUpdateCommand cmd = new CommitUpdateCommand(req, false); + core.getUpdateHandler().commit(cmd); + } catch (IOException ioEx) { + log.warn("Error committing on core [{}] during index upgrade", core.getName(), ioEx); + throw ioEx; + } + } + + private void processSegment( + LeafReaderContext leafReaderContext, + UpdateRequestProcessorChain processorChain, + SolrCore core, + SolrIndexSearcher solrIndexSearcher, + DocValuesIteratorCache dvICache) + throws Exception { + + Exception exceptionToThrow = null; + int numDocsProcessed = 0; + + String coreName = core.getName(); + IndexSchema indexSchema = core.getLatestSchema(); + + LeafReader leafReader = FilterLeafReader.unwrap(leafReaderContext.reader()); Review Comment: Lets not unwrap; it limited the range of uses of this tool. For example, maybe we want UninvertingReader to synthesize docValues from the index! ########## solr/core/src/java/org/apache/solr/handler/admin/api/UpgradeCoreIndex.java: ########## @@ -0,0 +1,473 @@ +/* + * 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.solr.handler.admin.api; + +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.nio.file.Path; +import java.util.List; +import java.util.Set; +import org.apache.lucene.document.Document; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexableField; +import org.apache.lucene.index.LeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.MergePolicy; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.index.StoredFields; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.Version; +import org.apache.solr.client.api.model.UpgradeCoreIndexRequestBody; +import org.apache.solr.client.api.model.UpgradeCoreIndexResponse; +import org.apache.solr.common.SolrException; +import org.apache.solr.common.SolrInputDocument; +import org.apache.solr.common.params.ModifiableSolrParams; +import org.apache.solr.common.params.UpdateParams; +import org.apache.solr.common.util.NamedList; +import org.apache.solr.core.CoreContainer; +import org.apache.solr.core.SolrCore; +import org.apache.solr.handler.RequestHandlerBase; +import org.apache.solr.handler.admin.CoreAdminHandler; +import org.apache.solr.index.LatestVersionMergePolicy; +import org.apache.solr.request.LocalSolrQueryRequest; +import org.apache.solr.request.SolrQueryRequest; +import org.apache.solr.request.SolrRequestHandler; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.schema.IndexSchema; +import org.apache.solr.schema.SchemaField; +import org.apache.solr.search.DocValuesIteratorCache; +import org.apache.solr.search.SolrDocumentFetcher; +import org.apache.solr.search.SolrIndexSearcher; +import org.apache.solr.update.AddUpdateCommand; +import org.apache.solr.update.CommitUpdateCommand; +import org.apache.solr.update.processor.UpdateRequestProcessor; +import org.apache.solr.update.processor.UpdateRequestProcessorChain; +import org.apache.solr.util.RefCounted; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Implements the UPGRADECOREINDEX CoreAdmin action, which upgrades an existing core's index + * in-place by reindexing documents from segments created by older Lucene versions. + * + * <p>This action is intended for user-managed or standalone installations when upgrading Solr + * across major versions. + * + * <p>The upgrade process: + * + * <ol> + * <li>Temporarily installs {@link LatestVersionMergePolicy} to prevent older-version segments + * from participating in merges during reindexing. + * <li>Iterates each segment whose {@code minVersion} is older than the current Lucene major + * version. For each live document, rebuilds a {@link SolrInputDocument} from stored fields, + * decorates it with non-stored DocValues fields (excluding copyField targets), and re-adds it + * through Solr's update pipeline. + * <li>Commits the changes and validates that no older-format segments remain. + * <li>Restores the original merge policy. + * </ol> + * + * <p><strong>Important:</strong> Only fields that are stored or have DocValues enabled can be + * preserved during upgrade. Fields that are neither stored, nor docValues-enabled or copyField + * targets, will lose their data. + * + * @see LatestVersionMergePolicy + * @see UpgradeCoreIndexRequestBody + * @see UpgradeCoreIndexResponse + */ +public class UpgradeCoreIndex extends CoreAdminAPIBase { + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + public enum CoreIndexUpgradeStatus { + UPGRADE_SUCCESSFUL, + ERROR, + NO_UPGRADE_NEEDED; + } + + private static final int RETRY_COUNT_FOR_SEGMENT_DELETION = 5; + + public UpgradeCoreIndex( + CoreContainer coreContainer, + CoreAdminHandler.CoreAdminAsyncTracker coreAdminAsyncTracker, + SolrQueryRequest req, + SolrQueryResponse rsp) { + super(coreContainer, coreAdminAsyncTracker, req, rsp); + } + + @Override + public boolean isExpensive() { + return true; + } + + public UpgradeCoreIndexResponse upgradeCoreIndex( + String coreName, UpgradeCoreIndexRequestBody requestBody) throws Exception { + ensureRequiredParameterProvided("coreName", coreName); + + final UpgradeCoreIndexResponse response = + instantiateJerseyResponse(UpgradeCoreIndexResponse.class); + + return handlePotentiallyAsynchronousTask( + response, + coreName, + requestBody.async, + "upgrade-index", + () -> { + try (SolrCore core = coreContainer.getCore(coreName)) { + + // Set LatestVersionMergePolicy to prevent older segments from + // participating in merges while we reindex. This is to prevent any older version + // segments from + // merging with any newly formed segments created due to reindexing and undoing the work + // we are doing. + RefCounted<IndexWriter> iwRef = null; + MergePolicy originalMergePolicy = null; + int numSegmentsEligibleForUpgrade = 0, numSegmentsUpgraded = 0; + try { + iwRef = core.getSolrCoreState().getIndexWriter(core); + + IndexWriter iw = iwRef.get(); + + originalMergePolicy = iw.getConfig().getMergePolicy(); + iw.getConfig() + .setMergePolicy(new LatestVersionMergePolicy(iw.getConfig().getMergePolicy())); + + RefCounted<SolrIndexSearcher> ssearcherRef = core.getSearcher(); + try { + List<LeafReaderContext> leafContexts = + ssearcherRef.get().getTopReaderContext().leaves(); + DocValuesIteratorCache dvICache = new DocValuesIteratorCache(ssearcherRef.get()); + + UpdateRequestProcessorChain updateProcessorChain = + getUpdateProcessorChain(core, requestBody.updateChain); + + for (LeafReaderContext lrc : leafContexts) { + if (shouldUpgradeSegment(lrc)) { + numSegmentsEligibleForUpgrade++; + } + } + if (numSegmentsEligibleForUpgrade == 0) { + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.NO_UPGRADE_NEEDED.toString(); + response.numSegmentsEligibleForUpgrade = 0; + return response; + } + + for (LeafReaderContext lrc : leafContexts) { + if (!shouldUpgradeSegment(lrc)) { + continue; + } + processSegment(lrc, updateProcessorChain, core, ssearcherRef.get(), dvICache); + numSegmentsUpgraded++; + } + } catch (Exception e) { + log.error("Error while processing core: [{}}]", coreName, e); + throw new CoreAdminAPIBaseException(e); + } finally { + // important to decrement searcher ref count after use since we obtained it via the + // SolrCore.getSearcher() method + ssearcherRef.decref(); + } + + try { + doCommit(core); + } catch (IOException e) { + throw new CoreAdminAPIBaseException(e); + } + + boolean indexUpgraded = isIndexUpgraded(core); + + if (!indexUpgraded) { + log.error( + "Validation failed for core '{}'. Some data is still present in the older (<{}.x) Lucene index format.", + coreName, + Version.LATEST.major); + throw new CoreAdminAPIBaseException( + new SolrException( + SolrException.ErrorCode.SERVER_ERROR, + "Validation failed for core '" + + coreName + + "'. Some data is still present in the older (<" + + Version.LATEST.major + + ".x) Lucene index format.")); + } + + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.UPGRADE_SUCCESSFUL.toString(); + response.numSegmentsEligibleForUpgrade = numSegmentsEligibleForUpgrade; + response.numSegmentsUpgraded = numSegmentsUpgraded; + } catch (Exception ioEx) { + throw new CoreAdminAPIBaseException(ioEx); + + } finally { + // Restore original merge policy + if (iwRef != null) { + IndexWriter iw = iwRef.get(); + if (originalMergePolicy != null) { + iw.getConfig().setMergePolicy(originalMergePolicy); + } + iwRef.decref(); + } + } + } + + return response; + }); + } + + private boolean shouldUpgradeSegment(LeafReaderContext lrc) { + Version segmentMinVersion = null; + + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + + SegmentCommitInfo si = ((SegmentReader) leafReader).getSegmentInfo(); + segmentMinVersion = si.info.getMinVersion(); + + return (segmentMinVersion == null || segmentMinVersion.major < Version.LATEST.major); + } + + @SuppressWarnings({"rawtypes"}) + private UpdateRequestProcessorChain getUpdateProcessorChain( + SolrCore core, String requestedUpdateChain) { + + // Try explicitly requested chain first + if (requestedUpdateChain != null) { + UpdateRequestProcessorChain resolvedChain = + core.getUpdateProcessingChain(requestedUpdateChain); + if (resolvedChain != null) { + return resolvedChain; + } else { + log.error( + "UPGRADECOREINDEX:: Requested update chain {} not found for core {}", + requestedUpdateChain, + core.getName()); + } + } + + // Try to find chain configured in /update handler + String updateChainName = null; + SolrRequestHandler reqHandler = core.getRequestHandler("/update"); + + NamedList initArgs = ((RequestHandlerBase) reqHandler).getInitArgs(); + + if (initArgs != null) { + // Check invariants first + Object invariants = initArgs.get("invariants"); + if (invariants instanceof NamedList) { + updateChainName = (String) ((NamedList) invariants).get(UpdateParams.UPDATE_CHAIN); + } + + // Check defaults if not found in invariants + if (updateChainName == null) { + Object defaults = initArgs.get("defaults"); + if (defaults instanceof NamedList) { + updateChainName = (String) ((NamedList) defaults).get(UpdateParams.UPDATE_CHAIN); + } + } + } + + // default chain is returned if updateChainName is null + return core.getUpdateProcessingChain(updateChainName); + } + + private boolean isIndexUpgraded(SolrCore core) throws IOException { + + try (FSDirectory dir = FSDirectory.open(Path.of(core.getIndexDir())); + IndexReader reader = DirectoryReader.open(dir)) { + + List<LeafReaderContext> leaves = reader.leaves(); + if (leaves == null || leaves.isEmpty()) { + // no segments to process/validate + return true; + } + + for (LeafReaderContext lrc : leaves) { + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + if (leafReader instanceof SegmentReader) { + SegmentReader segmentReader = (SegmentReader) leafReader; + SegmentCommitInfo si = segmentReader.getSegmentInfo(); + Version segMinVersion = si.info.getMinVersion(); + if (segMinVersion == null || segMinVersion.major != Version.LATEST.major) { + log.warn( + "isIndexUpgraded(): Core: {}, Segment [{}] is still at minVersion [{}] and is not updated to the latest version [{}]; numLiveDocs: [{}]", + core.getName(), + si.info.name, + (segMinVersion == null ? 6 : segMinVersion.major), + Version.LATEST.major, + segmentReader.numDocs()); + return false; + } + } + } + return true; + } catch (Exception e) { + log.error("Error while opening segmentInfos for core [{}]", core.getName(), e); + throw e; + } + } + + private void doCommit(SolrCore core) throws IOException { + try (LocalSolrQueryRequest req = new LocalSolrQueryRequest(core, new ModifiableSolrParams())) { + CommitUpdateCommand cmd = new CommitUpdateCommand(req, false); + core.getUpdateHandler().commit(cmd); + } catch (IOException ioEx) { + log.warn("Error committing on core [{}] during index upgrade", core.getName(), ioEx); + throw ioEx; + } + } + + private void processSegment( + LeafReaderContext leafReaderContext, + UpdateRequestProcessorChain processorChain, + SolrCore core, + SolrIndexSearcher solrIndexSearcher, + DocValuesIteratorCache dvICache) + throws Exception { + + Exception exceptionToThrow = null; + int numDocsProcessed = 0; + + String coreName = core.getName(); + IndexSchema indexSchema = core.getLatestSchema(); + + LeafReader leafReader = FilterLeafReader.unwrap(leafReaderContext.reader()); + SegmentReader segmentReader = (SegmentReader) leafReader; + final String segmentName = segmentReader.getSegmentName(); + Bits bits = segmentReader.getLiveDocs(); + SolrInputDocument solrDoc = null; + UpdateRequestProcessor processor = null; + LocalSolrQueryRequest solrRequest = null; Review Comment: Can you put this into try-with-resources to avoid the explicit close? ########## solr/core/src/java/org/apache/solr/handler/admin/api/UpgradeCoreIndex.java: ########## @@ -0,0 +1,473 @@ +/* + * 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.solr.handler.admin.api; + +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.nio.file.Path; +import java.util.List; +import java.util.Set; +import org.apache.lucene.document.Document; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexableField; +import org.apache.lucene.index.LeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.MergePolicy; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.index.StoredFields; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.Version; +import org.apache.solr.client.api.model.UpgradeCoreIndexRequestBody; +import org.apache.solr.client.api.model.UpgradeCoreIndexResponse; +import org.apache.solr.common.SolrException; +import org.apache.solr.common.SolrInputDocument; +import org.apache.solr.common.params.ModifiableSolrParams; +import org.apache.solr.common.params.UpdateParams; +import org.apache.solr.common.util.NamedList; +import org.apache.solr.core.CoreContainer; +import org.apache.solr.core.SolrCore; +import org.apache.solr.handler.RequestHandlerBase; +import org.apache.solr.handler.admin.CoreAdminHandler; +import org.apache.solr.index.LatestVersionMergePolicy; +import org.apache.solr.request.LocalSolrQueryRequest; +import org.apache.solr.request.SolrQueryRequest; +import org.apache.solr.request.SolrRequestHandler; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.schema.IndexSchema; +import org.apache.solr.schema.SchemaField; +import org.apache.solr.search.DocValuesIteratorCache; +import org.apache.solr.search.SolrDocumentFetcher; +import org.apache.solr.search.SolrIndexSearcher; +import org.apache.solr.update.AddUpdateCommand; +import org.apache.solr.update.CommitUpdateCommand; +import org.apache.solr.update.processor.UpdateRequestProcessor; +import org.apache.solr.update.processor.UpdateRequestProcessorChain; +import org.apache.solr.util.RefCounted; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Implements the UPGRADECOREINDEX CoreAdmin action, which upgrades an existing core's index + * in-place by reindexing documents from segments created by older Lucene versions. + * + * <p>This action is intended for user-managed or standalone installations when upgrading Solr + * across major versions. + * + * <p>The upgrade process: + * + * <ol> + * <li>Temporarily installs {@link LatestVersionMergePolicy} to prevent older-version segments + * from participating in merges during reindexing. + * <li>Iterates each segment whose {@code minVersion} is older than the current Lucene major + * version. For each live document, rebuilds a {@link SolrInputDocument} from stored fields, + * decorates it with non-stored DocValues fields (excluding copyField targets), and re-adds it + * through Solr's update pipeline. + * <li>Commits the changes and validates that no older-format segments remain. + * <li>Restores the original merge policy. + * </ol> + * + * <p><strong>Important:</strong> Only fields that are stored or have DocValues enabled can be + * preserved during upgrade. Fields that are neither stored, nor docValues-enabled or copyField + * targets, will lose their data. + * + * @see LatestVersionMergePolicy + * @see UpgradeCoreIndexRequestBody + * @see UpgradeCoreIndexResponse + */ +public class UpgradeCoreIndex extends CoreAdminAPIBase { + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + public enum CoreIndexUpgradeStatus { + UPGRADE_SUCCESSFUL, + ERROR, + NO_UPGRADE_NEEDED; + } + + private static final int RETRY_COUNT_FOR_SEGMENT_DELETION = 5; + + public UpgradeCoreIndex( + CoreContainer coreContainer, + CoreAdminHandler.CoreAdminAsyncTracker coreAdminAsyncTracker, + SolrQueryRequest req, + SolrQueryResponse rsp) { + super(coreContainer, coreAdminAsyncTracker, req, rsp); + } + + @Override + public boolean isExpensive() { + return true; + } + + public UpgradeCoreIndexResponse upgradeCoreIndex( + String coreName, UpgradeCoreIndexRequestBody requestBody) throws Exception { + ensureRequiredParameterProvided("coreName", coreName); + + final UpgradeCoreIndexResponse response = + instantiateJerseyResponse(UpgradeCoreIndexResponse.class); + + return handlePotentiallyAsynchronousTask( + response, + coreName, + requestBody.async, + "upgrade-index", + () -> { + try (SolrCore core = coreContainer.getCore(coreName)) { + + // Set LatestVersionMergePolicy to prevent older segments from + // participating in merges while we reindex. This is to prevent any older version + // segments from + // merging with any newly formed segments created due to reindexing and undoing the work + // we are doing. + RefCounted<IndexWriter> iwRef = null; + MergePolicy originalMergePolicy = null; + int numSegmentsEligibleForUpgrade = 0, numSegmentsUpgraded = 0; + try { + iwRef = core.getSolrCoreState().getIndexWriter(core); + + IndexWriter iw = iwRef.get(); + + originalMergePolicy = iw.getConfig().getMergePolicy(); + iw.getConfig() + .setMergePolicy(new LatestVersionMergePolicy(iw.getConfig().getMergePolicy())); + + RefCounted<SolrIndexSearcher> ssearcherRef = core.getSearcher(); + try { + List<LeafReaderContext> leafContexts = + ssearcherRef.get().getTopReaderContext().leaves(); + DocValuesIteratorCache dvICache = new DocValuesIteratorCache(ssearcherRef.get()); + + UpdateRequestProcessorChain updateProcessorChain = + getUpdateProcessorChain(core, requestBody.updateChain); + + for (LeafReaderContext lrc : leafContexts) { + if (shouldUpgradeSegment(lrc)) { + numSegmentsEligibleForUpgrade++; + } + } + if (numSegmentsEligibleForUpgrade == 0) { + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.NO_UPGRADE_NEEDED.toString(); + response.numSegmentsEligibleForUpgrade = 0; + return response; + } + + for (LeafReaderContext lrc : leafContexts) { + if (!shouldUpgradeSegment(lrc)) { + continue; + } + processSegment(lrc, updateProcessorChain, core, ssearcherRef.get(), dvICache); + numSegmentsUpgraded++; + } + } catch (Exception e) { + log.error("Error while processing core: [{}}]", coreName, e); + throw new CoreAdminAPIBaseException(e); + } finally { + // important to decrement searcher ref count after use since we obtained it via the + // SolrCore.getSearcher() method + ssearcherRef.decref(); + } + + try { + doCommit(core); + } catch (IOException e) { + throw new CoreAdminAPIBaseException(e); + } + + boolean indexUpgraded = isIndexUpgraded(core); + + if (!indexUpgraded) { + log.error( + "Validation failed for core '{}'. Some data is still present in the older (<{}.x) Lucene index format.", + coreName, + Version.LATEST.major); + throw new CoreAdminAPIBaseException( + new SolrException( + SolrException.ErrorCode.SERVER_ERROR, + "Validation failed for core '" + + coreName + + "'. Some data is still present in the older (<" + + Version.LATEST.major + + ".x) Lucene index format.")); + } + + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.UPGRADE_SUCCESSFUL.toString(); + response.numSegmentsEligibleForUpgrade = numSegmentsEligibleForUpgrade; + response.numSegmentsUpgraded = numSegmentsUpgraded; + } catch (Exception ioEx) { + throw new CoreAdminAPIBaseException(ioEx); + + } finally { + // Restore original merge policy + if (iwRef != null) { + IndexWriter iw = iwRef.get(); + if (originalMergePolicy != null) { + iw.getConfig().setMergePolicy(originalMergePolicy); + } + iwRef.decref(); + } + } + } + + return response; + }); + } + + private boolean shouldUpgradeSegment(LeafReaderContext lrc) { + Version segmentMinVersion = null; + + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + + SegmentCommitInfo si = ((SegmentReader) leafReader).getSegmentInfo(); + segmentMinVersion = si.info.getMinVersion(); + + return (segmentMinVersion == null || segmentMinVersion.major < Version.LATEST.major); + } + + @SuppressWarnings({"rawtypes"}) + private UpdateRequestProcessorChain getUpdateProcessorChain( + SolrCore core, String requestedUpdateChain) { + + // Try explicitly requested chain first + if (requestedUpdateChain != null) { + UpdateRequestProcessorChain resolvedChain = + core.getUpdateProcessingChain(requestedUpdateChain); + if (resolvedChain != null) { + return resolvedChain; + } else { + log.error( + "UPGRADECOREINDEX:: Requested update chain {} not found for core {}", + requestedUpdateChain, + core.getName()); + } + } + + // Try to find chain configured in /update handler + String updateChainName = null; + SolrRequestHandler reqHandler = core.getRequestHandler("/update"); + + NamedList initArgs = ((RequestHandlerBase) reqHandler).getInitArgs(); + + if (initArgs != null) { + // Check invariants first + Object invariants = initArgs.get("invariants"); + if (invariants instanceof NamedList) { + updateChainName = (String) ((NamedList) invariants).get(UpdateParams.UPDATE_CHAIN); + } + + // Check defaults if not found in invariants + if (updateChainName == null) { + Object defaults = initArgs.get("defaults"); + if (defaults instanceof NamedList) { + updateChainName = (String) ((NamedList) defaults).get(UpdateParams.UPDATE_CHAIN); + } + } + } + + // default chain is returned if updateChainName is null + return core.getUpdateProcessingChain(updateChainName); + } + + private boolean isIndexUpgraded(SolrCore core) throws IOException { + + try (FSDirectory dir = FSDirectory.open(Path.of(core.getIndexDir())); + IndexReader reader = DirectoryReader.open(dir)) { + + List<LeafReaderContext> leaves = reader.leaves(); + if (leaves == null || leaves.isEmpty()) { + // no segments to process/validate + return true; + } + + for (LeafReaderContext lrc : leaves) { + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + if (leafReader instanceof SegmentReader) { + SegmentReader segmentReader = (SegmentReader) leafReader; + SegmentCommitInfo si = segmentReader.getSegmentInfo(); + Version segMinVersion = si.info.getMinVersion(); + if (segMinVersion == null || segMinVersion.major != Version.LATEST.major) { + log.warn( + "isIndexUpgraded(): Core: {}, Segment [{}] is still at minVersion [{}] and is not updated to the latest version [{}]; numLiveDocs: [{}]", + core.getName(), + si.info.name, + (segMinVersion == null ? 6 : segMinVersion.major), + Version.LATEST.major, + segmentReader.numDocs()); + return false; + } + } + } + return true; + } catch (Exception e) { + log.error("Error while opening segmentInfos for core [{}]", core.getName(), e); + throw e; + } + } + + private void doCommit(SolrCore core) throws IOException { + try (LocalSolrQueryRequest req = new LocalSolrQueryRequest(core, new ModifiableSolrParams())) { + CommitUpdateCommand cmd = new CommitUpdateCommand(req, false); + core.getUpdateHandler().commit(cmd); + } catch (IOException ioEx) { + log.warn("Error committing on core [{}] during index upgrade", core.getName(), ioEx); + throw ioEx; + } + } + + private void processSegment( + LeafReaderContext leafReaderContext, + UpdateRequestProcessorChain processorChain, + SolrCore core, + SolrIndexSearcher solrIndexSearcher, + DocValuesIteratorCache dvICache) + throws Exception { + + Exception exceptionToThrow = null; + int numDocsProcessed = 0; + + String coreName = core.getName(); + IndexSchema indexSchema = core.getLatestSchema(); + + LeafReader leafReader = FilterLeafReader.unwrap(leafReaderContext.reader()); + SegmentReader segmentReader = (SegmentReader) leafReader; + final String segmentName = segmentReader.getSegmentName(); + Bits bits = segmentReader.getLiveDocs(); + SolrInputDocument solrDoc = null; + UpdateRequestProcessor processor = null; + LocalSolrQueryRequest solrRequest = null; + SolrDocumentFetcher docFetcher = solrIndexSearcher.getDocFetcher(); + try { + // Exclude copy field targets to avoid duplicating values on reindex + Set<String> nonStoredDVFields = docFetcher.getNonStoredDVsWithoutCopyTargets(); + solrRequest = new LocalSolrQueryRequest(core, new ModifiableSolrParams()); + + SolrQueryResponse rsp = new SolrQueryResponse(); + processor = processorChain.createProcessor(solrRequest, rsp); + StoredFields storedFields = segmentReader.storedFields(); + for (int luceneDocId = 0; luceneDocId < segmentReader.maxDoc(); luceneDocId++) { + if (bits != null && !bits.get(luceneDocId)) { + continue; + } + + Document doc = storedFields.document(luceneDocId); + solrDoc = toSolrInputDocument(doc, indexSchema); + + docFetcher.decorateDocValueFields( + solrDoc, leafReaderContext.docBase + luceneDocId, nonStoredDVFields, dvICache); + solrDoc.removeField("_version_"); Review Comment: hmm; are you sure about that? A quick look at `org.apache.solr.cloud.api.collections.ReindexCollectionCmd` doesn't find "version", thus I doubt even that one filters it out. ########## solr/core/src/java/org/apache/solr/handler/admin/api/UpgradeCoreIndex.java: ########## @@ -0,0 +1,473 @@ +/* + * 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.solr.handler.admin.api; + +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.nio.file.Path; +import java.util.List; +import java.util.Set; +import org.apache.lucene.document.Document; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexableField; +import org.apache.lucene.index.LeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.MergePolicy; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.index.StoredFields; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.Version; +import org.apache.solr.client.api.model.UpgradeCoreIndexRequestBody; +import org.apache.solr.client.api.model.UpgradeCoreIndexResponse; +import org.apache.solr.common.SolrException; +import org.apache.solr.common.SolrInputDocument; +import org.apache.solr.common.params.ModifiableSolrParams; +import org.apache.solr.common.params.UpdateParams; +import org.apache.solr.common.util.NamedList; +import org.apache.solr.core.CoreContainer; +import org.apache.solr.core.SolrCore; +import org.apache.solr.handler.RequestHandlerBase; +import org.apache.solr.handler.admin.CoreAdminHandler; +import org.apache.solr.index.LatestVersionMergePolicy; +import org.apache.solr.request.LocalSolrQueryRequest; +import org.apache.solr.request.SolrQueryRequest; +import org.apache.solr.request.SolrRequestHandler; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.schema.IndexSchema; +import org.apache.solr.schema.SchemaField; +import org.apache.solr.search.DocValuesIteratorCache; +import org.apache.solr.search.SolrDocumentFetcher; +import org.apache.solr.search.SolrIndexSearcher; +import org.apache.solr.update.AddUpdateCommand; +import org.apache.solr.update.CommitUpdateCommand; +import org.apache.solr.update.processor.UpdateRequestProcessor; +import org.apache.solr.update.processor.UpdateRequestProcessorChain; +import org.apache.solr.util.RefCounted; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Implements the UPGRADECOREINDEX CoreAdmin action, which upgrades an existing core's index + * in-place by reindexing documents from segments created by older Lucene versions. + * + * <p>This action is intended for user-managed or standalone installations when upgrading Solr + * across major versions. + * + * <p>The upgrade process: + * + * <ol> + * <li>Temporarily installs {@link LatestVersionMergePolicy} to prevent older-version segments + * from participating in merges during reindexing. + * <li>Iterates each segment whose {@code minVersion} is older than the current Lucene major + * version. For each live document, rebuilds a {@link SolrInputDocument} from stored fields, + * decorates it with non-stored DocValues fields (excluding copyField targets), and re-adds it + * through Solr's update pipeline. + * <li>Commits the changes and validates that no older-format segments remain. + * <li>Restores the original merge policy. + * </ol> + * + * <p><strong>Important:</strong> Only fields that are stored or have DocValues enabled can be + * preserved during upgrade. Fields that are neither stored, nor docValues-enabled or copyField + * targets, will lose their data. + * + * @see LatestVersionMergePolicy + * @see UpgradeCoreIndexRequestBody + * @see UpgradeCoreIndexResponse + */ +public class UpgradeCoreIndex extends CoreAdminAPIBase { + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + public enum CoreIndexUpgradeStatus { + UPGRADE_SUCCESSFUL, + ERROR, + NO_UPGRADE_NEEDED; + } + + private static final int RETRY_COUNT_FOR_SEGMENT_DELETION = 5; + + public UpgradeCoreIndex( + CoreContainer coreContainer, + CoreAdminHandler.CoreAdminAsyncTracker coreAdminAsyncTracker, + SolrQueryRequest req, + SolrQueryResponse rsp) { + super(coreContainer, coreAdminAsyncTracker, req, rsp); + } + + @Override + public boolean isExpensive() { + return true; + } + + public UpgradeCoreIndexResponse upgradeCoreIndex( + String coreName, UpgradeCoreIndexRequestBody requestBody) throws Exception { + ensureRequiredParameterProvided("coreName", coreName); + + final UpgradeCoreIndexResponse response = + instantiateJerseyResponse(UpgradeCoreIndexResponse.class); + + return handlePotentiallyAsynchronousTask( + response, + coreName, + requestBody.async, + "upgrade-index", + () -> { + try (SolrCore core = coreContainer.getCore(coreName)) { + + // Set LatestVersionMergePolicy to prevent older segments from + // participating in merges while we reindex. This is to prevent any older version + // segments from + // merging with any newly formed segments created due to reindexing and undoing the work + // we are doing. + RefCounted<IndexWriter> iwRef = null; + MergePolicy originalMergePolicy = null; + int numSegmentsEligibleForUpgrade = 0, numSegmentsUpgraded = 0; + try { + iwRef = core.getSolrCoreState().getIndexWriter(core); + + IndexWriter iw = iwRef.get(); + + originalMergePolicy = iw.getConfig().getMergePolicy(); + iw.getConfig() + .setMergePolicy(new LatestVersionMergePolicy(iw.getConfig().getMergePolicy())); + + RefCounted<SolrIndexSearcher> ssearcherRef = core.getSearcher(); + try { + List<LeafReaderContext> leafContexts = + ssearcherRef.get().getTopReaderContext().leaves(); + DocValuesIteratorCache dvICache = new DocValuesIteratorCache(ssearcherRef.get()); + + UpdateRequestProcessorChain updateProcessorChain = + getUpdateProcessorChain(core, requestBody.updateChain); + + for (LeafReaderContext lrc : leafContexts) { + if (shouldUpgradeSegment(lrc)) { + numSegmentsEligibleForUpgrade++; + } + } + if (numSegmentsEligibleForUpgrade == 0) { + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.NO_UPGRADE_NEEDED.toString(); + response.numSegmentsEligibleForUpgrade = 0; + return response; + } + + for (LeafReaderContext lrc : leafContexts) { + if (!shouldUpgradeSegment(lrc)) { + continue; + } + processSegment(lrc, updateProcessorChain, core, ssearcherRef.get(), dvICache); + numSegmentsUpgraded++; + } + } catch (Exception e) { + log.error("Error while processing core: [{}}]", coreName, e); + throw new CoreAdminAPIBaseException(e); + } finally { + // important to decrement searcher ref count after use since we obtained it via the + // SolrCore.getSearcher() method + ssearcherRef.decref(); + } + + try { + doCommit(core); + } catch (IOException e) { + throw new CoreAdminAPIBaseException(e); + } + + boolean indexUpgraded = isIndexUpgraded(core); + + if (!indexUpgraded) { + log.error( + "Validation failed for core '{}'. Some data is still present in the older (<{}.x) Lucene index format.", + coreName, + Version.LATEST.major); + throw new CoreAdminAPIBaseException( + new SolrException( + SolrException.ErrorCode.SERVER_ERROR, + "Validation failed for core '" + + coreName + + "'. Some data is still present in the older (<" + + Version.LATEST.major + + ".x) Lucene index format.")); + } + + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.UPGRADE_SUCCESSFUL.toString(); + response.numSegmentsEligibleForUpgrade = numSegmentsEligibleForUpgrade; + response.numSegmentsUpgraded = numSegmentsUpgraded; + } catch (Exception ioEx) { + throw new CoreAdminAPIBaseException(ioEx); + + } finally { + // Restore original merge policy + if (iwRef != null) { + IndexWriter iw = iwRef.get(); + if (originalMergePolicy != null) { + iw.getConfig().setMergePolicy(originalMergePolicy); + } + iwRef.decref(); + } + } + } + + return response; + }); + } + + private boolean shouldUpgradeSegment(LeafReaderContext lrc) { + Version segmentMinVersion = null; + + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + + SegmentCommitInfo si = ((SegmentReader) leafReader).getSegmentInfo(); + segmentMinVersion = si.info.getMinVersion(); + + return (segmentMinVersion == null || segmentMinVersion.major < Version.LATEST.major); + } + + @SuppressWarnings({"rawtypes"}) + private UpdateRequestProcessorChain getUpdateProcessorChain( + SolrCore core, String requestedUpdateChain) { + + // Try explicitly requested chain first + if (requestedUpdateChain != null) { + UpdateRequestProcessorChain resolvedChain = + core.getUpdateProcessingChain(requestedUpdateChain); + if (resolvedChain != null) { + return resolvedChain; + } else { + log.error( + "UPGRADECOREINDEX:: Requested update chain {} not found for core {}", + requestedUpdateChain, + core.getName()); + } + } + + // Try to find chain configured in /update handler + String updateChainName = null; + SolrRequestHandler reqHandler = core.getRequestHandler("/update"); + + NamedList initArgs = ((RequestHandlerBase) reqHandler).getInitArgs(); + + if (initArgs != null) { + // Check invariants first + Object invariants = initArgs.get("invariants"); + if (invariants instanceof NamedList) { + updateChainName = (String) ((NamedList) invariants).get(UpdateParams.UPDATE_CHAIN); + } + + // Check defaults if not found in invariants + if (updateChainName == null) { + Object defaults = initArgs.get("defaults"); + if (defaults instanceof NamedList) { + updateChainName = (String) ((NamedList) defaults).get(UpdateParams.UPDATE_CHAIN); + } + } + } + + // default chain is returned if updateChainName is null + return core.getUpdateProcessingChain(updateChainName); + } + + private boolean isIndexUpgraded(SolrCore core) throws IOException { + + try (FSDirectory dir = FSDirectory.open(Path.of(core.getIndexDir())); + IndexReader reader = DirectoryReader.open(dir)) { + + List<LeafReaderContext> leaves = reader.leaves(); + if (leaves == null || leaves.isEmpty()) { + // no segments to process/validate + return true; + } + + for (LeafReaderContext lrc : leaves) { + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + if (leafReader instanceof SegmentReader) { + SegmentReader segmentReader = (SegmentReader) leafReader; + SegmentCommitInfo si = segmentReader.getSegmentInfo(); + Version segMinVersion = si.info.getMinVersion(); + if (segMinVersion == null || segMinVersion.major != Version.LATEST.major) { + log.warn( + "isIndexUpgraded(): Core: {}, Segment [{}] is still at minVersion [{}] and is not updated to the latest version [{}]; numLiveDocs: [{}]", + core.getName(), + si.info.name, + (segMinVersion == null ? 6 : segMinVersion.major), + Version.LATEST.major, + segmentReader.numDocs()); + return false; + } + } + } + return true; + } catch (Exception e) { + log.error("Error while opening segmentInfos for core [{}]", core.getName(), e); + throw e; + } + } + + private void doCommit(SolrCore core) throws IOException { + try (LocalSolrQueryRequest req = new LocalSolrQueryRequest(core, new ModifiableSolrParams())) { + CommitUpdateCommand cmd = new CommitUpdateCommand(req, false); + core.getUpdateHandler().commit(cmd); + } catch (IOException ioEx) { + log.warn("Error committing on core [{}] during index upgrade", core.getName(), ioEx); + throw ioEx; + } + } + + private void processSegment( + LeafReaderContext leafReaderContext, + UpdateRequestProcessorChain processorChain, + SolrCore core, + SolrIndexSearcher solrIndexSearcher, + DocValuesIteratorCache dvICache) + throws Exception { + + Exception exceptionToThrow = null; + int numDocsProcessed = 0; + + String coreName = core.getName(); + IndexSchema indexSchema = core.getLatestSchema(); + + LeafReader leafReader = FilterLeafReader.unwrap(leafReaderContext.reader()); + SegmentReader segmentReader = (SegmentReader) leafReader; + final String segmentName = segmentReader.getSegmentName(); Review Comment: This line is the only reason why you casted leafReader to a segmentReader. And this variable is only used in logging. It'd be a shame to limit the functionality of this tool to require full unwrapping of the reader. Just log the leafReader instead? Or add a utility method used in logging to get a "pretty" name for the leaf reader that attempts to get the name by unwrapping. ########## solr/core/src/test/org/apache/solr/handler/admin/UpgradeCoreIndexActionTest.java: ########## @@ -0,0 +1,333 @@ +/* + * 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.solr.handler.admin; + +import java.lang.invoke.MethodHandles; +import java.lang.invoke.VarHandle; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.SegmentInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.util.Version; +import org.apache.solr.SolrTestCaseJ4; +import org.apache.solr.common.params.CommonAdminParams; +import org.apache.solr.common.params.CoreAdminParams; +import org.apache.solr.core.SolrCore; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.util.RefCounted; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +public class UpgradeCoreIndexActionTest extends SolrTestCaseJ4 { Review Comment: Could you please consider using `EmbeddedSolrServerTestRule` with `SolrTestCase` base class (not STCJ4) instead? I'm slowly trying to move our tests to be SolrClient based and away from STCJ4. Your test hear seems ripe for it. ########## solr/core/src/test/org/apache/solr/handler/admin/UpgradeCoreIndexActionTest.java: ########## @@ -0,0 +1,333 @@ +/* + * 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.solr.handler.admin; + +import java.lang.invoke.MethodHandles; +import java.lang.invoke.VarHandle; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.SegmentInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.util.Version; +import org.apache.solr.SolrTestCaseJ4; +import org.apache.solr.common.params.CommonAdminParams; +import org.apache.solr.common.params.CoreAdminParams; +import org.apache.solr.core.SolrCore; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.util.RefCounted; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +public class UpgradeCoreIndexActionTest extends SolrTestCaseJ4 { + private static final int DOCS_PER_SEGMENT = 3; + private static final String DV_FIELD = "dvonly_i_dvo"; + + private static VarHandle segmentInfoMinVersionHandle; + private static String savedDirectoryFactory; + + @BeforeClass + public static void beforeClass() throws Exception { + savedDirectoryFactory = System.getProperty("solr.directoryFactory"); + // UpgradeCoreIndex currently validates via FSDirectory; use a filesystem-backed factory. + System.setProperty("solr.directoryFactory", "solr.StandardDirectoryFactory"); + initCore("solrconfig-nomergepolicyfactory.xml", "schema.xml"); + segmentInfoMinVersionHandle = + MethodHandles.privateLookupIn(SegmentInfo.class, MethodHandles.lookup()) + .findVarHandle(SegmentInfo.class, "minVersion", Version.class); + } + + @AfterClass + public static void afterClass() { Review Comment: Solr's test infrastructure resets/clears all system properties at the end of each test. So you can remove this and stop bothering saving the directory. @epugh is at present removing all trace of such from tests you might have gotten inspiration from ########## solr/core/src/java/org/apache/solr/handler/admin/api/UpgradeCoreIndex.java: ########## @@ -0,0 +1,473 @@ +/* + * 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.solr.handler.admin.api; + +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.nio.file.Path; +import java.util.List; +import java.util.Set; +import org.apache.lucene.document.Document; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexableField; +import org.apache.lucene.index.LeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.MergePolicy; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.index.StoredFields; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.Version; +import org.apache.solr.client.api.model.UpgradeCoreIndexRequestBody; +import org.apache.solr.client.api.model.UpgradeCoreIndexResponse; +import org.apache.solr.common.SolrException; +import org.apache.solr.common.SolrInputDocument; +import org.apache.solr.common.params.ModifiableSolrParams; +import org.apache.solr.common.params.UpdateParams; +import org.apache.solr.common.util.NamedList; +import org.apache.solr.core.CoreContainer; +import org.apache.solr.core.SolrCore; +import org.apache.solr.handler.RequestHandlerBase; +import org.apache.solr.handler.admin.CoreAdminHandler; +import org.apache.solr.index.LatestVersionMergePolicy; +import org.apache.solr.request.LocalSolrQueryRequest; +import org.apache.solr.request.SolrQueryRequest; +import org.apache.solr.request.SolrRequestHandler; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.schema.IndexSchema; +import org.apache.solr.schema.SchemaField; +import org.apache.solr.search.DocValuesIteratorCache; +import org.apache.solr.search.SolrDocumentFetcher; +import org.apache.solr.search.SolrIndexSearcher; +import org.apache.solr.update.AddUpdateCommand; +import org.apache.solr.update.CommitUpdateCommand; +import org.apache.solr.update.processor.UpdateRequestProcessor; +import org.apache.solr.update.processor.UpdateRequestProcessorChain; +import org.apache.solr.util.RefCounted; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Implements the UPGRADECOREINDEX CoreAdmin action, which upgrades an existing core's index + * in-place by reindexing documents from segments created by older Lucene versions. + * + * <p>This action is intended for user-managed or standalone installations when upgrading Solr + * across major versions. + * + * <p>The upgrade process: + * + * <ol> + * <li>Temporarily installs {@link LatestVersionMergePolicy} to prevent older-version segments + * from participating in merges during reindexing. + * <li>Iterates each segment whose {@code minVersion} is older than the current Lucene major + * version. For each live document, rebuilds a {@link SolrInputDocument} from stored fields, + * decorates it with non-stored DocValues fields (excluding copyField targets), and re-adds it + * through Solr's update pipeline. + * <li>Commits the changes and validates that no older-format segments remain. + * <li>Restores the original merge policy. + * </ol> + * + * <p><strong>Important:</strong> Only fields that are stored or have DocValues enabled can be + * preserved during upgrade. Fields that are neither stored, nor docValues-enabled or copyField + * targets, will lose their data. + * + * @see LatestVersionMergePolicy + * @see UpgradeCoreIndexRequestBody + * @see UpgradeCoreIndexResponse + */ +public class UpgradeCoreIndex extends CoreAdminAPIBase { + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + public enum CoreIndexUpgradeStatus { + UPGRADE_SUCCESSFUL, + ERROR, + NO_UPGRADE_NEEDED; + } + + private static final int RETRY_COUNT_FOR_SEGMENT_DELETION = 5; + + public UpgradeCoreIndex( + CoreContainer coreContainer, + CoreAdminHandler.CoreAdminAsyncTracker coreAdminAsyncTracker, + SolrQueryRequest req, + SolrQueryResponse rsp) { + super(coreContainer, coreAdminAsyncTracker, req, rsp); + } + + @Override + public boolean isExpensive() { + return true; + } + + public UpgradeCoreIndexResponse upgradeCoreIndex( + String coreName, UpgradeCoreIndexRequestBody requestBody) throws Exception { + ensureRequiredParameterProvided("coreName", coreName); + + final UpgradeCoreIndexResponse response = + instantiateJerseyResponse(UpgradeCoreIndexResponse.class); + + return handlePotentiallyAsynchronousTask( + response, + coreName, + requestBody.async, + "upgrade-index", + () -> { + try (SolrCore core = coreContainer.getCore(coreName)) { + + // Set LatestVersionMergePolicy to prevent older segments from + // participating in merges while we reindex. This is to prevent any older version + // segments from + // merging with any newly formed segments created due to reindexing and undoing the work + // we are doing. + RefCounted<IndexWriter> iwRef = null; + MergePolicy originalMergePolicy = null; + int numSegmentsEligibleForUpgrade = 0, numSegmentsUpgraded = 0; + try { + iwRef = core.getSolrCoreState().getIndexWriter(core); + + IndexWriter iw = iwRef.get(); + + originalMergePolicy = iw.getConfig().getMergePolicy(); + iw.getConfig() + .setMergePolicy(new LatestVersionMergePolicy(iw.getConfig().getMergePolicy())); + + RefCounted<SolrIndexSearcher> ssearcherRef = core.getSearcher(); + try { + List<LeafReaderContext> leafContexts = + ssearcherRef.get().getTopReaderContext().leaves(); + DocValuesIteratorCache dvICache = new DocValuesIteratorCache(ssearcherRef.get()); + + UpdateRequestProcessorChain updateProcessorChain = + getUpdateProcessorChain(core, requestBody.updateChain); + + for (LeafReaderContext lrc : leafContexts) { + if (shouldUpgradeSegment(lrc)) { + numSegmentsEligibleForUpgrade++; + } + } + if (numSegmentsEligibleForUpgrade == 0) { + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.NO_UPGRADE_NEEDED.toString(); + response.numSegmentsEligibleForUpgrade = 0; + return response; + } + + for (LeafReaderContext lrc : leafContexts) { + if (!shouldUpgradeSegment(lrc)) { + continue; + } + processSegment(lrc, updateProcessorChain, core, ssearcherRef.get(), dvICache); + numSegmentsUpgraded++; + } + } catch (Exception e) { + log.error("Error while processing core: [{}}]", coreName, e); + throw new CoreAdminAPIBaseException(e); + } finally { + // important to decrement searcher ref count after use since we obtained it via the + // SolrCore.getSearcher() method + ssearcherRef.decref(); + } + + try { + doCommit(core); + } catch (IOException e) { + throw new CoreAdminAPIBaseException(e); + } + + boolean indexUpgraded = isIndexUpgraded(core); + + if (!indexUpgraded) { + log.error( + "Validation failed for core '{}'. Some data is still present in the older (<{}.x) Lucene index format.", + coreName, + Version.LATEST.major); + throw new CoreAdminAPIBaseException( + new SolrException( + SolrException.ErrorCode.SERVER_ERROR, + "Validation failed for core '" + + coreName + + "'. Some data is still present in the older (<" + + Version.LATEST.major + + ".x) Lucene index format.")); + } + + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.UPGRADE_SUCCESSFUL.toString(); + response.numSegmentsEligibleForUpgrade = numSegmentsEligibleForUpgrade; + response.numSegmentsUpgraded = numSegmentsUpgraded; + } catch (Exception ioEx) { + throw new CoreAdminAPIBaseException(ioEx); + + } finally { + // Restore original merge policy + if (iwRef != null) { + IndexWriter iw = iwRef.get(); + if (originalMergePolicy != null) { + iw.getConfig().setMergePolicy(originalMergePolicy); + } + iwRef.decref(); + } + } + } + + return response; + }); + } + + private boolean shouldUpgradeSegment(LeafReaderContext lrc) { + Version segmentMinVersion = null; + + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + + SegmentCommitInfo si = ((SegmentReader) leafReader).getSegmentInfo(); + segmentMinVersion = si.info.getMinVersion(); + + return (segmentMinVersion == null || segmentMinVersion.major < Version.LATEST.major); + } + + @SuppressWarnings({"rawtypes"}) + private UpdateRequestProcessorChain getUpdateProcessorChain( + SolrCore core, String requestedUpdateChain) { + + // Try explicitly requested chain first + if (requestedUpdateChain != null) { + UpdateRequestProcessorChain resolvedChain = + core.getUpdateProcessingChain(requestedUpdateChain); + if (resolvedChain != null) { + return resolvedChain; + } else { + log.error( + "UPGRADECOREINDEX:: Requested update chain {} not found for core {}", + requestedUpdateChain, + core.getName()); + } + } + + // Try to find chain configured in /update handler + String updateChainName = null; + SolrRequestHandler reqHandler = core.getRequestHandler("/update"); + + NamedList initArgs = ((RequestHandlerBase) reqHandler).getInitArgs(); + + if (initArgs != null) { + // Check invariants first + Object invariants = initArgs.get("invariants"); + if (invariants instanceof NamedList) { + updateChainName = (String) ((NamedList) invariants).get(UpdateParams.UPDATE_CHAIN); + } + + // Check defaults if not found in invariants + if (updateChainName == null) { + Object defaults = initArgs.get("defaults"); + if (defaults instanceof NamedList) { + updateChainName = (String) ((NamedList) defaults).get(UpdateParams.UPDATE_CHAIN); + } + } + } + + // default chain is returned if updateChainName is null + return core.getUpdateProcessingChain(updateChainName); + } + + private boolean isIndexUpgraded(SolrCore core) throws IOException { + + try (FSDirectory dir = FSDirectory.open(Path.of(core.getIndexDir())); + IndexReader reader = DirectoryReader.open(dir)) { + + List<LeafReaderContext> leaves = reader.leaves(); + if (leaves == null || leaves.isEmpty()) { + // no segments to process/validate + return true; + } + + for (LeafReaderContext lrc : leaves) { + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + if (leafReader instanceof SegmentReader) { + SegmentReader segmentReader = (SegmentReader) leafReader; + SegmentCommitInfo si = segmentReader.getSegmentInfo(); + Version segMinVersion = si.info.getMinVersion(); + if (segMinVersion == null || segMinVersion.major != Version.LATEST.major) { + log.warn( + "isIndexUpgraded(): Core: {}, Segment [{}] is still at minVersion [{}] and is not updated to the latest version [{}]; numLiveDocs: [{}]", + core.getName(), + si.info.name, + (segMinVersion == null ? 6 : segMinVersion.major), + Version.LATEST.major, + segmentReader.numDocs()); + return false; + } + } + } + return true; + } catch (Exception e) { + log.error("Error while opening segmentInfos for core [{}]", core.getName(), e); + throw e; + } + } + + private void doCommit(SolrCore core) throws IOException { + try (LocalSolrQueryRequest req = new LocalSolrQueryRequest(core, new ModifiableSolrParams())) { + CommitUpdateCommand cmd = new CommitUpdateCommand(req, false); + core.getUpdateHandler().commit(cmd); + } catch (IOException ioEx) { + log.warn("Error committing on core [{}] during index upgrade", core.getName(), ioEx); + throw ioEx; + } + } + + private void processSegment( + LeafReaderContext leafReaderContext, + UpdateRequestProcessorChain processorChain, + SolrCore core, + SolrIndexSearcher solrIndexSearcher, + DocValuesIteratorCache dvICache) + throws Exception { + + Exception exceptionToThrow = null; + int numDocsProcessed = 0; + + String coreName = core.getName(); + IndexSchema indexSchema = core.getLatestSchema(); + + LeafReader leafReader = FilterLeafReader.unwrap(leafReaderContext.reader()); + SegmentReader segmentReader = (SegmentReader) leafReader; + final String segmentName = segmentReader.getSegmentName(); + Bits bits = segmentReader.getLiveDocs(); + SolrInputDocument solrDoc = null; + UpdateRequestProcessor processor = null; + LocalSolrQueryRequest solrRequest = null; + SolrDocumentFetcher docFetcher = solrIndexSearcher.getDocFetcher(); + try { + // Exclude copy field targets to avoid duplicating values on reindex + Set<String> nonStoredDVFields = docFetcher.getNonStoredDVsWithoutCopyTargets(); + solrRequest = new LocalSolrQueryRequest(core, new ModifiableSolrParams()); + + SolrQueryResponse rsp = new SolrQueryResponse(); + processor = processorChain.createProcessor(solrRequest, rsp); + StoredFields storedFields = segmentReader.storedFields(); + for (int luceneDocId = 0; luceneDocId < segmentReader.maxDoc(); luceneDocId++) { + if (bits != null && !bits.get(luceneDocId)) { + continue; + } + + Document doc = storedFields.document(luceneDocId); + solrDoc = toSolrInputDocument(doc, indexSchema); + + docFetcher.decorateDocValueFields( + solrDoc, leafReaderContext.docBase + luceneDocId, nonStoredDVFields, dvICache); + solrDoc.removeField("_version_"); + AddUpdateCommand currDocCmd = new AddUpdateCommand(solrRequest); + currDocCmd.solrDoc = solrDoc; + processor.processAdd(currDocCmd); + numDocsProcessed++; + } + } catch (Exception e) { + log.error("Error while processing segment [{}] in core [{}]", segmentName, coreName, e); + exceptionToThrow = e; + } finally { + if (processor != null) { + try { + processor.finish(); + } catch (Exception e) { + log.error( + "Exception during processor.finish() for segment [{}] in core [{}]", + segmentName, + coreName, + e); + if (exceptionToThrow == null) { + exceptionToThrow = e; + } else { + exceptionToThrow.addSuppressed(e); + } + } + try { + processor.close(); + } catch (Exception e) { + log.error( + "Exception while closing update processor for segment [{}] in core [{}]", + segmentName, + coreName, + e); + if (exceptionToThrow == null) { + exceptionToThrow = e; + } else { + exceptionToThrow.addSuppressed(e); + } + } + } + if (solrRequest != null) { + try { + solrRequest.close(); + } catch (Exception e) { + if (exceptionToThrow == null) { + exceptionToThrow = e; + } else { + exceptionToThrow.addSuppressed(e); + } + } + } + } + + log.info( + "End processing segment : {}, core: {} docs processed: {}", + segmentName, + coreName, + numDocsProcessed); + + if (exceptionToThrow != null) { + throw exceptionToThrow; + } + } + + /* + * Convert a lucene Document to a SolrInputDocument + */ + protected SolrInputDocument toSolrInputDocument( Review Comment: This code looks duplicative of similar code in `RealtimeGetComponent`. Right? Can we dedupulicate? ########## solr/core/src/java/org/apache/solr/handler/admin/api/UpgradeCoreIndex.java: ########## @@ -0,0 +1,473 @@ +/* + * 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.solr.handler.admin.api; + +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.nio.file.Path; +import java.util.List; +import java.util.Set; +import org.apache.lucene.document.Document; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexableField; +import org.apache.lucene.index.LeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.MergePolicy; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.index.StoredFields; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.Version; +import org.apache.solr.client.api.model.UpgradeCoreIndexRequestBody; +import org.apache.solr.client.api.model.UpgradeCoreIndexResponse; +import org.apache.solr.common.SolrException; +import org.apache.solr.common.SolrInputDocument; +import org.apache.solr.common.params.ModifiableSolrParams; +import org.apache.solr.common.params.UpdateParams; +import org.apache.solr.common.util.NamedList; +import org.apache.solr.core.CoreContainer; +import org.apache.solr.core.SolrCore; +import org.apache.solr.handler.RequestHandlerBase; +import org.apache.solr.handler.admin.CoreAdminHandler; +import org.apache.solr.index.LatestVersionMergePolicy; +import org.apache.solr.request.LocalSolrQueryRequest; +import org.apache.solr.request.SolrQueryRequest; +import org.apache.solr.request.SolrRequestHandler; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.schema.IndexSchema; +import org.apache.solr.schema.SchemaField; +import org.apache.solr.search.DocValuesIteratorCache; +import org.apache.solr.search.SolrDocumentFetcher; +import org.apache.solr.search.SolrIndexSearcher; +import org.apache.solr.update.AddUpdateCommand; +import org.apache.solr.update.CommitUpdateCommand; +import org.apache.solr.update.processor.UpdateRequestProcessor; +import org.apache.solr.update.processor.UpdateRequestProcessorChain; +import org.apache.solr.util.RefCounted; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Implements the UPGRADECOREINDEX CoreAdmin action, which upgrades an existing core's index + * in-place by reindexing documents from segments created by older Lucene versions. + * + * <p>This action is intended for user-managed or standalone installations when upgrading Solr + * across major versions. + * + * <p>The upgrade process: + * + * <ol> + * <li>Temporarily installs {@link LatestVersionMergePolicy} to prevent older-version segments + * from participating in merges during reindexing. + * <li>Iterates each segment whose {@code minVersion} is older than the current Lucene major + * version. For each live document, rebuilds a {@link SolrInputDocument} from stored fields, + * decorates it with non-stored DocValues fields (excluding copyField targets), and re-adds it + * through Solr's update pipeline. + * <li>Commits the changes and validates that no older-format segments remain. + * <li>Restores the original merge policy. + * </ol> + * + * <p><strong>Important:</strong> Only fields that are stored or have DocValues enabled can be + * preserved during upgrade. Fields that are neither stored, nor docValues-enabled or copyField + * targets, will lose their data. + * + * @see LatestVersionMergePolicy + * @see UpgradeCoreIndexRequestBody + * @see UpgradeCoreIndexResponse + */ +public class UpgradeCoreIndex extends CoreAdminAPIBase { + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + public enum CoreIndexUpgradeStatus { + UPGRADE_SUCCESSFUL, + ERROR, + NO_UPGRADE_NEEDED; + } + + private static final int RETRY_COUNT_FOR_SEGMENT_DELETION = 5; + + public UpgradeCoreIndex( + CoreContainer coreContainer, + CoreAdminHandler.CoreAdminAsyncTracker coreAdminAsyncTracker, + SolrQueryRequest req, + SolrQueryResponse rsp) { + super(coreContainer, coreAdminAsyncTracker, req, rsp); + } + + @Override + public boolean isExpensive() { + return true; + } + + public UpgradeCoreIndexResponse upgradeCoreIndex( + String coreName, UpgradeCoreIndexRequestBody requestBody) throws Exception { + ensureRequiredParameterProvided("coreName", coreName); + + final UpgradeCoreIndexResponse response = + instantiateJerseyResponse(UpgradeCoreIndexResponse.class); + + return handlePotentiallyAsynchronousTask( + response, + coreName, + requestBody.async, + "upgrade-index", + () -> { + try (SolrCore core = coreContainer.getCore(coreName)) { + + // Set LatestVersionMergePolicy to prevent older segments from + // participating in merges while we reindex. This is to prevent any older version + // segments from + // merging with any newly formed segments created due to reindexing and undoing the work + // we are doing. + RefCounted<IndexWriter> iwRef = null; + MergePolicy originalMergePolicy = null; + int numSegmentsEligibleForUpgrade = 0, numSegmentsUpgraded = 0; + try { + iwRef = core.getSolrCoreState().getIndexWriter(core); + + IndexWriter iw = iwRef.get(); + + originalMergePolicy = iw.getConfig().getMergePolicy(); + iw.getConfig() + .setMergePolicy(new LatestVersionMergePolicy(iw.getConfig().getMergePolicy())); + + RefCounted<SolrIndexSearcher> ssearcherRef = core.getSearcher(); + try { + List<LeafReaderContext> leafContexts = + ssearcherRef.get().getTopReaderContext().leaves(); + DocValuesIteratorCache dvICache = new DocValuesIteratorCache(ssearcherRef.get()); + + UpdateRequestProcessorChain updateProcessorChain = + getUpdateProcessorChain(core, requestBody.updateChain); + + for (LeafReaderContext lrc : leafContexts) { + if (shouldUpgradeSegment(lrc)) { + numSegmentsEligibleForUpgrade++; + } + } + if (numSegmentsEligibleForUpgrade == 0) { + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.NO_UPGRADE_NEEDED.toString(); + response.numSegmentsEligibleForUpgrade = 0; + return response; + } + + for (LeafReaderContext lrc : leafContexts) { + if (!shouldUpgradeSegment(lrc)) { + continue; + } + processSegment(lrc, updateProcessorChain, core, ssearcherRef.get(), dvICache); + numSegmentsUpgraded++; + } + } catch (Exception e) { + log.error("Error while processing core: [{}}]", coreName, e); + throw new CoreAdminAPIBaseException(e); + } finally { + // important to decrement searcher ref count after use since we obtained it via the + // SolrCore.getSearcher() method + ssearcherRef.decref(); + } + + try { + doCommit(core); + } catch (IOException e) { + throw new CoreAdminAPIBaseException(e); + } + + boolean indexUpgraded = isIndexUpgraded(core); + + if (!indexUpgraded) { + log.error( + "Validation failed for core '{}'. Some data is still present in the older (<{}.x) Lucene index format.", + coreName, + Version.LATEST.major); + throw new CoreAdminAPIBaseException( + new SolrException( + SolrException.ErrorCode.SERVER_ERROR, + "Validation failed for core '" + + coreName + + "'. Some data is still present in the older (<" + + Version.LATEST.major + + ".x) Lucene index format.")); + } + + response.core = coreName; + response.upgradeStatus = CoreIndexUpgradeStatus.UPGRADE_SUCCESSFUL.toString(); + response.numSegmentsEligibleForUpgrade = numSegmentsEligibleForUpgrade; + response.numSegmentsUpgraded = numSegmentsUpgraded; + } catch (Exception ioEx) { + throw new CoreAdminAPIBaseException(ioEx); + + } finally { + // Restore original merge policy + if (iwRef != null) { + IndexWriter iw = iwRef.get(); + if (originalMergePolicy != null) { + iw.getConfig().setMergePolicy(originalMergePolicy); + } + iwRef.decref(); + } + } + } + + return response; + }); + } + + private boolean shouldUpgradeSegment(LeafReaderContext lrc) { + Version segmentMinVersion = null; + + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + + SegmentCommitInfo si = ((SegmentReader) leafReader).getSegmentInfo(); + segmentMinVersion = si.info.getMinVersion(); + + return (segmentMinVersion == null || segmentMinVersion.major < Version.LATEST.major); + } + + @SuppressWarnings({"rawtypes"}) + private UpdateRequestProcessorChain getUpdateProcessorChain( + SolrCore core, String requestedUpdateChain) { + + // Try explicitly requested chain first + if (requestedUpdateChain != null) { + UpdateRequestProcessorChain resolvedChain = + core.getUpdateProcessingChain(requestedUpdateChain); + if (resolvedChain != null) { + return resolvedChain; + } else { + log.error( + "UPGRADECOREINDEX:: Requested update chain {} not found for core {}", + requestedUpdateChain, + core.getName()); + } + } + + // Try to find chain configured in /update handler + String updateChainName = null; + SolrRequestHandler reqHandler = core.getRequestHandler("/update"); + + NamedList initArgs = ((RequestHandlerBase) reqHandler).getInitArgs(); + + if (initArgs != null) { + // Check invariants first + Object invariants = initArgs.get("invariants"); + if (invariants instanceof NamedList) { + updateChainName = (String) ((NamedList) invariants).get(UpdateParams.UPDATE_CHAIN); + } + + // Check defaults if not found in invariants + if (updateChainName == null) { + Object defaults = initArgs.get("defaults"); + if (defaults instanceof NamedList) { + updateChainName = (String) ((NamedList) defaults).get(UpdateParams.UPDATE_CHAIN); + } + } + } + + // default chain is returned if updateChainName is null + return core.getUpdateProcessingChain(updateChainName); + } + + private boolean isIndexUpgraded(SolrCore core) throws IOException { + + try (FSDirectory dir = FSDirectory.open(Path.of(core.getIndexDir())); + IndexReader reader = DirectoryReader.open(dir)) { + + List<LeafReaderContext> leaves = reader.leaves(); + if (leaves == null || leaves.isEmpty()) { + // no segments to process/validate + return true; + } + + for (LeafReaderContext lrc : leaves) { + LeafReader leafReader = lrc.reader(); + leafReader = FilterLeafReader.unwrap(leafReader); + if (leafReader instanceof SegmentReader) { + SegmentReader segmentReader = (SegmentReader) leafReader; + SegmentCommitInfo si = segmentReader.getSegmentInfo(); + Version segMinVersion = si.info.getMinVersion(); + if (segMinVersion == null || segMinVersion.major != Version.LATEST.major) { + log.warn( + "isIndexUpgraded(): Core: {}, Segment [{}] is still at minVersion [{}] and is not updated to the latest version [{}]; numLiveDocs: [{}]", + core.getName(), + si.info.name, + (segMinVersion == null ? 6 : segMinVersion.major), + Version.LATEST.major, + segmentReader.numDocs()); + return false; + } + } + } + return true; + } catch (Exception e) { + log.error("Error while opening segmentInfos for core [{}]", core.getName(), e); + throw e; + } + } + + private void doCommit(SolrCore core) throws IOException { + try (LocalSolrQueryRequest req = new LocalSolrQueryRequest(core, new ModifiableSolrParams())) { + CommitUpdateCommand cmd = new CommitUpdateCommand(req, false); + core.getUpdateHandler().commit(cmd); + } catch (IOException ioEx) { + log.warn("Error committing on core [{}] during index upgrade", core.getName(), ioEx); + throw ioEx; + } + } + + private void processSegment( + LeafReaderContext leafReaderContext, + UpdateRequestProcessorChain processorChain, + SolrCore core, + SolrIndexSearcher solrIndexSearcher, + DocValuesIteratorCache dvICache) + throws Exception { + + Exception exceptionToThrow = null; + int numDocsProcessed = 0; + + String coreName = core.getName(); + IndexSchema indexSchema = core.getLatestSchema(); + + LeafReader leafReader = FilterLeafReader.unwrap(leafReaderContext.reader()); + SegmentReader segmentReader = (SegmentReader) leafReader; + final String segmentName = segmentReader.getSegmentName(); + Bits bits = segmentReader.getLiveDocs(); + SolrInputDocument solrDoc = null; + UpdateRequestProcessor processor = null; + LocalSolrQueryRequest solrRequest = null; + SolrDocumentFetcher docFetcher = solrIndexSearcher.getDocFetcher(); + try { + // Exclude copy field targets to avoid duplicating values on reindex + Set<String> nonStoredDVFields = docFetcher.getNonStoredDVsWithoutCopyTargets(); + solrRequest = new LocalSolrQueryRequest(core, new ModifiableSolrParams()); + + SolrQueryResponse rsp = new SolrQueryResponse(); + processor = processorChain.createProcessor(solrRequest, rsp); + StoredFields storedFields = segmentReader.storedFields(); + for (int luceneDocId = 0; luceneDocId < segmentReader.maxDoc(); luceneDocId++) { + if (bits != null && !bits.get(luceneDocId)) { + continue; + } + + Document doc = storedFields.document(luceneDocId); + solrDoc = toSolrInputDocument(doc, indexSchema); + + docFetcher.decorateDocValueFields( + solrDoc, leafReaderContext.docBase + luceneDocId, nonStoredDVFields, dvICache); + solrDoc.removeField("_version_"); + AddUpdateCommand currDocCmd = new AddUpdateCommand(solrRequest); + currDocCmd.solrDoc = solrDoc; + processor.processAdd(currDocCmd); + numDocsProcessed++; + } + } catch (Exception e) { + log.error("Error while processing segment [{}] in core [{}]", segmentName, coreName, e); + exceptionToThrow = e; + } finally { + if (processor != null) { + try { + processor.finish(); + } catch (Exception e) { + log.error( + "Exception during processor.finish() for segment [{}] in core [{}]", + segmentName, + coreName, + e); + if (exceptionToThrow == null) { + exceptionToThrow = e; + } else { + exceptionToThrow.addSuppressed(e); + } + } + try { + processor.close(); + } catch (Exception e) { + log.error( + "Exception while closing update processor for segment [{}] in core [{}]", + segmentName, + coreName, + e); + if (exceptionToThrow == null) { + exceptionToThrow = e; + } else { + exceptionToThrow.addSuppressed(e); + } + } + } + if (solrRequest != null) { + try { + solrRequest.close(); + } catch (Exception e) { + if (exceptionToThrow == null) { + exceptionToThrow = e; + } else { + exceptionToThrow.addSuppressed(e); + } + } + } + } + + log.info( + "End processing segment : {}, core: {} docs processed: {}", + segmentName, + coreName, + numDocsProcessed); + + if (exceptionToThrow != null) { + throw exceptionToThrow; + } + } + + /* Review Comment: one more asterisk and then it's actually javadoc ########## solr/core/src/test/org/apache/solr/handler/admin/UpgradeCoreIndexActionTest.java: ########## @@ -0,0 +1,333 @@ +/* + * 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.solr.handler.admin; + +import java.lang.invoke.MethodHandles; +import java.lang.invoke.VarHandle; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.SegmentInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.util.Version; +import org.apache.solr.SolrTestCaseJ4; +import org.apache.solr.common.params.CommonAdminParams; +import org.apache.solr.common.params.CoreAdminParams; +import org.apache.solr.core.SolrCore; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.util.RefCounted; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +public class UpgradeCoreIndexActionTest extends SolrTestCaseJ4 { + private static final int DOCS_PER_SEGMENT = 3; + private static final String DV_FIELD = "dvonly_i_dvo"; + + private static VarHandle segmentInfoMinVersionHandle; + private static String savedDirectoryFactory; + + @BeforeClass + public static void beforeClass() throws Exception { + savedDirectoryFactory = System.getProperty("solr.directoryFactory"); + // UpgradeCoreIndex currently validates via FSDirectory; use a filesystem-backed factory. + System.setProperty("solr.directoryFactory", "solr.StandardDirectoryFactory"); + initCore("solrconfig-nomergepolicyfactory.xml", "schema.xml"); + segmentInfoMinVersionHandle = + MethodHandles.privateLookupIn(SegmentInfo.class, MethodHandles.lookup()) + .findVarHandle(SegmentInfo.class, "minVersion", Version.class); + } + + @AfterClass + public static void afterClass() { + if (savedDirectoryFactory == null) { + System.clearProperty("solr.directoryFactory"); + } else { + System.setProperty("solr.directoryFactory", savedDirectoryFactory); + } + } + + @Before + public void resetIndex() { + assertU(delQ("*:*")); + assertU(commit("openSearcher", "true")); + } + + @Test + public void testUpgradeCoreIndexSelectiveReindexDeletesOldSegments() throws Exception { + final SolrCore core = h.getCore(); + final String coreName = core.getName(); + + final SegmentLayout layout = buildThreeSegments(coreName); + final Version simulatedOldMinVersion = Version.fromBits(Version.LATEST.major - 1, 0, 0); + + // Simulate: + // - seg1: "pure 9x" (minVersion=9) + // - seg2: "pure 10x" (minVersion=10) + // - seg3: "minVersion 9x, version 10x" (merged segment; minVersion=9) + setMinVersionForSegments(core, Set.of(layout.seg1, layout.seg3), simulatedOldMinVersion); + + final Set<String> segmentsBeforeUpgrade = listSegmentNames(core); + + CoreAdminHandler admin = new CoreAdminHandler(h.getCoreContainer()); Review Comment: This is invasive... like why can't you send a normal looking request into Solr in a SolrJ/SolrClient way without needing to directly manipulate Solr server classes like this. This gets to my suggestion at the class declaration about using a SolrClient instead -- encourages client side approach. Only reach into Solr internals when actually needed to accomplish a goal. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
