Skip to content

Commit 10928d5

Browse files
nodecenodece
committed
[feat] Build pulsar-sql based on pulsar 4.0.0
1 parent 4801845 commit 10928d5

File tree

16 files changed

+651
-89
lines changed

16 files changed

+651
-89
lines changed

‎.gitignore‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
.idea
2+
target/
3+
dependency-reduced-pom.xml

‎pulsar-sql/pom.xml‎

Lines changed: 34 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
<parent>
2626
<groupId>org.apache.pulsar</groupId>
2727
<artifactId>pulsar</artifactId>
28-
<version>3.0.8-SNAPSHOT</version>
28+
<version>4.0.0</version>
2929
</parent>
3030

3131
<artifactId>pulsar-sql</artifactId>
@@ -37,6 +37,7 @@
3737
<!-- use okio version that matches the okhttp3 version -->
3838
<okio.version>1.17.2</okio.version>
3939
<airlift.version>213</airlift.version>
40+
<trino.version>368</trino.version>
4041
</properties>
4142

4243
<dependencyManagement>
@@ -106,18 +107,45 @@
106107
<build>
107108
<plugins>
108109
<plugin>
109-
<groupId>org.apache.maven.plugins</groupId>
110-
<artifactId>maven-checkstyle-plugin</artifactId>
110+
<groupId>org.commonjava.maven.plugins</groupId>
111+
<artifactId>directory-maven-plugin</artifactId>
112+
<version>${directory-maven-plugin.version}</version>
111113
<executions>
112114
<execution>
113-
<id>checkstyle</id>
114-
<phase>verify</phase>
115+
<id>directories</id>
115116
<goals>
116-
<goal>check</goal>
117+
<goal>directory-of</goal>
117118
</goals>
119+
<phase>initialize</phase>
120+
<configuration>
121+
<property>pulsar.basedir</property>
122+
<project>
123+
<groupId>org.apache.pulsar</groupId>
124+
<artifactId>pulsar-sql</artifactId>
125+
</project>
126+
</configuration>
118127
</execution>
119128
</executions>
120129
</plugin>
130+
<plugin>
131+
<groupId>org.apache.maven.plugins</groupId>
132+
<artifactId>maven-checkstyle-plugin</artifactId>
133+
<version>${maven-checkstyle-plugin.version}</version>
134+
<dependencies>
135+
<dependency>
136+
<groupId>com.puppycrawl.tools</groupId>
137+
<artifactId>checkstyle</artifactId>
138+
<version>${puppycrawl.checkstyle.version}</version>
139+
</dependency>
140+
</dependencies>
141+
<configuration>
142+
<configLocation>${pulsar.basedir}/src/checkstyle.xml</configLocation>
143+
<suppressionsLocation>${pulsar.basedir}/src/suppressions.xml</suppressionsLocation>
144+
<includeTestSourceDirectory>true</includeTestSourceDirectory>
145+
<encoding>UTF-8</encoding>
146+
<excludes>**/proto/*</excludes>
147+
</configuration>
148+
</plugin>
121149
</plugins>
122150
</build>
123151

‎pulsar-sql/presto-distribution/pom.xml‎

Lines changed: 1 addition & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
<parent>
2626
<groupId>org.apache.pulsar</groupId>
2727
<artifactId>pulsar-sql</artifactId>
28-
<version>3.0.8-SNAPSHOT</version>
28+
<version>4.0.0</version>
2929
</parent>
3030

3131
<artifactId>pulsar-presto-distribution</artifactId>
@@ -329,22 +329,6 @@
329329
</execution>
330330
</executions>
331331
</plugin>
332-
333-
<plugin>
334-
<groupId>com.mycila</groupId>
335-
<artifactId>license-maven-plugin</artifactId>
336-
<version>${license-maven-plugin.version}</version>
337-
<configuration>
338-
<licenseSets>
339-
<licenseSet>
340-
<header>../../src/license-header.txt</header>
341-
</licenseSet>
342-
</licenseSets>
343-
<mapping>
344-
<java>SLASHSTAR_STYLE</java>
345-
</mapping>
346-
</configuration>
347-
</plugin>
348332
</plugins>
349333
<extensions>
350334
<extension>

‎pulsar-sql/presto-pulsar-plugin/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
<parent>
2626
<groupId>org.apache.pulsar</groupId>
2727
<artifactId>pulsar-sql</artifactId>
28-
<version>3.0.8-SNAPSHOT</version>
28+
<version>4.0.0</version>
2929
</parent>
3030

3131
<artifactId>pulsar-presto-connector</artifactId>

‎pulsar-sql/presto-pulsar/pom.xml‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
<parent>
2626
<groupId>org.apache.pulsar</groupId>
2727
<artifactId>pulsar-sql</artifactId>
28-
<version>3.0.8-SNAPSHOT</version>
28+
<version>4.0.0</version>
2929
</parent>
3030

3131
<artifactId>pulsar-presto-connector-original</artifactId>
@@ -143,6 +143,13 @@
143143
<scope>test</scope>
144144
</dependency>
145145

146+
<dependency>
147+
<groupId>org.apache.pulsar</groupId>
148+
<artifactId>managed-ledger</artifactId>
149+
<version>${project.version}</version>
150+
<scope>test</scope>
151+
</dependency>
152+
146153
<dependency>
147154
<groupId>org.apache.pulsar</groupId>
148155
<artifactId>testmocks</artifactId>

‎pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import static io.airlift.slice.SizeOf.SIZE_OF_LONG;
2424
import static io.trino.decoder.FieldValueProviders.bytesValueProvider;
2525
import static io.trino.decoder.FieldValueProviders.longValueProvider;
26+
import static org.apache.bookkeeper.mledger.PositionFactory.LATEST;
2627
import com.fasterxml.jackson.core.JsonProcessingException;
2728
import com.fasterxml.jackson.databind.ObjectMapper;
2829
import com.google.common.annotations.VisibleForTesting;
@@ -48,6 +49,7 @@
4849
import java.util.Optional;
4950
import java.util.Set;
5051
import java.util.concurrent.CompletableFuture;
52+
import java.util.concurrent.ConcurrentHashMap;
5153
import java.util.concurrent.ExecutionException;
5254
import java.util.concurrent.atomic.AtomicBoolean;
5355
import java.util.concurrent.atomic.AtomicLong;
@@ -58,7 +60,6 @@
5860
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
5961
import org.apache.bookkeeper.mledger.Position;
6062
import org.apache.bookkeeper.mledger.ReadOnlyCursor;
61-
import org.apache.bookkeeper.mledger.impl.PositionImpl;
6263
import org.apache.bookkeeper.mledger.impl.ReadOnlyCursorImpl;
6364
import org.apache.pulsar.client.api.PulsarClientException;
6465
import org.apache.pulsar.client.api.Schema;
@@ -75,7 +76,6 @@
7576
import org.apache.pulsar.common.schema.KeyValueEncodingType;
7677
import org.apache.pulsar.common.schema.SchemaInfo;
7778
import org.apache.pulsar.common.schema.SchemaType;
78-
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
7979
import org.apache.pulsar.sql.presto.util.CacheSizeAllocator;
8080
import org.apache.pulsar.sql.presto.util.NoStrictCacheSizeAllocator;
8181
import org.apache.pulsar.sql.presto.util.NullCacheSizeAllocator;
@@ -123,8 +123,7 @@ public class PulsarRecordCursor implements RecordCursor {
123123

124124
PulsarDispatchingRowDecoderFactory decoderFactory;
125125

126-
protected ConcurrentOpenHashMap<String, ChunkedMessageCtx> chunkedMessagesMap =
127-
ConcurrentOpenHashMap.<String, ChunkedMessageCtx>newBuilder().build();
126+
protected ConcurrentHashMap<String, ChunkedMessageCtx> chunkedMessagesMap = new ConcurrentHashMap<>();
128127

129128
private static final Logger log = Logger.get(PulsarRecordCursor.class);
130129

@@ -366,7 +365,7 @@ public void accept(Entry entry) {
366365
}
367366

368367
private boolean entryExceedSplitEndPosition(Entry entry) {
369-
return ((PositionImpl) entry.getPosition()).compareTo(pulsarSplit.getEndPosition()) >= 0;
368+
return entry.getPosition().compareTo(pulsarSplit.getEndPosition()) >= 0;
370369
}
371370

372371
@VisibleForTesting
@@ -383,7 +382,7 @@ public void run() {
383382

384383
if (outstandingReadsRequests.get() > 0) {
385384
if (!cursor.hasMoreEntries()
386-
|| (((PositionImpl) cursor.getReadPosition()).compareTo(pulsarSplit.getEndPosition()) >= 0
385+
|| (cursor.getReadPosition().compareTo(pulsarSplit.getEndPosition()) >= 0
387386
&& chunkedMessagesMap.isEmpty())) {
388387
isDone = true;
389388

@@ -401,7 +400,7 @@ public void run() {
401400

402401
long numEntries = readOnlyCursorImpl.getCurrentLedgerInfo().getEntries();
403402
long entriesToSkip =
404-
(numEntries - ((PositionImpl) cursor.getReadPosition()).getEntryId()) + 1;
403+
(numEntries - (cursor.getReadPosition()).getEntryId()) + 1;
405404
cursor.skipEntries(Math.toIntExact((entriesToSkip)));
406405

407406
entriesProcessed += entriesToSkip;
@@ -413,7 +412,7 @@ public void run() {
413412
// if the available size is invalid and the entry queue size is 0, read one entry
414413
outstandingReadsRequests.decrementAndGet();
415414
cursor.asyncReadEntries(batchSize, entryQueueCacheSizeAllocator.getAvailableCacheSize(),
416-
this, System.nanoTime(), PositionImpl.LATEST);
415+
this, System.nanoTime(), LATEST);
417416
}
418417

419418
// stats for successful read request

‎pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarSplit.java‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,8 @@
3131
import java.io.IOException;
3232
import java.util.List;
3333
import java.util.Map;
34-
import org.apache.bookkeeper.mledger.impl.PositionImpl;
34+
import org.apache.bookkeeper.mledger.Position;
35+
import org.apache.bookkeeper.mledger.impl.ImmutablePositionImpl;
3536
import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
3637
import org.apache.pulsar.common.schema.SchemaInfo;
3738
import org.apache.pulsar.common.schema.SchemaType;
@@ -58,8 +59,8 @@ public class PulsarSplit implements ConnectorSplit {
5859
private final TupleDomain<ColumnHandle> tupleDomain;
5960
private final SchemaInfo schemaInfo;
6061

61-
private final PositionImpl startPosition;
62-
private final PositionImpl endPosition;
62+
private final Position startPosition;
63+
private final Position endPosition;
6364
private final String schemaInfoProperties;
6465

6566
private final OffloadPoliciesImpl offloadPolicies;
@@ -95,8 +96,8 @@ public PulsarSplit(
9596
this.startPositionLedgerId = startPositionLedgerId;
9697
this.endPositionLedgerId = endPositionLedgerId;
9798
this.tupleDomain = requireNonNull(tupleDomain, "tupleDomain is null");
98-
this.startPosition = PositionImpl.get(startPositionLedgerId, startPositionEntryId);
99-
this.endPosition = PositionImpl.get(endPositionLedgerId, endPositionEntryId);
99+
this.startPosition = new ImmutablePositionImpl(startPositionLedgerId, startPositionEntryId);
100+
this.endPosition = new ImmutablePositionImpl(endPositionLedgerId, endPositionEntryId);
100101
this.schemaInfoProperties = schemaInfoProperties;
101102
this.offloadPolicies = offloadPolicies;
102103

@@ -174,11 +175,11 @@ public TupleDomain<ColumnHandle> getTupleDomain() {
174175
return tupleDomain;
175176
}
176177

177-
public PositionImpl getStartPosition() {
178+
public Position getStartPosition() {
178179
return startPosition;
179180
}
180181

181-
public PositionImpl getEndPosition() {
182+
public Position getEndPosition() {
182183
return endPosition;
183184
}
184185

‎pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarSplitManager.java‎

Lines changed: 19 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import static io.trino.spi.StandardErrorCode.QUERY_REJECTED;
2323
import static java.util.Objects.requireNonNull;
2424
import static org.apache.bookkeeper.mledger.ManagedCursor.FindPositionConstraint.SearchAllAvailableEntries;
25+
import static org.apache.bookkeeper.mledger.PositionFactory.EARLIEST;
2526
import static org.apache.pulsar.sql.presto.PulsarConnectorUtils.restoreNamespaceDelimiterIfNeeded;
2627
import com.fasterxml.jackson.databind.ObjectMapper;
2728
import com.google.common.annotations.VisibleForTesting;
@@ -50,8 +51,8 @@
5051
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
5152
import org.apache.bookkeeper.mledger.ManagedLedgerException;
5253
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
54+
import org.apache.bookkeeper.mledger.Position;
5355
import org.apache.bookkeeper.mledger.ReadOnlyCursor;
54-
import org.apache.bookkeeper.mledger.impl.PositionImpl;
5556
import org.apache.commons.lang3.exception.ExceptionUtils;
5657
import org.apache.pulsar.client.admin.PulsarAdmin;
5758
import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -271,7 +272,7 @@ Collection<PulsarSplit> getSplitsForTopic(String topicNamePersistenceEncoding,
271272
try {
272273
readOnlyCursor = managedLedgerFactory.openReadOnlyCursor(
273274
topicNamePersistenceEncoding,
274-
PositionImpl.EARLIEST, managedLedgerConfig);
275+
EARLIEST, managedLedgerConfig);
275276

276277
long numEntries = readOnlyCursor.getNumberOfEntries();
277278
if (numEntries <= 0) {
@@ -286,12 +287,12 @@ Collection<PulsarSplit> getSplitsForTopic(String topicNamePersistenceEncoding,
286287
topicNamePersistenceEncoding,
287288
numEntries);
288289

289-
PositionImpl initialStartPosition;
290+
Position initialStartPosition;
290291
if (predicatePushdownInfo != null) {
291292
numEntries = predicatePushdownInfo.getNumOfEntries();
292293
initialStartPosition = predicatePushdownInfo.getStartPosition();
293294
} else {
294-
initialStartPosition = (PositionImpl) readOnlyCursor.getReadPosition();
295+
initialStartPosition = readOnlyCursor.getReadPosition();
295296
}
296297

297298

@@ -307,9 +308,9 @@ Collection<PulsarSplit> getSplitsForTopic(String topicNamePersistenceEncoding,
307308
List<PulsarSplit> splits = new LinkedList<>();
308309
for (int i = 0; i < numSplits; i++) {
309310
long entriesForSplit = (remainder > i) ? avgEntriesPerSplit + 1 : avgEntriesPerSplit;
310-
PositionImpl startPosition = (PositionImpl) readOnlyCursor.getReadPosition();
311+
Position startPosition = readOnlyCursor.getReadPosition();
311312
readOnlyCursor.skipEntries(Math.toIntExact(entriesForSplit));
312-
PositionImpl endPosition = (PositionImpl) readOnlyCursor.getReadPosition();
313+
Position endPosition = readOnlyCursor.getReadPosition();
313314

314315
PulsarSplit pulsarSplit = new PulsarSplit(i, this.connectorId,
315316
restoreNamespaceDelimiterIfNeeded(tableHandle.getSchemaName(), pulsarConnectorConfig),
@@ -341,11 +342,11 @@ Collection<PulsarSplit> getSplitsForTopic(String topicNamePersistenceEncoding,
341342

342343
@Data
343344
private static class PredicatePushdownInfo {
344-
private PositionImpl startPosition;
345-
private PositionImpl endPosition;
345+
private Position startPosition;
346+
private Position endPosition;
346347
private long numOfEntries;
347348

348-
private PredicatePushdownInfo(PositionImpl startPosition, PositionImpl endPosition, long numOfEntries) {
349+
private PredicatePushdownInfo(Position startPosition, Position endPosition, long numOfEntries) {
349350
this.startPosition = startPosition;
350351
this.endPosition = endPosition;
351352
this.numOfEntries = numOfEntries;
@@ -363,7 +364,7 @@ public static PredicatePushdownInfo getPredicatePushdownInfo(String connectorId,
363364
try {
364365
readOnlyCursor = managedLedgerFactory.openReadOnlyCursor(
365366
topicNamePersistenceEncoding,
366-
PositionImpl.EARLIEST, managedLedgerConfig);
367+
EARLIEST, managedLedgerConfig);
367368

368369
if (tupleDomain.getDomains().isPresent()) {
369370
Domain domain = tupleDomain.getDomains().get().get(PulsarInternalColumn.PUBLISH_TIME
@@ -390,20 +391,20 @@ public static PredicatePushdownInfo getPredicatePushdownInfo(String connectorId,
390391
lowerBoundTs = block.getLong(0, 0) / 1000;
391392
}
392393

393-
PositionImpl overallStartPos;
394+
Position overallStartPos;
394395
if (lowerBoundTs == null) {
395-
overallStartPos = (PositionImpl) readOnlyCursor.getReadPosition();
396+
overallStartPos = readOnlyCursor.getReadPosition();
396397
} else {
397398
overallStartPos = findPosition(readOnlyCursor, lowerBoundTs);
398399
if (overallStartPos == null) {
399-
overallStartPos = (PositionImpl) readOnlyCursor.getReadPosition();
400+
overallStartPos = readOnlyCursor.getReadPosition();
400401
}
401402
}
402403

403-
PositionImpl overallEndPos;
404+
Position overallEndPos;
404405
if (upperBoundTs == null) {
405406
readOnlyCursor.skipEntries(Math.toIntExact(totalNumEntries));
406-
overallEndPos = (PositionImpl) readOnlyCursor.getReadPosition();
407+
overallEndPos = readOnlyCursor.getReadPosition();
407408
} else {
408409
overallEndPos = findPosition(readOnlyCursor, upperBoundTs);
409410
if (overallEndPos == null) {
@@ -414,7 +415,7 @@ public static PredicatePushdownInfo getPredicatePushdownInfo(String connectorId,
414415
// Just use a close bound since presto can always filter out the extra entries even if
415416
// the bound
416417
// should be open or a mixture of open and closed
417-
com.google.common.collect.Range<PositionImpl> posRange =
418+
com.google.common.collect.Range<Position> posRange =
418419
com.google.common.collect.Range.range(overallStartPos,
419420
com.google.common.collect.BoundType.CLOSED,
420421
overallEndPos, com.google.common.collect.BoundType.CLOSED);
@@ -437,10 +438,10 @@ public static PredicatePushdownInfo getPredicatePushdownInfo(String connectorId,
437438
}
438439
}
439440

440-
private static PositionImpl findPosition(ReadOnlyCursor readOnlyCursor, long timestamp) throws
441+
private static Position findPosition(ReadOnlyCursor readOnlyCursor, long timestamp) throws
441442
ManagedLedgerException,
442443
InterruptedException {
443-
return (PositionImpl) readOnlyCursor.findNewestMatching(
444+
return readOnlyCursor.findNewestMatching(
444445
SearchAllAvailableEntries,
445446
entry -> {
446447
try {

0 commit comments

Comments
 (0)