Skip to content

Commit 6b3dca3

Browse files
marko-bekhtamarko-bekhta
committed
HSEARCH-5076 WIP batch rewrite
1 parent 549b469 commit 6b3dca3

File tree

34 files changed

+1051
-279
lines changed

34 files changed

+1051
-279
lines changed

‎build/config/src/main/java/org/hibernate/search/build/report/loggers/LoggerCategoriesProcessor.java

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@
99
import java.io.Writer;
1010
import java.nio.charset.StandardCharsets;
1111
import java.util.ArrayList;
12-
import java.util.Collection;
1312
import java.util.Collections;
1413
import java.util.LinkedHashMap;
1514
import java.util.List;

‎build/enforcer/src/main/java/org/hibernate/search/build/enforcer/MavenProjectUtils.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,8 @@ public class MavenProjectUtils {
1111
public static final String HIBERNATE_SEARCH_PARENT_PUBLIC = "hibernate-search-parent-public";
1212
public static final String HIBERNATE_SEARCH_PARENT_PUBLIC_LUCENE_NEXT = "hibernate-search-parent-public-lucene-next";
1313
public static final String HIBERNATE_SEARCH_PARENT_INTEGRATION_TEST = "hibernate-search-parent-integrationtest";
14-
public static final String HIBERNATE_SEARCH_PARENT_INTEGRATION_TEST_LUCENE_NEXT = "hibernate-search-parent-integrationtest-lucene-next";
14+
public static final String HIBERNATE_SEARCH_PARENT_INTEGRATION_TEST_LUCENE_NEXT =
15+
"hibernate-search-parent-integrationtest-lucene-next";
1516
public static final String HIBERNATE_SEARCH_PARENT_RELOCATION = "hibernate-search-parent-relocation";
1617
public static final String DEPLOY_SKIP = "deploy.skip";
1718

@@ -35,7 +36,7 @@ public static boolean isAnyParentRelocationParent(MavenProject project) {
3536
public static boolean isAnyParentIntegrationTestParent(MavenProject project) {
3637
return project.hasParent()
3738
&& ( HIBERNATE_SEARCH_PARENT_INTEGRATION_TEST.equals( project.getParent().getArtifactId() )
38-
|| HIBERNATE_SEARCH_PARENT_INTEGRATION_TEST_LUCENE_NEXT.equals( project.getParent().getArtifactId() )
39+
|| HIBERNATE_SEARCH_PARENT_INTEGRATION_TEST_LUCENE_NEXT.equals( project.getParent().getArtifactId() )
3940
|| isAnyParentIntegrationTestParent( project.getParent() ) );
4041
}
4142

‎build/jqassistant/rules/rules.xml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -280,6 +280,7 @@
280280
WHEN 'hibernate-search-mapper-pojo-standalone' THEN 'StandalonePojo'
281281
WHEN 'hibernate-search-mapper-orm' THEN 'HibernateOrm'
282282
WHEN 'hibernate-search-mapper-orm-outbox-polling' THEN 'OutboxPolling'
283+
WHEN 'hibernate-search-mapper-orm-jakarta-batch-core' THEN 'BatchCore'
283284
WHEN 'hibernate-search-mapper-orm-jakarta-batch-jberet' THEN 'JBeret'
284285
ELSE 'UNKNOWN-MODULE-SPECIFIC-KEYWORD-PLEASE-UPDATE-JQASSISTANT-RULES'
285286
END

‎integrationtest/mapper/orm-jakarta-batch/src/test/java/org/hibernate/search/integrationtest/jakarta/batch/util/JobTestUtil.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
import org.hibernate.SessionFactory;
2323
import org.hibernate.engine.spi.SessionFactoryImplementor;
2424
import org.hibernate.search.jakarta.batch.core.massindexing.MassIndexingJob;
25-
import org.hibernate.search.jakarta.batch.core.massindexing.util.impl.EntityTypeDescriptor;
25+
import org.hibernate.search.jakarta.batch.core.massindexing.util.impl.BatchCoreEntityTypeDescriptor;
2626
import org.hibernate.search.mapper.orm.Search;
2727
import org.hibernate.search.mapper.orm.loading.spi.HibernateOrmLoadingTypeContext;
2828
import org.hibernate.search.mapper.orm.mapping.SearchMapping;
@@ -141,11 +141,11 @@ private static <T> List<T> find(Session session, Class<T> clazz, String key, Str
141141
.fetchHits( 1000 );
142142
}
143143

144-
public static EntityTypeDescriptor<?, ?> createEntityTypeDescriptor(EntityManagerFactory emf, Class<?> clazz) {
144+
public static BatchCoreEntityTypeDescriptor<?, ?> createEntityTypeDescriptor(EntityManagerFactory emf, Class<?> clazz) {
145145
SearchMapping mapping = Search.mapping( emf );
146146
BatchMappingContext mappingContext = (BatchMappingContext) mapping;
147147
HibernateOrmLoadingTypeContext<?> type = mappingContext.typeContextProvider()
148148
.byEntityName().getOrFail( mapping.indexedEntity( clazz ).jpaName() );
149-
return EntityTypeDescriptor.create( emf.unwrap( SessionFactoryImplementor.class ), type );
149+
return BatchCoreEntityTypeDescriptor.create( emf.unwrap( SessionFactoryImplementor.class ), type );
150150
}
151151
}

‎mapper/orm-jakarta-batch/core/src/main/java/org/hibernate/search/jakarta/batch/core/massindexing/impl/JobContextData.java

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515

1616
import jakarta.persistence.EntityManagerFactory;
1717

18-
import org.hibernate.search.jakarta.batch.core.massindexing.util.impl.EntityTypeDescriptor;
18+
import org.hibernate.search.jakarta.batch.core.massindexing.util.impl.BatchCoreEntityTypeDescriptor;
1919
import org.hibernate.search.mapper.orm.tenancy.spi.TenancyConfiguration;
2020
import org.hibernate.search.mapper.pojo.massindexing.MassIndexingDefaultCleanOperation;
2121

@@ -33,7 +33,7 @@ public class JobContextData {
3333
* In Jakarta Batch standard, only string values can be propagated using job properties, but class types are frequently
3434
* used too. So this map has string keys to facilitate lookup for values extracted from job properties.
3535
*/
36-
private Map<String, EntityTypeDescriptor<?, ?>> entityTypeDescriptorMap;
36+
private Map<String, BatchCoreEntityTypeDescriptor<?, ?>> entityTypeDescriptorMap;
3737

3838
private TenancyConfiguration tenancyConfiguration;
3939
private MassIndexingDefaultCleanOperation massIndexingDefaultCleanOperation;
@@ -50,8 +50,8 @@ public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) {
5050
this.entityManagerFactory = entityManagerFactory;
5151
}
5252

53-
public void setEntityTypeDescriptors(Collection<EntityTypeDescriptor<?, ?>> descriptors) {
54-
for ( EntityTypeDescriptor<?, ?> descriptor : descriptors ) {
53+
public void setEntityTypeDescriptors(Collection<BatchCoreEntityTypeDescriptor<?, ?>> descriptors) {
54+
for ( BatchCoreEntityTypeDescriptor<?, ?> descriptor : descriptors ) {
5555
entityTypeDescriptorMap.put( descriptor.jpaEntityName(), descriptor );
5656
}
5757
}
@@ -72,22 +72,22 @@ public void setMassIndexingDefaultCleanOperation(MassIndexingDefaultCleanOperati
7272
this.massIndexingDefaultCleanOperation = massIndexingDefaultCleanOperation;
7373
}
7474

75-
public EntityTypeDescriptor<?, ?> getEntityTypeDescriptor(String entityName) {
76-
EntityTypeDescriptor<?, ?> descriptor = entityTypeDescriptorMap.get( entityName );
75+
public BatchCoreEntityTypeDescriptor<?, ?> getEntityTypeDescriptor(String entityName) {
76+
BatchCoreEntityTypeDescriptor<?, ?> descriptor = entityTypeDescriptorMap.get( entityName );
7777
if ( descriptor == null ) {
7878
String msg = String.format( Locale.ROOT, "entity type %s not found.", entityName );
7979
throw new NoSuchElementException( msg );
8080
}
8181
return descriptor;
8282
}
8383

84-
public List<EntityTypeDescriptor<?, ?>> getEntityTypeDescriptors() {
84+
public List<BatchCoreEntityTypeDescriptor<?, ?>> getEntityTypeDescriptors() {
8585
return new ArrayList<>( entityTypeDescriptorMap.values() );
8686
}
8787

8888
public List<Class<?>> getEntityTypes() {
8989
return entityTypeDescriptorMap.values().stream()
90-
.map( EntityTypeDescriptor::javaClass )
90+
.map( BatchCoreEntityTypeDescriptor::javaClass )
9191
.collect( Collectors.toList() );
9292
}
9393

‎mapper/orm-jakarta-batch/core/src/main/java/org/hibernate/search/jakarta/batch/core/massindexing/loading/impl/BatchCoreDefaultHibernateOrmBatchEntityLoader.java

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
* Copyright Red Hat Inc. and Hibernate Authors
4+
*/
5+
package org.hibernate.search.jakarta.batch.core.massindexing.loading.impl;
6+
7+
import java.util.List;
8+
9+
import jakarta.persistence.LockModeType;
10+
11+
import org.hibernate.Session;
12+
import org.hibernate.query.Query;
13+
import org.hibernate.query.QueryFlushMode;
14+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchEntityLoader;
15+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchEntityLoadingOptions;
16+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchEntitySink;
17+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchLoadingTypeContext;
18+
19+
public class BatchCoreDefaultHibernateOrmBatchEntityLoader<E> implements HibernateOrmBatchEntityLoader {
20+
private static final String ID_PARAMETER_NAME = "ids";
21+
22+
private final HibernateOrmBatchEntitySink<E> sink;
23+
private final Query<E> query;
24+
25+
public BatchCoreDefaultHibernateOrmBatchEntityLoader(HibernateOrmBatchLoadingTypeContext<E> typeContext,
26+
HibernateOrmBatchEntitySink<E> sink, HibernateOrmBatchEntityLoadingOptions options) {
27+
this.sink = sink;
28+
29+
StringBuilder query = new StringBuilder();
30+
query.append( "select e from " )
31+
.append( typeContext.jpaEntityName() )
32+
.append( " e where e." )
33+
.append( typeContext.uniquePropertyName() )
34+
.append( " in(:" )
35+
.append( ID_PARAMETER_NAME )
36+
.append( ")" );
37+
38+
this.query = options.context( Session.class ).createQuery( query.toString(), typeContext.javaClass() )
39+
.setReadOnly( true )
40+
.setCacheable( false )
41+
.setLockMode( LockModeType.NONE )
42+
.setCacheMode( options.cacheMode() )
43+
.setQueryFlushMode( QueryFlushMode.NO_FLUSH )
44+
.setFetchSize( options.batchSize() );
45+
}
46+
47+
@Override
48+
public void close() {
49+
}
50+
51+
@Override
52+
public void load(List<Object> identifiers) {
53+
sink.accept( query.setParameter( ID_PARAMETER_NAME, identifiers ).list() );
54+
}
55+
56+
}

‎mapper/orm-jakarta-batch/core/src/main/java/org/hibernate/search/jakarta/batch/core/massindexing/loading/impl/BatchCoreDefaultHibernateOrmBatchIdentifierLoader.java

Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,193 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
* Copyright Red Hat Inc. and Hibernate Authors
4+
*/
5+
package org.hibernate.search.jakarta.batch.core.massindexing.loading.impl;
6+
7+
import java.util.HashSet;
8+
import java.util.Optional;
9+
import java.util.OptionalLong;
10+
import java.util.Set;
11+
import java.util.stream.Collectors;
12+
13+
import jakarta.persistence.LockModeType;
14+
15+
import org.hibernate.ScrollMode;
16+
import org.hibernate.ScrollableResults;
17+
import org.hibernate.StatelessSession;
18+
import org.hibernate.query.Query;
19+
import org.hibernate.search.jakarta.batch.core.massindexing.util.impl.IdOrder;
20+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchIdentifierLoader;
21+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchIdentifierLoadingOptions;
22+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchLoadingTypeContext;
23+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchReindexCondition;
24+
import org.hibernate.search.util.common.AssertionFailure;
25+
import org.hibernate.search.util.common.impl.Closer;
26+
27+
public class BatchCoreDefaultHibernateOrmBatchIdentifierLoader<E> implements HibernateOrmBatchIdentifierLoader {
28+
29+
private final StatelessSession session;
30+
private final String ormEntityName;
31+
private final String uniquePropertyName;
32+
private final IdOrder idOrder;
33+
private final HibernateOrmBatchIdentifierLoadingOptions options;
34+
private final IdLoader idLoader;
35+
36+
public BatchCoreDefaultHibernateOrmBatchIdentifierLoader(HibernateOrmBatchLoadingTypeContext<E> typeContext,
37+
HibernateOrmBatchIdentifierLoadingOptions options, IdOrder idOrder) {
38+
this.session = options.context( StatelessSession.class );
39+
this.ormEntityName = typeContext.jpaEntityName();
40+
this.uniquePropertyName = typeContext.uniquePropertyName();
41+
this.idOrder = idOrder;
42+
this.options = options;
43+
this.idLoader = options.maxResults().orElse( -1 ) == 1 ? new QuerySingleIdLoader() : new ScrollIdLoader();
44+
}
45+
46+
@Override
47+
public void close() {
48+
try ( Closer<RuntimeException> closer = new Closer<>() ) {
49+
if ( idLoader != null ) {
50+
closer.push( IdLoader::close, idLoader );
51+
}
52+
}
53+
}
54+
55+
@Override
56+
public OptionalLong totalCount() {
57+
StringBuilder query = new StringBuilder();
58+
query.append( "select count(e) from " )
59+
.append( ormEntityName )
60+
.append( " e " );
61+
62+
return OptionalLong.of( createQuery( session, query,
63+
options.reindexOnlyCondition().map( Set::of ).orElseGet( Set::of ), Long.class, Optional.empty() )
64+
.uniqueResult() );
65+
}
66+
67+
@Override
68+
public Object next() {
69+
return idLoader.next();
70+
}
71+
72+
@Override
73+
public boolean hasNext() {
74+
return idLoader.hasNext();
75+
}
76+
77+
private Query<Object> createQueryLoading(StatelessSession session) {
78+
StringBuilder query = new StringBuilder();
79+
query.append( "select e." )
80+
.append( uniquePropertyName )
81+
.append( " from " )
82+
.append( ormEntityName )
83+
.append( " e " );
84+
Set<HibernateOrmBatchReindexCondition> conditions = new HashSet<>();
85+
options.reindexOnlyCondition().ifPresent( conditions::add );
86+
options.lowerBound().ifPresent( b -> conditions
87+
.add( idOrder.idGreater( "HIBERNATE_SEARCH_ID_LOWER_BOUND_", b, options.lowerBoundInclusive() ) ) );
88+
options.upperBound().ifPresent( b -> conditions
89+
.add( idOrder.idLesser( "HIBERNATE_SEARCH_ID_UPPER_BOUND_", b, options.upperBoundInclusive() ) ) );
90+
91+
Query<Object> select = createQuery( session, query, conditions, Object.class, Optional.of( idOrder.ascOrder() ) )
92+
.setFetchSize( options.fetchSize() )
93+
.setReadOnly( true )
94+
.setCacheable( false )
95+
.setLockMode( LockModeType.NONE );
96+
options.offset().ifPresent( select::setFirstResult );
97+
options.maxResults().ifPresent( select::setMaxResults );
98+
return select;
99+
}
100+
101+
private <T> Query<T> createQuery(StatelessSession session,
102+
StringBuilder hql, Set<HibernateOrmBatchReindexCondition> conditions, Class<T> returnedType,
103+
Optional<String> order) {
104+
if ( !conditions.isEmpty() ) {
105+
hql.append( " where " );
106+
hql.append( conditions.stream()
107+
.map( c -> "( " + c.conditionString() + " )" )
108+
.collect( Collectors.joining( " AND ", " ", " " ) )
109+
);
110+
}
111+
order.ifPresent( o -> hql.append( " ORDER BY " ).append( o ) );
112+
Query<T> query = session.createQuery( hql.toString(), returnedType )
113+
.setCacheable( false );
114+
115+
for ( var condition : conditions ) {
116+
for ( var entry : condition.params().entrySet() ) {
117+
query.setParameter( entry.getKey(), entry.getValue() );
118+
}
119+
}
120+
121+
return query;
122+
}
123+
124+
private interface IdLoader {
125+
Object next();
126+
127+
boolean hasNext();
128+
129+
void close();
130+
}
131+
132+
private class QuerySingleIdLoader implements IdLoader {
133+
134+
private boolean hasNextCalled = false;
135+
private boolean nextCalled = false;
136+
137+
private Query<Object> id = createQueryLoading( session );
138+
private Object currentId;
139+
140+
@Override
141+
public Object next() {
142+
if ( hasNextCalled ) {
143+
nextCalled = true;
144+
hasNextCalled = false;
145+
return currentId;
146+
}
147+
else {
148+
throw new AssertionFailure( "Cannot call next() before calling hasNext()" );
149+
}
150+
}
151+
152+
@Override
153+
public boolean hasNext() {
154+
if ( nextCalled ) {
155+
// we expect to have just a single ID, so if we called next and got the id we don't need to execute the query anymore:
156+
return false;
157+
}
158+
currentId = id.getSingleResultOrNull();
159+
hasNextCalled = true;
160+
return currentId != null;
161+
}
162+
163+
@Override
164+
public void close() {
165+
id = null;
166+
}
167+
}
168+
169+
private class ScrollIdLoader implements IdLoader {
170+
private ScrollableResults<Object> id = createQueryLoading( session ).scroll( ScrollMode.FORWARD_ONLY );
171+
172+
@Override
173+
public Object next() {
174+
return id.get();
175+
}
176+
177+
@Override
178+
public boolean hasNext() {
179+
return id.next();
180+
}
181+
182+
@Override
183+
public void close() {
184+
try ( Closer<RuntimeException> closer = new Closer<>() ) {
185+
if ( id != null ) {
186+
closer.push( ScrollableResults::close, id );
187+
id = null;
188+
}
189+
}
190+
}
191+
}
192+
193+
}

0 commit comments

Comments
 (0)