You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/08/24 17:58:43 UTC
[GitHub] srkukarni closed pull request #2414: allow predicates concerning
publish time to push down to pulsar
srkukarni closed pull request #2414: allow predicates concerning publish time to push down to pulsar
URL: https://github.com/apache/incubator-pulsar/pull/2414
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java
index 584a376fac..a1187b8255 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java
@@ -395,7 +395,7 @@ void asyncSkipEntries(int numEntriesToSkip, IndividualDeletedEntries deletedEntr
final SkipEntriesCallback callback, Object ctx);
/**
- * Find the newest entry that matches the given predicate.
+ * Find the newest entry that matches the given predicate. Will only search among active entries
*
* @param condition
* predicate that reads an entry an applies a condition
@@ -405,9 +405,25 @@ void asyncSkipEntries(int numEntriesToSkip, IndividualDeletedEntries deletedEntr
*/
Position findNewestMatching(Predicate<Entry> condition) throws InterruptedException, ManagedLedgerException;
+
+ /**
+ * Find the newest entry that matches the given predicate.
+ *
+ * @param constraint
+ * search only active entries or all entries
+ * @param condition
+ * predicate that reads an entry an applies a condition
+ * @return Position of the newest entry that matches the given predicate
+ * @throws InterruptedException
+ * @throws ManagedLedgerException
+ */
+ Position findNewestMatching(FindPositionConstraint constraint, Predicate<Entry> condition) throws InterruptedException, ManagedLedgerException;
+
/**
* Find the newest entry that matches the given predicate.
*
+ * @param constraint
+ * search only active entries or all entries
* @param condition
* predicate that reads an entry an applies a condition
* @param callback
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ReadOnlyCursor.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ReadOnlyCursor.java
index 995bdf0c76..10997427cb 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ReadOnlyCursor.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ReadOnlyCursor.java
@@ -20,14 +20,16 @@
import java.util.List;
+import com.google.common.base.Predicate;
+import com.google.common.collect.Range;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
public interface ReadOnlyCursor {
/**
* Read entries from the ManagedLedger, up to the specified number. The returned list can be smaller.
*
- * @param numberOfEntriesToRead
- * maximum number of entries to return
+ * @param numberOfEntriesToRead maximum number of entries to return
* @return the list of entries
* @throws ManagedLedgerException
*/
@@ -36,13 +38,10 @@
/**
* Asynchronously read entries from the ManagedLedger.
*
+ * @param numberOfEntriesToRead maximum number of entries to return
+ * @param callback callback object
+ * @param ctx opaque context
* @see #readEntries(int)
- * @param numberOfEntriesToRead
- * maximum number of entries to return
- * @param callback
- * callback object
- * @param ctx
- * opaque context
*/
void asyncReadEntries(int numberOfEntriesToRead, ReadEntriesCallback callback, Object ctx);
@@ -55,7 +54,7 @@
/**
* Tells whether this cursor has already consumed all the available entries.
- *
+ * <p>
* <p/>
* This method is not blocking.
*
@@ -65,7 +64,7 @@
/**
* Return the number of messages that this cursor still has to read.
- *
+ * <p>
* <p/>
* This method has linear time complexity on the number of ledgers included in the managed ledger.
*
@@ -76,11 +75,31 @@
/**
* Skip n entries from the read position of this cursor.
*
- * @param numEntriesToSkip
- * number of entries to skip
+ * @param numEntriesToSkip number of entries to skip
*/
void skipEntries(int numEntriesToSkip);
+ /**
+ * Find the newest entry that matches the given predicate.
+ *
+ * @param constraint search only active entries or all entries
+ * @param condition predicate that reads an entry an applies a condition
+ * @return Position of the newest entry that matches the given predicate
+ * @throws InterruptedException
+ * @throws ManagedLedgerException
+ */
+ Position findNewestMatching(ManagedCursor.FindPositionConstraint constraint, Predicate<Entry> condition) throws InterruptedException, ManagedLedgerException;
+
+ /**
+ * Return the number of messages that this cursor still has to read.
+ *
+ * <p/>This method has linear time complexity on the number of ledgers included in the managed ledger.
+ *
+ * @param range the range between two positions
+ * @return the number of entries in range
+ */
+ long getNumberOfEntries(Range<PositionImpl> range);
+
/**
* Close the cursor and releases the associated resources.
*
@@ -92,10 +111,8 @@
/**
* Close the cursor asynchronously and release the associated resources.
*
- * @param callback
- * callback object
- * @param ctx
- * opaque context
+ * @param callback callback object
+ * @param ctx opaque context
*/
void asyncClose(AsyncCallbacks.CloseCallback callback, Object ctx);
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index bab354762a..50b47229d6 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -687,6 +687,11 @@ public long getNumberOfEntriesInStorage() {
@Override
public Position findNewestMatching(Predicate<Entry> condition) throws InterruptedException, ManagedLedgerException {
+ return findNewestMatching(FindPositionConstraint.SearchActiveEntries, condition);
+ }
+
+ @Override
+ public Position findNewestMatching(FindPositionConstraint constraint, Predicate<Entry> condition) throws InterruptedException, ManagedLedgerException {
final CountDownLatch counter = new CountDownLatch(1);
class Result {
ManagedLedgerException exception = null;
@@ -694,7 +699,7 @@ public Position findNewestMatching(Predicate<Entry> condition) throws Interrupte
}
final Result result = new Result();
- asyncFindNewestMatching(FindPositionConstraint.SearchActiveEntries, condition, new FindEntryCallback() {
+ asyncFindNewestMatching(constraint, condition, new FindEntryCallback() {
@Override
public void findEntryComplete(Position position, Object ctx) {
result.position = position;
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyCursorImpl.java
index df0c08c707..c1a2216b4b 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyCursorImpl.java
@@ -32,7 +32,7 @@
public class ReadOnlyCursorImpl extends ManagedCursorImpl implements ReadOnlyCursor {
public ReadOnlyCursorImpl(BookKeeper bookkeeper, ManagedLedgerConfig config, ManagedLedgerImpl ledger,
- PositionImpl startPosition, String cursorName) {
+ PositionImpl startPosition, String cursorName) {
super(bookkeeper, config, ledger, cursorName);
if (startPosition.equals(PositionImpl.earliest)) {
@@ -47,7 +47,7 @@ public ReadOnlyCursorImpl(BookKeeper bookkeeper, ManagedLedgerConfig config, Man
messagesConsumedCounter = -getNumberOfEntries(Range.closed(readPosition, ledger.getLastPosition()));
}
- state = State.NoLedger;
+ this.state = State.NoLedger;
}
@Override
@@ -62,4 +62,7 @@ public void asyncClose(final AsyncCallbacks.CloseCallback callback, final Object
callback.closeComplete(ctx);
}
+ public long getNumberOfEntries(Range<PositionImpl> range) {
+ return this.ledger.getNumberOfEntries(range);
+ }
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java
index 8e3ea7f119..052aa0e771 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java
@@ -21,9 +21,12 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import com.google.common.collect.Range;
+import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.common.util.OrderedExecutor;
import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerNotFoundException;
@@ -35,6 +38,7 @@
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
+@Slf4j
public class ReadOnlyManagedLedgerImpl extends ManagedLedgerImpl {
public ReadOnlyManagedLedgerImpl(ManagedLedgerFactoryImpl factory, BookKeeper bookKeeper, MetaStore store,
@@ -124,6 +128,22 @@ private ReadOnlyCursor createReadOnlyCursor(PositionImpl startPosition) {
return cursor;
}
+ @Override
+ void asyncReadEntry(PositionImpl position, AsyncCallbacks.ReadEntryCallback callback, Object ctx) {
+ this.getLedgerHandle(position.getLedgerId()).thenAccept((ledger) -> {
+ this.entryCache.asyncReadEntry(ledger, position, callback, ctx);
+ }).exceptionally((ex) -> {
+ log.error("[{}] Error opening ledger for reading at position {} - {}", new Object[]{this.name, position, ex.getMessage()});
+ callback.readEntryFailed(ManagedLedgerException.getManagedLedgerException(ex.getCause()), ctx);
+ return null;
+ });
+ }
+
+ @Override
+ public long getNumberOfEntries() {
+ return getNumberOfEntries(Range.openClosed(PositionImpl.earliest, getLastPosition()));
+ }
+
protected boolean isReadOnly() {
return true;
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java
index ecf6acfc50..3e404d9be0 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java
@@ -197,6 +197,11 @@ public Position findNewestMatching(Predicate<Entry> condition)
return null;
}
+ @Override
+ public Position findNewestMatching(FindPositionConstraint constraint, Predicate<Entry> condition) throws InterruptedException, ManagedLedgerException {
+ return null;
+ }
+
@Override
public void asyncFindNewestMatching(FindPositionConstraint constraint, Predicate<Entry> condition,
AsyncCallbacks.FindEntryCallback callback, Object ctx) {
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarInternalColumn.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarInternalColumn.java
index 0d391a6a73..8341775971 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarInternalColumn.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarInternalColumn.java
@@ -20,6 +20,7 @@
import com.facebook.presto.spi.type.BigintType;
import com.facebook.presto.spi.type.TimestampType;
+import com.facebook.presto.spi.type.TimestampWithTimeZoneType;
import com.facebook.presto.spi.type.Type;
import com.facebook.presto.spi.type.VarcharType;
import com.fasterxml.jackson.core.JsonProcessingException;
@@ -132,8 +133,9 @@ public Object getData(Message message) {
public static final PulsarInternalColumn EVENT_TIME = new EventTimeColumn("__event_time__", TimestampType
.TIMESTAMP, "Application defined timestamp in milliseconds of when the event occurred");
- public static final PulsarInternalColumn PUBLISH_TIME = new PublishTimeColumn("__publish_time__", TimestampType
- .TIMESTAMP, "The timestamp in milliseconds of when event as published");
+ public static final PulsarInternalColumn PUBLISH_TIME = new PublishTimeColumn("__publish_time__",
+ TimestampWithTimeZoneType.TIMESTAMP_WITH_TIME_ZONE,
+ "The timestamp in milliseconds of when event as published");
public static final PulsarInternalColumn MESSAGE_ID = new MessageIdColumn("__message_id__", VarcharType.VARCHAR,
"The message ID of the message used to generate this row");
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarMetadata.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarMetadata.java
index c04156eca7..20a56bb7bd 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarMetadata.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarMetadata.java
@@ -32,11 +32,14 @@
import com.facebook.presto.spi.SchemaTablePrefix;
import com.facebook.presto.spi.TableNotFoundException;
import com.facebook.presto.spi.connector.ConnectorMetadata;
+import com.facebook.presto.spi.predicate.Domain;
+import com.facebook.presto.spi.predicate.Range;
import com.facebook.presto.spi.type.BigintType;
import com.facebook.presto.spi.type.BooleanType;
import com.facebook.presto.spi.type.DoubleType;
import com.facebook.presto.spi.type.IntegerType;
import com.facebook.presto.spi.type.RealType;
+import com.facebook.presto.spi.type.SqlTimestampWithTimeZone;
import com.facebook.presto.spi.type.Type;
import com.facebook.presto.spi.type.VarbinaryType;
import com.facebook.presto.spi.type.VarcharType;
@@ -62,6 +65,8 @@
import java.util.Optional;
import java.util.Set;
import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
import static org.apache.pulsar.sql.presto.PulsarHandleResolver.convertColumnHandle;
import static org.apache.pulsar.sql.presto.PulsarHandleResolver.convertTableHandle;
@@ -116,8 +121,9 @@ public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTable
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session, ConnectorTableHandle table,
Constraint<ColumnHandle> constraint,
Optional<Set<ColumnHandle>> desiredColumns) {
+
PulsarTableHandle handle = convertTableHandle(table);
- ConnectorTableLayout layout = new ConnectorTableLayout(new PulsarTableLayoutHandle(handle));
+ ConnectorTableLayout layout = new ConnectorTableLayout(new PulsarTableLayoutHandle(handle, constraint.getSummary()));
return ImmutableList.of(new ConnectorTableLayoutResult(layout, constraint.getSummary()));
}
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java
index 3523228763..8593c55ccf 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java
@@ -20,6 +20,7 @@
import com.facebook.presto.spi.PrestoException;
import com.facebook.presto.spi.RecordCursor;
+import com.facebook.presto.spi.type.TimestampWithTimeZoneType;
import com.facebook.presto.spi.type.Type;
import com.facebook.presto.spi.type.VarbinaryType;
import com.facebook.presto.spi.type.VarcharType;
@@ -55,12 +56,15 @@
import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
import static com.facebook.presto.spi.type.BigintType.BIGINT;
+import static com.facebook.presto.spi.type.DateTimeEncoding.packDateTimeWithZone;
+import static com.facebook.presto.spi.type.DateTimeEncoding.unpackMillisUtc;
import static com.facebook.presto.spi.type.DateType.DATE;
import static com.facebook.presto.spi.type.IntegerType.INTEGER;
import static com.facebook.presto.spi.type.RealType.REAL;
import static com.facebook.presto.spi.type.SmallintType.SMALLINT;
import static com.facebook.presto.spi.type.TimeType.TIME;
import static com.facebook.presto.spi.type.TimestampType.TIMESTAMP;
+import static com.facebook.presto.spi.type.TimestampWithTimeZoneType.TIMESTAMP_WITH_TIME_ZONE;
import static com.facebook.presto.spi.type.TinyintType.TINYINT;
import static com.google.common.base.Preconditions.checkArgument;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -264,7 +268,7 @@ public long getLong(int field) {
if (type.equals(BIGINT)) {
return ((Number) record).longValue();
} else if (type.equals(DATE)) {
- return MILLISECONDS.toDays(new Date(TimeUnit.DAYS.toMillis(((Number) record).longValue())).getTime());
+ return ((Number) record).longValue();
} else if (type.equals(INTEGER)) {
return (int) record;
} else if (type.equals(REAL)) {
@@ -272,9 +276,11 @@ public long getLong(int field) {
} else if (type.equals(SMALLINT)) {
return ((Number) record).shortValue();
} else if (type.equals(TIME)) {
- return new Time(((Number) record).longValue()).getTime();
+ return ((Number) record).longValue();
} else if (type.equals(TIMESTAMP)) {
- return new Timestamp(((Number) record).longValue()).getTime();
+ return ((Number) record).longValue();
+ } else if (type.equals(TIMESTAMP_WITH_TIME_ZONE)) {
+ return packDateTimeWithZone(((Number) record).longValue(), 0);
} else if (type.equals(TINYINT)) {
return Byte.parseByte(record.toString());
} else {
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarSplit.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarSplit.java
index 580eb479c8..4bd1171ac0 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarSplit.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarSplit.java
@@ -18,8 +18,10 @@
*/
package org.apache.pulsar.sql.presto;
+import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.ConnectorSplit;
import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.predicate.TupleDomain;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableList;
@@ -43,7 +45,7 @@
private final long endPositionEntryId;
private final long startPositionLedgerId;
private final long endPositionLedgerId;
-
+ private final TupleDomain<ColumnHandle> tupleDomain;
@JsonCreator
public PulsarSplit(
@@ -57,7 +59,8 @@ public PulsarSplit(
@JsonProperty("startPositionEntryId") long startPositionEntryId,
@JsonProperty("endPositionEntryId") long endPositionEntryId,
@JsonProperty("startPositionLedgerId") long startPositionLedgerId,
- @JsonProperty("endPositionLedgerId") long endPositionLedgerId) {
+ @JsonProperty("endPositionLedgerId") long endPositionLedgerId,
+ @JsonProperty("tupleDomain") TupleDomain<ColumnHandle> tupleDomain) {
this.splitId = splitId;
this.schemaName = requireNonNull(schemaName, "schema name is null");
this.connectorId = requireNonNull(connectorId, "connector id is null");
@@ -69,6 +72,7 @@ public PulsarSplit(
this.endPositionEntryId = endPositionEntryId;
this.startPositionLedgerId = startPositionLedgerId;
this.endPositionLedgerId = endPositionLedgerId;
+ this.tupleDomain = requireNonNull(tupleDomain, "tupleDomain is null");
}
@JsonProperty
@@ -126,6 +130,11 @@ public long getEndPositionLedgerId() {
return endPositionLedgerId;
}
+ @JsonProperty
+ public TupleDomain<ColumnHandle> getTupleDomain() {
+ return tupleDomain;
+ }
+
public PositionImpl getStartPosition() {
return PositionImpl.get(startPositionLedgerId, startPositionEntryId);
}
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarSplitManager.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarSplitManager.java
index 721b8e6d88..488609d7a3 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarSplitManager.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarSplitManager.java
@@ -18,14 +18,21 @@
*/
package org.apache.pulsar.sql.presto;
+import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.ConnectorSession;
import com.facebook.presto.spi.ConnectorSplitSource;
import com.facebook.presto.spi.ConnectorTableLayoutHandle;
import com.facebook.presto.spi.FixedSplitSource;
import com.facebook.presto.spi.connector.ConnectorSplitManager;
import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
+import com.facebook.presto.spi.predicate.Domain;
+import com.facebook.presto.spi.predicate.Range;
+import com.facebook.presto.spi.predicate.TupleDomain;
+import com.facebook.presto.spi.type.SqlTimestampWithTimeZone;
import com.google.common.annotations.VisibleForTesting;
import io.airlift.log.Logger;
+import lombok.Data;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
@@ -35,9 +42,11 @@
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.shade.com.google.common.base.Predicate;
import org.apache.pulsar.shade.org.apache.bookkeeper.conf.ClientConfiguration;
import javax.inject.Inject;
@@ -46,7 +55,9 @@
import java.util.LinkedList;
import java.util.List;
+import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
+import static org.apache.bookkeeper.mledger.ManagedCursor.FindPositionConstraint.SearchAllAvailableEntries;
public class PulsarSplitManager implements ConnectorSplitManager {
@@ -79,6 +90,7 @@ public ConnectorSplitSource getSplits(ConnectorTransactionHandle transactionHand
PulsarTableLayoutHandle layoutHandle = (PulsarTableLayoutHandle) layout;
PulsarTableHandle tableHandle = layoutHandle.getTable();
+ TupleDomain<ColumnHandle> tupleDomain = layoutHandle.getTupleDomain();
TopicName topicName = TopicName.get("persistent", NamespaceName.get(tableHandle.getSchemaName()),
tableHandle.getTableName());
@@ -95,10 +107,10 @@ public ConnectorSplitSource getSplits(ConnectorTransactionHandle transactionHand
Collection<PulsarSplit> splits;
try {
if (!PulsarConnectorUtils.isPartitionedTopic(topicName, this.pulsarAdmin)) {
- splits = getSplitsNonPartitionedTopic(numSplits, topicName, tableHandle, schemaInfo);
+ splits = getSplitsNonPartitionedTopic(numSplits, topicName, tableHandle, schemaInfo, tupleDomain);
log.debug("Splits for non-partitioned topic %s: %s", topicName, splits);
} else {
- splits = getSplitsPartitionedTopic(numSplits, topicName, tableHandle, schemaInfo);
+ splits = getSplitsPartitionedTopic(numSplits, topicName, tableHandle, schemaInfo, tupleDomain);
log.debug("Splits for partitioned topic %s: %s", topicName, splits);
}
} catch (Exception e) {
@@ -119,7 +131,7 @@ ManagedLedgerFactory getManagedLedgerFactory() throws Exception {
@VisibleForTesting
Collection<PulsarSplit> getSplitsPartitionedTopic(int numSplits, TopicName topicName, PulsarTableHandle
- tableHandle, SchemaInfo schemaInfo) throws Exception {
+ tableHandle, SchemaInfo schemaInfo, TupleDomain<ColumnHandle> tupleDomain) throws Exception {
int numPartitions;
try {
numPartitions = (this.pulsarAdmin.topics().getPartitionedTopicMetadata(topicName.toString())).partitions;
@@ -148,7 +160,8 @@ ManagedLedgerFactory getManagedLedgerFactory() throws Exception {
splitsForThisPartition,
tableHandle,
schemaInfo,
- topicName.getPartition(i).getLocalName())
+ topicName.getPartition(i).getLocalName(),
+ tupleDomain)
);
}
return splits;
@@ -163,12 +176,38 @@ ManagedLedgerFactory getManagedLedgerFactory() throws Exception {
}
}
+ @VisibleForTesting
+ Collection<PulsarSplit> getSplitsNonPartitionedTopic(int numSplits, TopicName topicName, PulsarTableHandle
+ tableHandle, SchemaInfo schemaInfo, TupleDomain<ColumnHandle> tupleDomain) throws Exception {
+ ManagedLedgerFactory managedLedgerFactory = null;
+ try {
+ managedLedgerFactory = getManagedLedgerFactory();
+
+ return getSplitsForTopic(
+ topicName.getPersistenceNamingEncoding(),
+ managedLedgerFactory,
+ numSplits,
+ tableHandle,
+ schemaInfo,
+ tableHandle.getTableName(), tupleDomain);
+ } finally {
+ if (managedLedgerFactory != null) {
+ try {
+ managedLedgerFactory.shutdown();
+ } catch (Exception e) {
+ log.error(e);
+ }
+ }
+ }
+ }
+
@VisibleForTesting
Collection<PulsarSplit> getSplitsForTopic(String topicNamePersistenceEncoding,
ManagedLedgerFactory managedLedgerFactory,
int numSplits,
PulsarTableHandle tableHandle,
- SchemaInfo schemaInfo, String tableName)
+ SchemaInfo schemaInfo, String tableName,
+ TupleDomain<ColumnHandle> tupleDomain)
throws ManagedLedgerException, InterruptedException {
ReadOnlyCursor readOnlyCursor = null;
@@ -176,10 +215,33 @@ ManagedLedgerFactory getManagedLedgerFactory() throws Exception {
readOnlyCursor = managedLedgerFactory.openReadOnlyCursor(
topicNamePersistenceEncoding,
PositionImpl.earliest, new ManagedLedgerConfig());
+
long numEntries = readOnlyCursor.getNumberOfEntries();
if (numEntries <= 0) {
return Collections.EMPTY_LIST;
}
+
+ PredicatePushdownInfo predicatePushdownInfo = PredicatePushdownInfo.getPredicatePushdownInfo(
+ this.connectorId,
+ tupleDomain,
+ managedLedgerFactory,
+ topicNamePersistenceEncoding,
+ numEntries);
+
+ PositionImpl initialStartPosition;
+ if (predicatePushdownInfo != null) {
+ numEntries = predicatePushdownInfo.getNumOfEntries();
+ initialStartPosition = predicatePushdownInfo.getStartPosition();
+ } else {
+ initialStartPosition = (PositionImpl) readOnlyCursor.getReadPosition();
+ }
+
+
+ readOnlyCursor.close();
+ readOnlyCursor = managedLedgerFactory.openReadOnlyCursor(
+ topicNamePersistenceEncoding,
+ initialStartPosition, new ManagedLedgerConfig());
+
long remainder = numEntries % numSplits;
long avgEntriesPerSplit = numEntries / numSplits;
@@ -200,7 +262,8 @@ ManagedLedgerFactory getManagedLedgerFactory() throws Exception {
startPosition.getEntryId(),
endPosition.getEntryId(),
startPosition.getLedgerId(),
- endPosition.getLedgerId()));
+ endPosition.getLedgerId(),
+ tupleDomain));
}
return splits;
} finally {
@@ -214,28 +277,119 @@ ManagedLedgerFactory getManagedLedgerFactory() throws Exception {
}
}
- @VisibleForTesting
- Collection<PulsarSplit> getSplitsNonPartitionedTopic(int numSplits, TopicName topicName, PulsarTableHandle
- tableHandle, SchemaInfo schemaInfo) throws Exception {
- ManagedLedgerFactory managedLedgerFactory = null;
- try {
- managedLedgerFactory = getManagedLedgerFactory();
+ @Data
+ private static class PredicatePushdownInfo {
+ private PositionImpl startPosition;
+ private PositionImpl endPosition;
+ private long numOfEntries;
- return getSplitsForTopic(
- topicName.getPersistenceNamingEncoding(),
- managedLedgerFactory,
- numSplits,
- tableHandle,
- schemaInfo,
- tableHandle.getTableName());
- } finally {
- if (managedLedgerFactory != null) {
+ private PredicatePushdownInfo(PositionImpl startPosition, PositionImpl endPosition, long numOfEntries) {
+ this.startPosition = startPosition;
+ this.endPosition = endPosition;
+ this.numOfEntries = numOfEntries;
+ }
+
+ public static PredicatePushdownInfo getPredicatePushdownInfo(String connectorId,
+ TupleDomain<ColumnHandle> tupleDomain,
+ ManagedLedgerFactory managedLedgerFactory,
+ String topicNamePersistenceEncoding,
+ long totalNumEntries) throws
+ ManagedLedgerException, InterruptedException {
+
+ ReadOnlyCursor readOnlyCursor = null;
+ try {
+ readOnlyCursor = managedLedgerFactory.openReadOnlyCursor(
+ topicNamePersistenceEncoding,
+ PositionImpl.earliest, new ManagedLedgerConfig());
+
+ if (tupleDomain.getDomains().isPresent()) {
+ Domain domain = tupleDomain.getDomains().get().get(PulsarInternalColumn.PUBLISH_TIME
+ .getColumnHandle(connectorId, false));
+ if (domain != null) {
+ // TODO support arbitrary number of ranges
+ // only worry about one range for now
+ if (domain.getValues().getRanges().getRangeCount() == 1) {
+
+ checkArgument(domain.getType().isOrderable(), "Domain type must be orderable");
+
+ Long upperBoundTs = null;
+ Long lowerBoundTs = null;
+
+ Range range = domain.getValues().getRanges().getOrderedRanges().get(0);
+
+ if (!range.getHigh().isUpperUnbounded()) {
+ upperBoundTs = new SqlTimestampWithTimeZone(range.getHigh().getValueBlock().get()
+ .getLong(0, 0)).getMillisUtc();
+ }
+
+ if (!range.getLow().isLowerUnbounded()) {
+ lowerBoundTs = new SqlTimestampWithTimeZone(range.getLow().getValueBlock().get()
+ .getLong(0, 0)).getMillisUtc();
+ }
+
+ PositionImpl overallStartPos;
+ if (lowerBoundTs == null) {
+ overallStartPos = (PositionImpl) readOnlyCursor.getReadPosition();
+ } else {
+ overallStartPos = findPosition(readOnlyCursor, lowerBoundTs);
+ }
+
+ PositionImpl overallEndPos;
+ if (upperBoundTs == null) {
+
+ readOnlyCursor.skipEntries(Math.toIntExact(totalNumEntries));
+ overallEndPos = (PositionImpl) readOnlyCursor.getReadPosition();
+ } else {
+ overallEndPos = findPosition(readOnlyCursor, upperBoundTs);
+ }
+
+ // Just use a close bound since presto can always filter out the extra entries even if
+ // the bound
+ // should be open or a mixture of open and closed
+ org.apache.pulsar.shade.com.google.common.collect.Range<PositionImpl> posRange
+ = org.apache.pulsar.shade.com.google.common.collect.Range.range(overallStartPos,
+ org.apache.pulsar.shade.com.google.common.collect.BoundType.CLOSED,
+ overallEndPos, org.apache.pulsar.shade.com.google.common.collect.BoundType.CLOSED);
+
+ long numOfEntries = readOnlyCursor.getNumberOfEntries(posRange) - 1;
+
+ PredicatePushdownInfo predicatePushdownInfo
+ = new PredicatePushdownInfo(overallStartPos, overallEndPos, numOfEntries);
+ log.debug("Predicate pushdown optimization calculated: %s", predicatePushdownInfo);
+ return predicatePushdownInfo;
+ }
+ }
+ }
+ } finally {
+ if (readOnlyCursor != null) {
+ readOnlyCursor.close();
+ }
+ }
+ return null;
+ }
+ }
+
+ private static PositionImpl findPosition(ReadOnlyCursor readOnlyCursor, long timestamp) throws
+ ManagedLedgerException,
+ InterruptedException {
+ return (PositionImpl) readOnlyCursor.findNewestMatching(SearchAllAvailableEntries, new Predicate<Entry>() {
+ @Override
+ public boolean apply(Entry entry) {
+ MessageImpl msg = null;
try {
- managedLedgerFactory.shutdown();
+ msg = MessageImpl.deserialize(entry.getDataBuffer());
+
+ return msg.getPublishTime() <= timestamp;
} catch (Exception e) {
- log.error(e);
+ log.error(e, "Failed To deserialize message when finding position with error: %s", e);
+ } finally {
+ entry.release();
+ if (msg != null) {
+ msg.recycle();
+ }
}
+ return false;
}
- }
+ });
}
}
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarTableLayoutHandle.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarTableLayoutHandle.java
index 7bdbddad53..a7c240b940 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarTableLayoutHandle.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarTableLayoutHandle.java
@@ -18,18 +18,26 @@
*/
package org.apache.pulsar.sql.presto;
+import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.ConnectorTableLayoutHandle;
+import com.facebook.presto.spi.predicate.TupleDomain;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Objects;
+
import static java.util.Objects.requireNonNull;
public class PulsarTableLayoutHandle implements ConnectorTableLayoutHandle {
private final PulsarTableHandle table;
+ private final TupleDomain<ColumnHandle> tupleDomain;
@JsonCreator
- public PulsarTableLayoutHandle(@JsonProperty("table") PulsarTableHandle table) {
+ public PulsarTableLayoutHandle(@JsonProperty("table") PulsarTableHandle table,
+ @JsonProperty("tupleDomain") TupleDomain<ColumnHandle> domain) {
+
this.table = requireNonNull(table, "table is null");
+ this.tupleDomain = requireNonNull(domain, "tupleDomain is null");
}
@JsonProperty
@@ -37,8 +45,35 @@ public PulsarTableHandle getTable() {
return table;
}
+ @JsonProperty
+ public TupleDomain<ColumnHandle> getTupleDomain()
+ {
+ return tupleDomain;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PulsarTableLayoutHandle that = (PulsarTableLayoutHandle) o;
+ return Objects.equals(table, that.table) &&
+ Objects.equals(tupleDomain, that.tupleDomain);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(table, tupleDomain);
+ }
+
@Override
- public String toString() {
+ public String toString()
+ {
return table.toString();
}
}
diff --git a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java
index a3fa12d31b..4da7ea57f5 100644
--- a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java
+++ b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.sql.presto;
+import com.facebook.presto.spi.predicate.TupleDomain;
import com.facebook.presto.spi.type.BigintType;
import com.facebook.presto.spi.type.BooleanType;
import com.facebook.presto.spi.type.DoubleType;
@@ -28,6 +29,7 @@
import io.airlift.log.Logger;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
+import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.ReadOnlyCursor;
import org.apache.bookkeeper.mledger.impl.EntryImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
@@ -78,6 +80,7 @@
import static org.mockito.Matchers.anyInt;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
@@ -86,6 +89,8 @@
@Test(singleThreaded = true)
public abstract class TestPulsarConnector {
+ protected static final long currentTimeMs = 1534806330000L;
+
protected PulsarConnectorConfig pulsarConnectorConfig;
protected PulsarMetadata pulsarMetadata;
@@ -104,7 +109,7 @@
protected static List<TopicName> partitionedTopicNames;
protected static Map<String, Integer> partitionedTopicsToPartitions;
protected static Map<String, SchemaInfo> topicsToSchemas;
- protected static Map<String, Long> topicsToEntries;
+ protected static Map<String, Long> topicsToNumEntries;
protected static final NamespaceName NAMESPACE_NAME_1 = NamespaceName.get("tenant-1", "ns-1");
protected static final NamespaceName NAMESPACE_NAME_2 = NamespaceName.get("tenant-1", "ns-2");
@@ -274,19 +279,19 @@ public void setDate(int date) {
fooTypes.put("time", TIME);
fooTypes.put("date", DATE);
- topicsToEntries = new HashMap<>();
- topicsToEntries.put(TOPIC_1.getSchemaName(), 1233L);
- topicsToEntries.put(TOPIC_2.getSchemaName(), 0L);
- topicsToEntries.put(TOPIC_3.getSchemaName(), 100L);
- topicsToEntries.put(TOPIC_4.getSchemaName(), 12345L);
- topicsToEntries.put(TOPIC_5.getSchemaName(), 8000L);
- topicsToEntries.put(TOPIC_6.getSchemaName(), 1L);
- topicsToEntries.put(PARTITIONED_TOPIC_1.getSchemaName(), 1233L);
- topicsToEntries.put(PARTITIONED_TOPIC_2.getSchemaName(), 8000L);
- topicsToEntries.put(PARTITIONED_TOPIC_3.getSchemaName(), 100L);
- topicsToEntries.put(PARTITIONED_TOPIC_4.getSchemaName(), 0L);
- topicsToEntries.put(PARTITIONED_TOPIC_5.getSchemaName(), 800L);
- topicsToEntries.put(PARTITIONED_TOPIC_6.getSchemaName(), 1L);
+ topicsToNumEntries = new HashMap<>();
+ topicsToNumEntries.put(TOPIC_1.getSchemaName(), 1233L);
+ topicsToNumEntries.put(TOPIC_2.getSchemaName(), 0L);
+ topicsToNumEntries.put(TOPIC_3.getSchemaName(), 100L);
+ topicsToNumEntries.put(TOPIC_4.getSchemaName(), 12345L);
+ topicsToNumEntries.put(TOPIC_5.getSchemaName(), 8000L);
+ topicsToNumEntries.put(TOPIC_6.getSchemaName(), 1L);
+ topicsToNumEntries.put(PARTITIONED_TOPIC_1.getSchemaName(), 1233L);
+ topicsToNumEntries.put(PARTITIONED_TOPIC_2.getSchemaName(), 8000L);
+ topicsToNumEntries.put(PARTITIONED_TOPIC_3.getSchemaName(), 100L);
+ topicsToNumEntries.put(PARTITIONED_TOPIC_4.getSchemaName(), 0L);
+ topicsToNumEntries.put(PARTITIONED_TOPIC_5.getSchemaName(), 800L);
+ topicsToNumEntries.put(PARTITIONED_TOPIC_6.getSchemaName(), 1L);
fooColumnHandles = new LinkedList<>();
for (int i = 0; i < Foo.class.getDeclaredFields().length; i++) {
@@ -315,14 +320,54 @@ public PulsarColumnHandle apply(PulsarInternalColumn pulsarInternalColumn) {
for (TopicName topicName : allTopics) {
splits.put(topicName, new PulsarSplit(0, pulsarConnectorId.toString(),
topicName.getNamespace(), topicName.getLocalName(),
- topicsToEntries.get(topicName.getSchemaName()),
+ topicsToNumEntries.get(topicName.getSchemaName()),
new String(topicsToSchemas.get(topicName.getSchemaName()).getSchema()),
topicsToSchemas.get(topicName.getSchemaName()).getType(),
- 0, topicsToEntries.get(topicName.getSchemaName()),
- 0, 0));
+ 0, topicsToNumEntries.get(topicName.getSchemaName()),
+ 0, 0, TupleDomain.all()));
}
}
+ private static List<Entry> getTopicEntries(String topicSchemaName) {
+ List<Entry> entries = new LinkedList<>();
+
+ long count = topicsToNumEntries.get(topicSchemaName);
+ for (int i=0 ; i < count; i++) {
+
+ Foo foo = new Foo();
+ foo.field1 = (int) count;
+ foo.field2 = String.valueOf(count);
+ foo.field3 = count;
+ foo.field4 = count;
+ foo.field5 = count % 2 == 0;
+ foo.field6 = count;
+ foo.timestamp = System.currentTimeMillis();
+
+ LocalTime now = LocalTime.now(ZoneId.systemDefault());
+ foo.time = now.toSecondOfDay() * 1000;
+
+ LocalDate localDate = LocalDate.now();
+ LocalDate epoch = LocalDate.ofEpochDay(0);
+ foo.date = Math.toIntExact(ChronoUnit.DAYS.between(epoch, localDate));
+
+ PulsarApi.MessageMetadata messageMetadata = PulsarApi.MessageMetadata.newBuilder()
+ .setProducerName("test-producer").setSequenceId(i)
+ .setPublishTime(currentTimeMs + i).build();
+
+ Schema schema = topicsToSchemas.get(topicSchemaName).getType() == SchemaType.AVRO ? AvroSchema.of(Foo.class) : JSONSchema.of(Foo.class);
+
+ org.apache.pulsar.shade.io.netty.buffer.ByteBuf payload
+ = org.apache.pulsar.shade.io.netty.buffer.Unpooled.copiedBuffer(schema.encode(foo));
+
+ ByteBuf byteBuf = serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, messageMetadata, payload);
+
+ Entry entry = EntryImpl.create(0, i, byteBuf);
+ log.info("create entry: %s", entry.getEntryId());
+ entries.add(entry);
+ }
+ return entries;
+ }
+
public long completedBytes = 0L;
private static final Logger log = Logger.get(TestPulsarConnector.class);
@@ -464,12 +509,18 @@ public SchemaInfo answer(InvocationOnMock invocationOnMock) throws Throwable {
public ReadOnlyCursor answer(InvocationOnMock invocationOnMock) throws Throwable {
Object[] args = invocationOnMock.getArguments();
String topic = (String) args[0];
- positions.put(topic, 0);
+ PositionImpl positionImpl = (PositionImpl) args[1];
+
+ int position = positionImpl.getEntryId() == -1 ? 0 : (int) positionImpl.getEntryId();
+
+ positions.put(topic, position);
String schemaName = TopicName.get(
TopicName.get(
topic.replaceAll("/persistent", ""))
.getPartitionedTopicName()).getSchemaName();
- long entries = topicsToEntries.get(schemaName);
+ long entries = topicsToNumEntries.get(schemaName);
+
+
ReadOnlyCursor readOnlyCursor = mock(ReadOnlyCursor.class);
doReturn(entries).when(readOnlyCursor).getNumberOfEntries();
@@ -545,6 +596,43 @@ public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable {
}
});
+ when(readOnlyCursor.findNewestMatching(any(), any())).then(new Answer<Position>() {
+ @Override
+ public Position answer(InvocationOnMock invocationOnMock) throws Throwable {
+ Object[] args = invocationOnMock.getArguments();
+ org.apache.pulsar.shade.com.google.common.base.Predicate<Entry> predicate
+ = (org.apache.pulsar.shade.com.google.common.base.Predicate<Entry>) args[1];
+
+ String schemaName = TopicName.get(
+ TopicName.get(
+ topic.replaceAll("/persistent", ""))
+ .getPartitionedTopicName()).getSchemaName();
+ List<Entry> entries = getTopicEntries(schemaName);
+
+ Integer target = null;
+ for (int i=entries.size() - 1; i >= 0; i--) {
+ Entry entry = entries.get(i);
+ if (predicate.apply(entry)) {
+ target = i;
+ break;
+ }
+ }
+
+ return target == null ? null : new PositionImpl(0, target);
+ }
+ });
+
+ when(readOnlyCursor.getNumberOfEntries(any())).then(new Answer<Long>() {
+ @Override
+ public Long answer(InvocationOnMock invocationOnMock) throws Throwable {
+ Object[] args = invocationOnMock.getArguments();
+ org.apache.pulsar.shade.com.google.common.collect.Range<PositionImpl> range
+ = (org.apache.pulsar.shade.com.google.common.collect.Range<PositionImpl> ) args[0];
+
+ return (range.upperEndpoint().getEntryId() + 1) - range.lowerEndpoint().getEntryId();
+ }
+ });
+
return readOnlyCursor;
}
});
diff --git a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarRecordCursor.java b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarRecordCursor.java
index 8d983ed95a..74126a61a5 100644
--- a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarRecordCursor.java
+++ b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarRecordCursor.java
@@ -71,7 +71,7 @@ public void testTopics() throws Exception {
}
count++;
}
- Assert.assertEquals(count, topicsToEntries.get(topicName.getSchemaName()).longValue());
+ Assert.assertEquals(count, topicsToNumEntries.get(topicName.getSchemaName()).longValue());
Assert.assertEquals(pulsarRecordCursor.getCompletedBytes(), completedBytes);
cleanup();
}
diff --git a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarSplitManager.java b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarSplitManager.java
index decdc13181..7c9b4bb1c0 100644
--- a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarSplitManager.java
+++ b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarSplitManager.java
@@ -18,9 +18,15 @@
*/
package org.apache.pulsar.sql.presto;
+import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.ConnectorSession;
import com.facebook.presto.spi.ConnectorSplitSource;
import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
+import com.facebook.presto.spi.predicate.Domain;
+import com.facebook.presto.spi.predicate.Range;
+import com.facebook.presto.spi.predicate.TupleDomain;
+import com.facebook.presto.spi.predicate.ValueSet;
+import com.facebook.presto.spi.type.TimeZoneKey;
import io.airlift.log.Logger;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.common.naming.TopicName;
@@ -30,10 +36,14 @@
import org.testng.annotations.Test;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.function.Predicate;
import java.util.stream.Collectors;
+import static com.facebook.presto.spi.type.DateTimeEncoding.packDateTimeWithZone;
+import static com.facebook.presto.spi.type.TimestampWithTimeZoneType.TIMESTAMP_WITH_TIME_ZONE;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyInt;
import static org.mockito.Mockito.doAnswer;
@@ -69,10 +79,10 @@ public void testTopic() throws Exception {
topicName.getNamespace(),
topicName.getLocalName(),
topicName.getLocalName());
- PulsarTableLayoutHandle pulsarTableLayoutHandle = new PulsarTableLayoutHandle(pulsarTableHandle);
+ PulsarTableLayoutHandle pulsarTableLayoutHandle = new PulsarTableLayoutHandle(pulsarTableHandle, TupleDomain.all());
final ResultCaptor<Collection<PulsarSplit>> resultCaptor = new ResultCaptor<>();
- doAnswer(resultCaptor).when(this.pulsarSplitManager).getSplitsNonPartitionedTopic(anyInt(), any(), any(), any());
+ doAnswer(resultCaptor).when(this.pulsarSplitManager).getSplitsNonPartitionedTopic(anyInt(), any(), any(), any(), any());
ConnectorSplitSource connectorSplitSource = this.pulsarSplitManager.getSplits(
@@ -80,7 +90,7 @@ public void testTopic() throws Exception {
pulsarTableLayoutHandle, null);
verify(this.pulsarSplitManager, times(1))
- .getSplitsNonPartitionedTopic(anyInt(), any(), any(), any());
+ .getSplitsNonPartitionedTopic(anyInt(), any(), any(), any(), any());
int totalSize = 0;
for (PulsarSplit pulsarSplit : resultCaptor.getResult()) {
@@ -100,7 +110,7 @@ public void testTopic() throws Exception {
totalSize += pulsarSplit.getSplitSize();
}
- Assert.assertEquals(totalSize, topicsToEntries.get(topicName.getSchemaName()).intValue());
+ Assert.assertEquals(totalSize, topicsToNumEntries.get(topicName.getSchemaName()).intValue());
cleanup();
}
@@ -115,16 +125,16 @@ public void testPartitionedTopic() throws Exception {
topicName.getNamespace(),
topicName.getLocalName(),
topicName.getLocalName());
- PulsarTableLayoutHandle pulsarTableLayoutHandle = new PulsarTableLayoutHandle(pulsarTableHandle);
+ PulsarTableLayoutHandle pulsarTableLayoutHandle = new PulsarTableLayoutHandle(pulsarTableHandle, TupleDomain.all());
final ResultCaptor<Collection<PulsarSplit>> resultCaptor = new ResultCaptor<>();
- doAnswer(resultCaptor).when(this.pulsarSplitManager).getSplitsPartitionedTopic(anyInt(), any(), any(), any());
+ doAnswer(resultCaptor).when(this.pulsarSplitManager).getSplitsPartitionedTopic(anyInt(), any(), any(), any(), any());
this.pulsarSplitManager.getSplits(mock(ConnectorTransactionHandle.class), mock(ConnectorSession.class),
pulsarTableLayoutHandle, null);
verify(this.pulsarSplitManager, times(1))
- .getSplitsPartitionedTopic(anyInt(), any(), any(), any());
+ .getSplitsPartitionedTopic(anyInt(), any(), any(), any(), any());
int partitions = partitionedTopicsToPartitions.get(topicName.toString());
@@ -148,7 +158,7 @@ public void testPartitionedTopic() throws Exception {
totalSize += pulsarSplit.getSplitSize();
}
- Assert.assertEquals(totalSize, topicsToEntries.get(topicName.getSchemaName()).intValue());
+ Assert.assertEquals(totalSize, topicsToNumEntries.get(topicName.getSchemaName()).intValue());
}
cleanup();
@@ -165,5 +175,126 @@ public boolean test(PulsarSplit pulsarSplit) {
}
}).collect(Collectors.toList());
}
+
+ @Test
+ public void testPublishTimePredicatePushdown() throws Exception {
+
+ TopicName topicName = TOPIC_1;
+
+ setup();
+ log.info("!----- topic: %s -----!", topicName);
+ PulsarTableHandle pulsarTableHandle = new PulsarTableHandle(pulsarConnectorId.toString(),
+ topicName.getNamespace(),
+ topicName.getLocalName(),
+ topicName.getLocalName());
+
+
+ Map<ColumnHandle, Domain> domainMap = new HashMap<>();
+ Domain domain = Domain.create(ValueSet.ofRanges(Range.range(TIMESTAMP_WITH_TIME_ZONE, packDateTimeWithZone
+ (currentTimeMs + 1L, TimeZoneKey.UTC_KEY), true, packDateTimeWithZone(currentTimeMs + 50L,
+ TimeZoneKey.UTC_KEY), true)), false);
+ domainMap.put(PulsarInternalColumn.PUBLISH_TIME.getColumnHandle(pulsarConnectorId.toString(), false), domain);
+ TupleDomain<ColumnHandle> tupleDomain = TupleDomain.withColumnDomains(domainMap);
+
+ PulsarTableLayoutHandle pulsarTableLayoutHandle = new PulsarTableLayoutHandle(pulsarTableHandle, tupleDomain);
+
+ final ResultCaptor<Collection<PulsarSplit>> resultCaptor = new ResultCaptor<>();
+ doAnswer(resultCaptor).when(this.pulsarSplitManager).getSplitsNonPartitionedTopic(anyInt(), any(), any(), any
+ (), any());
+
+ ConnectorSplitSource connectorSplitSource = this.pulsarSplitManager.getSplits(
+ mock(ConnectorTransactionHandle.class), mock(ConnectorSession.class),
+ pulsarTableLayoutHandle, null);
+
+
+ verify(this.pulsarSplitManager, times(1))
+ .getSplitsNonPartitionedTopic(anyInt(), any(), any(), any(), any());
+
+ int totalSize = 0;
+ int initalStart = 1;
+ for (PulsarSplit pulsarSplit : resultCaptor.getResult()) {
+ Assert.assertEquals(pulsarSplit.getConnectorId(), pulsarConnectorId.toString());
+ Assert.assertEquals(pulsarSplit.getSchemaName(), topicName.getNamespace());
+ Assert.assertEquals(pulsarSplit.getTableName(), topicName.getLocalName());
+ Assert.assertEquals(pulsarSplit.getSchema(),
+ new String(topicsToSchemas.get(topicName.getSchemaName()).getSchema()));
+ Assert.assertEquals(pulsarSplit.getSchemaType(), topicsToSchemas.get(topicName.getSchemaName()).getType());
+ Assert.assertEquals(pulsarSplit.getStartPositionEntryId(), initalStart);
+ Assert.assertEquals(pulsarSplit.getStartPositionLedgerId(), 0);
+ Assert.assertEquals(pulsarSplit.getStartPosition(), PositionImpl.get(0, initalStart));
+ Assert.assertEquals(pulsarSplit.getEndPositionLedgerId(), 0);
+ Assert.assertEquals(pulsarSplit.getEndPositionEntryId(), initalStart + pulsarSplit.getSplitSize());
+ Assert.assertEquals(pulsarSplit.getEndPosition(), PositionImpl.get(0, initalStart + pulsarSplit
+ .getSplitSize()));
+
+ initalStart += pulsarSplit.getSplitSize();
+ totalSize += pulsarSplit.getSplitSize();
+ }
+ Assert.assertEquals(totalSize, 49);
+
+ }
+
+ @Test
+ public void testPublishTimePredicatePushdownPartitionedTopic() throws Exception {
+
+ TopicName topicName = PARTITIONED_TOPIC_1;
+
+ setup();
+ log.info("!----- topic: %s -----!", topicName);
+ PulsarTableHandle pulsarTableHandle = new PulsarTableHandle(pulsarConnectorId.toString(),
+ topicName.getNamespace(),
+ topicName.getLocalName(),
+ topicName.getLocalName());
+
+
+ Map<ColumnHandle, Domain> domainMap = new HashMap<>();
+ Domain domain = Domain.create(ValueSet.ofRanges(Range.range(TIMESTAMP_WITH_TIME_ZONE, packDateTimeWithZone
+ (currentTimeMs + 1L, TimeZoneKey.UTC_KEY), true, packDateTimeWithZone(currentTimeMs + 50L,
+ TimeZoneKey.UTC_KEY), true)), false);
+ domainMap.put(PulsarInternalColumn.PUBLISH_TIME.getColumnHandle(pulsarConnectorId.toString(), false), domain);
+ TupleDomain<ColumnHandle> tupleDomain = TupleDomain.withColumnDomains(domainMap);
+
+ PulsarTableLayoutHandle pulsarTableLayoutHandle = new PulsarTableLayoutHandle(pulsarTableHandle, tupleDomain);
+
+ final ResultCaptor<Collection<PulsarSplit>> resultCaptor = new ResultCaptor<>();
+ doAnswer(resultCaptor).when(this.pulsarSplitManager)
+ .getSplitsPartitionedTopic(anyInt(), any(), any(), any(), any());
+
+ ConnectorSplitSource connectorSplitSource = this.pulsarSplitManager.getSplits(
+ mock(ConnectorTransactionHandle.class), mock(ConnectorSession.class),
+ pulsarTableLayoutHandle, null);
+
+
+ verify(this.pulsarSplitManager, times(1))
+ .getSplitsPartitionedTopic(anyInt(), any(), any(), any(), any());
+
+
+ int partitions = partitionedTopicsToPartitions.get(topicName.toString());
+ for (int i = 0; i < partitions; i++) {
+ List<PulsarSplit> splits = getSplitsForPartition(topicName.getPartition(i), resultCaptor.getResult());
+ int totalSize = 0;
+ int initalStart = 1;
+ for (PulsarSplit pulsarSplit : splits) {
+ Assert.assertEquals(pulsarSplit.getConnectorId(), pulsarConnectorId.toString());
+ Assert.assertEquals(pulsarSplit.getSchemaName(), topicName.getNamespace());
+ Assert.assertEquals(pulsarSplit.getTableName(), topicName.getPartition(i).getLocalName());
+ Assert.assertEquals(pulsarSplit.getSchema(),
+ new String(topicsToSchemas.get(topicName.getSchemaName()).getSchema()));
+ Assert.assertEquals(pulsarSplit.getSchemaType(), topicsToSchemas.get(topicName.getSchemaName()).getType());
+ Assert.assertEquals(pulsarSplit.getStartPositionEntryId(), initalStart);
+ Assert.assertEquals(pulsarSplit.getStartPositionLedgerId(), 0);
+ Assert.assertEquals(pulsarSplit.getStartPosition(), PositionImpl.get(0, initalStart));
+ Assert.assertEquals(pulsarSplit.getEndPositionLedgerId(), 0);
+ Assert.assertEquals(pulsarSplit.getEndPositionEntryId(), initalStart + pulsarSplit.getSplitSize());
+ Assert.assertEquals(pulsarSplit.getEndPosition(), PositionImpl.get(0, initalStart + pulsarSplit.getSplitSize()));
+
+ initalStart += pulsarSplit.getSplitSize();
+ totalSize += pulsarSplit.getSplitSize();
+ }
+
+ Assert.assertEquals(totalSize, 49);
+ }
+ }
+
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services