Skip to content

Commit a4f38cb

Browse files
marko-bekhtamarko-bekhta
committed
HSEARCH-5076 WIP batch rewrite
1 parent 3037821 commit a4f38cb

32 files changed

+1048
-277
lines changed

‎build/jqassistant/rules/rules.xml

+1
Original file line numberDiff line numberDiff line change
@@ -278,6 +278,7 @@
278278
WHEN 'hibernate-search-mapper-pojo-standalone' THEN 'StandalonePojo'
279279
WHEN 'hibernate-search-mapper-orm' THEN 'HibernateOrm'
280280
WHEN 'hibernate-search-mapper-orm-outbox-polling' THEN 'OutboxPolling'
281+
WHEN 'hibernate-search-mapper-orm-jakarta-batch-core' THEN 'BatchCore'
281282
WHEN 'hibernate-search-mapper-orm-jakarta-batch-jberet' THEN 'JBeret'
282283
ELSE 'UNKNOWN-MODULE-SPECIFIC-KEYWORD-PLEASE-UPDATE-JQASSISTANT-RULES'
283284
END

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

+3-3
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@
2323
import org.hibernate.engine.spi.SessionFactoryImplementor;
2424
import org.hibernate.search.jakarta.batch.core.logging.impl.Log;
2525
import org.hibernate.search.jakarta.batch.core.massindexing.MassIndexingJob;
26-
import org.hibernate.search.jakarta.batch.core.massindexing.util.impl.EntityTypeDescriptor;
26+
import org.hibernate.search.jakarta.batch.core.massindexing.util.impl.BatchCoreEntityTypeDescriptor;
2727
import org.hibernate.search.mapper.orm.Search;
2828
import org.hibernate.search.mapper.orm.loading.spi.HibernateOrmLoadingTypeContext;
2929
import org.hibernate.search.mapper.orm.mapping.SearchMapping;
@@ -145,11 +145,11 @@ private static <T> List<T> find(Session session, Class<T> clazz, String key, Str
145145
.fetchHits( 1000 );
146146
}
147147

148-
public static EntityTypeDescriptor<?, ?> createEntityTypeDescriptor(EntityManagerFactory emf, Class<?> clazz) {
148+
public static BatchCoreEntityTypeDescriptor<?, ?> createEntityTypeDescriptor(EntityManagerFactory emf, Class<?> clazz) {
149149
SearchMapping mapping = Search.mapping( emf );
150150
BatchMappingContext mappingContext = (BatchMappingContext) mapping;
151151
HibernateOrmLoadingTypeContext<?> type = mappingContext.typeContextProvider()
152152
.byEntityName().getOrFail( mapping.indexedEntity( clazz ).jpaName() );
153-
return EntityTypeDescriptor.create( emf.unwrap( SessionFactoryImplementor.class ), type );
153+
return BatchCoreEntityTypeDescriptor.create( emf.unwrap( SessionFactoryImplementor.class ), type );
154154
}
155155
}

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

+8-8
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

+56
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.FlushMode;
12+
import org.hibernate.Session;
13+
import org.hibernate.query.Query;
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+
.setHibernateFlushMode( FlushMode.MANUAL )
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

+193
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+
}

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

+46
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
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 org.hibernate.metamodel.mapping.EmbeddableMappingType;
8+
import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
9+
import org.hibernate.search.jakarta.batch.core.massindexing.util.impl.CompositeIdOrder;
10+
import org.hibernate.search.jakarta.batch.core.massindexing.util.impl.IdOrder;
11+
import org.hibernate.search.jakarta.batch.core.massindexing.util.impl.SingularIdOrder;
12+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchEntityLoader;
13+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchEntityLoadingOptions;
14+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchEntitySink;
15+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchIdentifierLoader;
16+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchIdentifierLoadingOptions;
17+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchLoadingStrategy;
18+
import org.hibernate.search.mapper.orm.loading.batch.HibernateOrmBatchLoadingTypeContext;
19+
import org.hibernate.search.mapper.orm.loading.spi.HibernateOrmLoadingTypeContext;
20+
21+
public class BatchCoreDefaultHibernateOrmBatchLoadingStrategy<E, I> implements HibernateOrmBatchLoadingStrategy<E, I> {
22+
23+
private final IdOrder idOrder;
24+
25+
public BatchCoreDefaultHibernateOrmBatchLoadingStrategy(HibernateOrmLoadingTypeContext<E> type) {
26+
EntityIdentifierMapping identifierMapping = type.entityMappingType().getIdentifierMapping();
27+
if ( identifierMapping.getPartMappingType() instanceof EmbeddableMappingType ) {
28+
idOrder = new CompositeIdOrder<>( type );
29+
}
30+
else {
31+
idOrder = new SingularIdOrder<>( type );
32+
}
33+
}
34+
35+
@Override
36+
public HibernateOrmBatchIdentifierLoader createIdentifierLoader(HibernateOrmBatchLoadingTypeContext<E> typeContext,
37+
HibernateOrmBatchIdentifierLoadingOptions options) {
38+
return new BatchCoreDefaultHibernateOrmBatchIdentifierLoader<>( typeContext, options, idOrder );
39+
}
40+
41+
@Override
42+
public HibernateOrmBatchEntityLoader createEntityLoader(HibernateOrmBatchLoadingTypeContext<E> typeContext,
43+
HibernateOrmBatchEntitySink<E> sink, HibernateOrmBatchEntityLoadingOptions options) {
44+
return new BatchCoreDefaultHibernateOrmBatchEntityLoader<>( typeContext, sink, options );
45+
}
46+
}

0 commit comments

Comments
 (0)