This is an automated email from the ASF dual-hosted git repository. dsmiley pushed a commit to branch branch_9x in repository https://gitbox.apache.org/repos/asf/solr.git
commit b90d30581f25ce53ae439fdc44ac84d1b5813a0b Author: Nick <[email protected]> AuthorDate: Wed Oct 7 07:19:32 2026 -0600 SOLR-10198: EmbeddedSolrServer StreamingResponseCallback: to SolrJ-native types. (#4965) -- instead of Lucene types. Thus a user's callback implementation can be standard whether it's using EmbeddedSolrServer or a normal SolrClient. Adapted for branch_9x: pattern-matching instanceof replaced by casts (Java 11), test updated to 9.x SolrJ packages and test-rule APIs. (cherry picked from commit e97c2a0884cbd68790fd7e7036fbb7800db27856) (cherry picked from commit 9ed938a1f7a86fcd405408db2635e9d4ebbb1eb0) --- changelog/unreleased/SOLR-10198.yml | 7 + .../client/solrj/embedded/EmbeddedSolrServer.java | 13 +- .../org/apache/solr/response/DocsStreamer.java | 91 ++++++++++ .../TestEmbeddedSolrServerStreamingTypes.java | 190 +++++++++++++++++++++ .../tagger/EmbeddedSolrNoSerializeTest.java | 3 +- 5 files changed, 299 insertions(+), 5 deletions(-) diff --git a/changelog/unreleased/SOLR-10198.yml b/changelog/unreleased/SOLR-10198.yml new file mode 100644 index 00000000000..f10ba33fee6 --- /dev/null +++ b/changelog/unreleased/SOLR-10198.yml @@ -0,0 +1,7 @@ +title: "EmbeddedSolrServer StreamingResponseCallback: Normalize Lucene stored fields to SolrJ-native types." +type: fixed +authors: + - name: Nick Shanin +links: + - name: SOLR-10198 + url: https://issues.apache.org/jira/browse/SOLR-10198 diff --git a/solr/core/src/java/org/apache/solr/client/solrj/embedded/EmbeddedSolrServer.java b/solr/core/src/java/org/apache/solr/client/solrj/embedded/EmbeddedSolrServer.java index 82edafe2b7b..384f1de9dda 100644 --- a/solr/core/src/java/org/apache/solr/client/solrj/embedded/EmbeddedSolrServer.java +++ b/solr/core/src/java/org/apache/solr/client/solrj/embedded/EmbeddedSolrServer.java @@ -59,9 +59,11 @@ import org.apache.solr.request.SolrQueryRequest; import org.apache.solr.request.SolrRequestHandler; import org.apache.solr.request.SolrRequestInfo; import org.apache.solr.response.BinaryResponseWriter; +import org.apache.solr.response.DocsStreamer; import org.apache.solr.response.QueryResponseWriterUtil; import org.apache.solr.response.ResultContext; import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.schema.IndexSchema; import org.apache.solr.servlet.SolrRequestParsers; /** @@ -302,7 +304,7 @@ public class EmbeddedSolrServer extends SolrClient { }; // invoke callbacks, and writes the rest to byteBuffer - try (var javaBinCodec = createJavaBinCodec(callback, resolver)) { + try (var javaBinCodec = createJavaBinCodec(callback, resolver, req.getSchema())) { javaBinCodec.setWritableDocFields(resolver).marshal(rsp.getValues(), byteBuffer); } } @@ -359,12 +361,17 @@ public class EmbeddedSolrServer extends SolrClient { } private JavaBinCodec createJavaBinCodec( - final StreamingResponseCallback callback, final BinaryResponseWriter.Resolver resolver) { + final StreamingResponseCallback callback, + final BinaryResponseWriter.Resolver resolver, + final IndexSchema schema) { return new JavaBinCodec(resolver) { @Override public void writeSolrDocument(SolrDocument doc) { - callback.streamSolrDocument(doc); + // The callback receives the document itself, so it bypasses the conversion the + // Resolver applies when a document is serialized. Convert stored fields here so + // streamed documents carry the same native types as query() results. + callback.streamSolrDocument(DocsStreamer.externalizeStoredValues(doc, schema)); // super.writeSolrDocument( doc, fields ); } diff --git a/solr/core/src/java/org/apache/solr/response/DocsStreamer.java b/solr/core/src/java/org/apache/solr/response/DocsStreamer.java index 60a879c2ed7..26f1c02a1f5 100644 --- a/solr/core/src/java/org/apache/solr/response/DocsStreamer.java +++ b/solr/core/src/java/org/apache/solr/response/DocsStreamer.java @@ -17,10 +17,14 @@ package org.apache.solr.response; import java.io.IOException; +import java.lang.invoke.MethodHandles; import java.util.ArrayList; +import java.util.Collection; import java.util.HashSet; import java.util.Iterator; import java.util.List; +import java.util.Map; +import java.util.Objects; import java.util.Set; import org.apache.lucene.document.Document; import org.apache.lucene.index.IndexableField; @@ -52,9 +56,14 @@ import org.apache.solr.search.DocList; import org.apache.solr.search.ReturnFields; import org.apache.solr.search.SolrDocumentFetcher; import org.apache.solr.search.SolrReturnFields; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** This streams SolrDocuments from a DocList and applies transformer */ public class DocsStreamer implements Iterator<SolrDocument> { + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + private static final Object FAILED_STORED_VALUE = new Object(); + public static final Set<Class<? extends FieldType>> KNOWN_TYPES = new HashSet<>(); private final org.apache.solr.response.ResultContext rctx; @@ -184,6 +193,88 @@ public class DocsStreamer implements Iterator<SolrDocument> { public void remove() { // do nothing } + /** + * Replace Lucene {@link IndexableField} values on a {@link SolrDocument} (and nested / child + * documents) with the SolrJ-native objects that clients see after JavaBin deserialization. The + * serialized JavaBin path already converts stored values through JavaBinResponseWriter.Resolver; + * this exists for the EmbeddedSolrServer streaming path, whose codec hands documents straight to + * the callback, so {@code queryAndStreamResponse} matches {@code query} / {@code HttpSolrClient}. + * + * <p>A stored value that cannot be converted is logged and omitted; conversion continues for the + * remaining values and fields. + * + * <p>Do not call this from {@link #convertLuceneDocToSolrDoc}; JSON/XML writers and some + * transformers still expect stored fields as {@link IndexableField}. + * + * @see #getValue(SchemaField, IndexableField) + */ + public static SolrDocument externalizeStoredValues(SolrDocument doc, IndexSchema schema) { + if (doc == null || schema == null) { + return doc; + } + List<String> failedFields = null; + for (Iterator<Map.Entry<String, Object>> it = doc.iterator(); it.hasNext(); ) { + Map.Entry<String, Object> entry = it.next(); + Object val = entry.getValue(); + Object converted = externalizeValue(val, schema); + if (FAILED_STORED_VALUE.equals(converted)) { + // The document's entry iterator does not support remove(), so collect the + // failed fields and remove them from the document after the loop instead. + if (failedFields == null) { + failedFields = new ArrayList<>(); + } + failedFields.add(entry.getKey()); + } else if (!Objects.equals(converted, val)) { + entry.setValue(converted); + } + } + if (failedFields != null) { + for (String failedField : failedFields) { + doc.remove(failedField); + } + } + List<SolrDocument> children = doc.getChildDocuments(); + if (children != null) { + for (SolrDocument child : children) { + externalizeStoredValues(child, schema); + } + } + return doc; + } + + private static Object externalizeValue(Object val, IndexSchema schema) { + if (val instanceof IndexableField) { + IndexableField f = (IndexableField) val; + try { + return getValue(schema.getFieldOrNull(f.name()), f); + } catch (Exception | AssertionError e) { + // AssertionError: point field types (IntPointField, DatePointField, ...) throw it + // from toObject when a stored value predates a change to a point type. + log.warn("Error reading a field : {}", f, e); + return FAILED_STORED_VALUE; + } + } + if (val instanceof SolrDocument) { + return externalizeStoredValues((SolrDocument) val, schema); + } + if (val instanceof Collection) { + Collection<?> coll = (Collection<?>) val; + List<Object> out = new ArrayList<>(coll.size()); + boolean changed = false; + for (Object item : coll) { + Object converted = externalizeValue(item, schema); + if (FAILED_STORED_VALUE.equals(converted)) { + changed = true; + continue; + } + changed |= !Objects.equals(converted, item); + out.add(converted); + } + return changed ? (out.isEmpty() ? FAILED_STORED_VALUE : out) : val; + } + return val; + } + public static Object getValue(SchemaField sf, IndexableField f) { FieldType ft = null; if (sf != null) { diff --git a/solr/core/src/test/org/apache/solr/client/solrj/embedded/TestEmbeddedSolrServerStreamingTypes.java b/solr/core/src/test/org/apache/solr/client/solrj/embedded/TestEmbeddedSolrServerStreamingTypes.java new file mode 100644 index 00000000000..85a6d3c455c --- /dev/null +++ b/solr/core/src/test/org/apache/solr/client/solrj/embedded/TestEmbeddedSolrServerStreamingTypes.java @@ -0,0 +1,190 @@ +/* + * 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.client.solrj.embedded; + +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.lucene.document.Document; +import org.apache.lucene.document.StoredField; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexableField; +import org.apache.solr.SolrTestCase; +import org.apache.solr.SolrTestCaseJ4; +import org.apache.solr.client.solrj.SolrQuery; +import org.apache.solr.client.solrj.StreamingResponseCallback; +import org.apache.solr.client.solrj.response.QueryResponse; +import org.apache.solr.common.SolrDocument; +import org.apache.solr.common.SolrInputDocument; +import org.apache.solr.core.SolrCore; +import org.apache.solr.schema.IndexSchema; +import org.apache.solr.util.EmbeddedSolrServerTestRule; +import org.apache.solr.util.RefCounted; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +/** + * EmbeddedSolrServer {@code queryAndStreamResponse} must return the same native field types as + * {@code query} / HttpSolrClient, not Lucene {@link IndexableField} instances. + */ +public class TestEmbeddedSolrServerStreamingTypes extends SolrTestCase { + + @ClassRule + public static final EmbeddedSolrServerTestRule solrTestRule = new EmbeddedSolrServerTestRule(); + + @BeforeClass + public static void beforeClass() throws Exception { + System.setProperty("solr.test.sys.prop2", "proptwo"); // used by collection1's solrconfig.xml + solrTestRule.startSolr(SolrTestCaseJ4.TEST_PATH()); + SolrTestCaseJ4.newRandomConfig(); + solrTestRule + .newCollection() + .withConfigSet(SolrTestCaseJ4.TEST_COLL1_CONF().toString()) + .create(); + } + + @Test + public void testQueryAndStreamResponseReturnsNativeFieldTypes() throws Exception { + EmbeddedSolrServer server = solrTestRule.getSolrClient("collection1"); + server.deleteByQuery("*:*"); + // city_s1 is *_s1: stored string, multiValued=false. schema.xml is version 1.0, so + // fields without multiValued="false" (including name) default to multiValued and + // query() returns a one-element list. foo_i_p is pint so the native type is Integer. + // name is multiValued: streaming must convert each element (Collection branch). + SolrInputDocument doc = new SolrInputDocument(); + doc.addField("id", "1"); + doc.addField("foo_i_p", 42); + doc.addField("city_s1", "Boston"); + doc.addField("name", "Alice"); + doc.addField("name", "Bob"); + server.add(doc); + server.commit(); + + SolrQuery q = new SolrQuery("*:*"); + q.setFields("id", "foo_i_p", "city_s1", "name"); + + QueryResponse rsp = server.query(q); + SolrDocument queried = rsp.getResults().get(0); + Object queriedId = queried.getFieldValue("id"); + Object queriedInt = queried.getFieldValue("foo_i_p"); + Object queriedName = queried.getFieldValue("city_s1"); + assertEquals("1", queriedId); + assertEquals(Integer.valueOf(42), queriedInt); + assertEquals("Boston", queriedName); + assertFalse("query() must not leak Lucene stored fields", queriedInt instanceof IndexableField); + + AtomicReference<SolrDocument> streamed = new AtomicReference<>(); + server.queryAndStreamResponse( + q, + new StreamingResponseCallback() { + @Override + public void streamSolrDocument(SolrDocument doc) { + streamed.set(doc); + } + + @Override + public void streamDocListInfo(long numFound, long start, Float maxScore) { + assertEquals(1, numFound); + } + }); + + SolrDocument streamedDoc = streamed.get(); + assertNotNull(streamedDoc); + Object streamedId = streamedDoc.getFieldValue("id"); + Object streamedInt = streamedDoc.getFieldValue("foo_i_p"); + Object streamedName = streamedDoc.getFieldValue("city_s1"); + assertFalse( + "queryAndStreamResponse must not leak Lucene stored fields", + streamedInt instanceof IndexableField); + assertEquals(queriedId.getClass(), streamedId.getClass()); + assertEquals(queriedInt.getClass(), streamedInt.getClass()); + assertEquals(queriedName.getClass(), streamedName.getClass()); + assertEquals(queriedId, streamedId); + assertEquals(queriedInt, streamedInt); + assertEquals(queriedName, streamedName); + assertEquals(Integer.class, streamedInt.getClass()); + assertEquals(String.class, streamedName.getClass()); + // multi-valued field: both paths return a list of native Strings + Object queriedNames = queried.getFieldValue("name"); + Object streamedNames = streamedDoc.getFieldValue("name"); + assertTrue(queriedNames instanceof List); + assertTrue(streamedNames instanceof List); + assertEquals(queriedNames, streamedNames); + assertEquals(Arrays.asList("Alice", "Bob"), streamedNames); + } + + @Test + public void testStreamingOmitsValueWhoseConversionThrowsAssertionError() throws Exception { + EmbeddedSolrServer server = solrTestRule.getSolrClient("collection1"); + server.deleteByQuery("*:*"); + server.commit(); + + // Write a Lucene document directly, bypassing the update pipeline: count_i_p is a + // pint field now, but the stored value written here is a string, as if it had been + // indexed before the field's type changed to pint. (A pint field the other test in + // this class does not use, so the raw field cannot clash with its field metadata.) + // IntPointField.toObject throws AssertionError for such a value (it has no numeric + // value); that must not fail the whole streamed response. The value is logged and + // omitted instead. The other fields are created through the schema so their Lucene + // field shapes match documents the update pipeline writes. + Document luceneDoc = new Document(); + luceneDoc.add(new StoredField("count_i_p", "not-a-number")); + try (SolrCore core = solrTestRule.getCoreContainer().getCore("collection1")) { + IndexSchema schema = core.getLatestSchema(); + for (IndexableField f : schema.getField("id").createFields("bad1")) { + luceneDoc.add(f); + } + for (IndexableField f : schema.getField("city_s1").createFields("Boston")) { + luceneDoc.add(f); + } + RefCounted<IndexWriter> iwRef = core.getSolrCoreState().getIndexWriter(core); + try { + iwRef.get().addDocument(luceneDoc); + } finally { + iwRef.decref(); + } + } + server.commit(); + + SolrQuery q = new SolrQuery("*:*"); + q.setFields("id", "count_i_p", "city_s1"); + + AtomicReference<SolrDocument> streamed = new AtomicReference<>(); + server.queryAndStreamResponse( + q, + new StreamingResponseCallback() { + @Override + public void streamSolrDocument(SolrDocument doc) { + streamed.set(doc); + } + + @Override + public void streamDocListInfo(long numFound, long start, Float maxScore) { + assertEquals(1, numFound); + } + }); + + SolrDocument streamedDoc = streamed.get(); + assertNotNull("the document must still be streamed", streamedDoc); + assertEquals("bad1", streamedDoc.getFieldValue("id")); + assertEquals("Boston", streamedDoc.getFieldValue("city_s1")); + assertFalse( + "the unconvertible value must be omitted, not streamed", + streamedDoc.containsKey("count_i_p")); + } +} diff --git a/solr/core/src/test/org/apache/solr/handler/tagger/EmbeddedSolrNoSerializeTest.java b/solr/core/src/test/org/apache/solr/handler/tagger/EmbeddedSolrNoSerializeTest.java index 84b23dea383..c88312ccfeb 100644 --- a/solr/core/src/test/org/apache/solr/handler/tagger/EmbeddedSolrNoSerializeTest.java +++ b/solr/core/src/test/org/apache/solr/handler/tagger/EmbeddedSolrNoSerializeTest.java @@ -27,7 +27,6 @@ import java.util.Collection; import java.util.Collections; import java.util.concurrent.atomic.AtomicReference; import java.util.function.BiFunction; -import org.apache.lucene.document.Field; import org.apache.solr.SolrTestCaseJ4; import org.apache.solr.client.solrj.SolrServerException; import org.apache.solr.client.solrj.StreamingResponseCallback; @@ -154,6 +153,6 @@ public class EmbeddedSolrNoSerializeTest extends SolrTestCaseJ4 { QueryResponse rsp = req.process(solrServer); assertNotNull(rsp.getResponse().get("tags")); assertNotNull(refDoc.get()); - assertEquals("Boston", ((Field) refDoc.get().getFieldValue("name")).stringValue()); + assertEquals("Boston", refDoc.get().getFieldValue("name")); } }
