You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2019/11/12 20:12:59 UTC
[incubator-pinot] branch upsert updated (c10ab01 -> 5f33a29)
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a change to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git.
from c10ab01 fix broker bug where upsert table is not queriable
new f35f1d7 [maven-release-plugin] prepare release pinot-0.2.2.5.43
new df8e3a8 [maven-release-plugin] prepare for next development iteration
new 5d46d78 Clean up imports for grigio
new 70a913c Implement FixedPartitionCountPartitioner
new 9ac12f3 fix issue in retention management and refactor
new 6df85f3 change offset mapping strategy for immutable seg to reduce memory
new 11adb6e change validFrom/validUntil column data type to int
new fea13a0 Add retries to RocksDB reads and writes
new 4708139 Fix FixedPartitionCountBytesPartitionerTest
new 7c8cc38 improve pinot server upsert segment loading time
new 51684f7 small improve on loading data for virtual column
new 5f33a29 add one more lag metrics to kc
The 12 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
pinot-api/pom.xml | 2 +-
pinot-azure-filesystem/pom.xml | 2 +-
pinot-broker/pom.xml | 2 +-
pinot-common/pom.xml | 2 +-
.../pinot-connector-kafka-0.11/pom.xml | 2 +-
pinot-connectors/pinot-connector-kafka-0.9/pom.xml | 2 +-
pinot-connectors/pom.xml | 2 +-
pinot-controller/pom.xml | 2 +-
pinot-core/pom.xml | 3 +-
.../pinot/core/indexsegment/UpsertSegment.java | 3 +-
.../immutable/ImmutableUpsertSegmentImpl.java | 187 ++++++++++++++-------
.../mutable/MutableUpsertSegmentImpl.java | 9 +-
.../segment/updater/UpsertWaterMarkManager.java | 13 +-
.../ValidFromInMemoryVirtualColumnProvider.java | 10 +-
.../ValidUntilInMemoryVirtualColumnProvider.java | 10 +-
.../VirtualColumnLongValueReaderWriter.java | 38 +++--
.../immutable/ImmutableUpsertSegmentImplTest.java | 126 ++++++++++++++
pinot-distribution/pom.xml | 2 +-
pinot-grigio/pinot-grigio-common/pom.xml | 8 +-
.../common/keyValueStore/KeyValueStoreTable.java | 2 -
.../pinot/grigio/common/messages/LogEventType.java | 8 +-
.../pinot/grigio/common/metrics/GrigioGauge.java | 1 +
.../pinot/grigio/common/metrics/GrigioMeter.java | 1 -
.../common/rpcQueue/QueueConsumerRecord.java | 8 +-
.../SegmentUpdateLogStorageProvider.java | 26 +--
.../common/storageProvider/UpdateLogEntry.java | 3 +
.../common/storageProvider/UpdateLogEntrySet.java | 69 ++++++++
.../storageProvider/UpdateLogStorageExplorer.java | 7 +-
.../storageProvider/UpdateLogStorageProvider.java | 8 +-
.../UpdateLogRetentionManagerImpl.java | 2 +-
.../UpdateLogTableRetentionManager.java | 99 ++++++++---
.../UpdateLogTableRetentionManagerImpl.java | 79 +--------
.../grigio}/common/utils/IdealStateHelper.java | 2 +-
.../common/rpcQueue/QueueConsumerRecordTest.java | 3 +-
.../SegmentUpdateLogStorageProviderTest.java | 73 ++++++--
.../UpdateLogTableRetentionManagerImplTest.java | 119 +++++++++++++
.../grigio/common/utils/IdealStateHelperTest.java | 74 ++++++++
pinot-grigio/pinot-grigio-coordinator/pom.xml | 38 +++--
...va => FixedPartitionCountBytesPartitioner.java} | 36 ++--
...java => FixedPartitionCountIntPartitioner.java} | 30 ++--
...er.java => FixedPartitionCountPartitioner.java} | 25 +--
.../common/keyValueStore/RocksDBBatchReader.java | 62 +++++++
.../common/keyValueStore/RocksDBBatchWriter.java | 42 +++--
.../keyValueStore/RocksDBKeyValueStoreTable.java | 31 +---
.../grigio/common/rpcQueue/KafkaQueueConsumer.java | 3 +-
.../rpcQueue/KeyCoordinatorQueueProducer.java | 4 +-
.../rpcQueue/LogCoordinatorQueueProducer.java | 4 +-
.../common/rpcQueue/VersionMsgQueueProducer.java | 4 +-
.../KCUpdateLogRetentionManagerImpl.java | 2 +-
.../KCUpdateLogTableRetentionManagerImpl.java | 58 ++-----
.../internal/DistributedKeyCoordinatorCore.java | 2 +
.../keyCoordinator/starter/KeyCoordinatorConf.java | 2 +-
.../starter/KeyCoordinatorStarter.java | 2 +-
.../FixedPartitionCountBytesPartitionerTest.java | 75 +++++++++
.../FixedPartitionCountIntPartitionerTest.java | 74 ++++++++
.../KCUpdateLogRetentionManagerImplTest.java | 114 +++++++++++++
.../KCUpdateLogTableRetentionManagerImplTest.java | 148 ++++++++++++++++
.../org.mockito.plugins.MockMaker | 1 +
pinot-grigio/pom.xml | 2 +-
pinot-hadoop-filesystem/pom.xml | 2 +-
pinot-hadoop/pom.xml | 2 +-
pinot-integration-tests/pom.xml | 2 +-
pinot-minion/pom.xml | 2 +-
pinot-orc/pom.xml | 2 +-
pinot-parquet/pom.xml | 2 +-
pinot-perf/pom.xml | 2 +-
pinot-server/pom.xml | 3 +-
.../pinot/server/starter/ServerInstance.java | 2 +-
pinot-tools/pom.xml | 2 +-
pinot-transport/pom.xml | 2 +-
pom.xml | 7 +-
71 files changed, 1362 insertions(+), 436 deletions(-)
create mode 100644 pinot-core/src/test/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImplTest.java
create mode 100644 pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogEntrySet.java
rename {pinot-common/src/main/java/org/apache/pinot => pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio}/common/utils/IdealStateHelper.java (98%)
create mode 100644 pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManagerImplTest.java
create mode 100644 pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/utils/IdealStateHelperTest.java
copy pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/{IntPartitioner.java => FixedPartitionCountBytesPartitioner.java} (56%)
copy pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/{IntPartitioner.java => FixedPartitionCountIntPartitioner.java} (64%)
rename pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/{IntPartitioner.java => FixedPartitionCountPartitioner.java} (65%)
create mode 100644 pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBBatchReader.java
copy pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/ResourceUtils.java => pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBBatchWriter.java (51%)
create mode 100644 pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountBytesPartitionerTest.java
create mode 100644 pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountIntPartitionerTest.java
create mode 100644 pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogRetentionManagerImplTest.java
create mode 100644 pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogTableRetentionManagerImplTest.java
create mode 100644 pinot-grigio/pinot-grigio-coordinator/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 03/12: Clean up imports for grigio
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 5d46d78fbb838c702114e96fd1d65a3c3ac25107
Author: Bo Zhang <bz...@uber.com>
AuthorDate: Fri Sep 27 16:44:01 2019 -0700
Clean up imports for grigio
Reviewers: #streaming_pinot, sjames
Reviewed By: #streaming_pinot, sjames
Differential Revision: https://code.uberinternal.com/D3519749
---
pinot-core/pom.xml | 1 -
pinot-grigio/pinot-grigio-common/pom.xml | 1 +
pinot-grigio/pinot-grigio-coordinator/pom.xml | 23 +++++++++++------------
pinot-server/pom.xml | 1 -
pom.xml | 5 +++++
5 files changed, 17 insertions(+), 14 deletions(-)
diff --git a/pinot-core/pom.xml b/pinot-core/pom.xml
index b949eaf..577c235 100644
--- a/pinot-core/pom.xml
+++ b/pinot-core/pom.xml
@@ -66,7 +66,6 @@
<dependency>
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-grigio-common</artifactId>
- <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.testng</groupId>
diff --git a/pinot-grigio/pinot-grigio-common/pom.xml b/pinot-grigio/pinot-grigio-common/pom.xml
index 82f3570..251cb5a 100644
--- a/pinot-grigio/pinot-grigio-common/pom.xml
+++ b/pinot-grigio/pinot-grigio-common/pom.xml
@@ -28,6 +28,7 @@
<modelVersion>4.0.0</modelVersion>
<artifactId>pinot-grigio-common</artifactId>
+ <name>Pinot Grigio Common</name>
<properties>
<pinot.root>${basedir}/../..</pinot.root>
</properties>
diff --git a/pinot-grigio/pinot-grigio-coordinator/pom.xml b/pinot-grigio/pinot-grigio-coordinator/pom.xml
index 3857d92..6222a8f 100644
--- a/pinot-grigio/pinot-grigio-coordinator/pom.xml
+++ b/pinot-grigio/pinot-grigio-coordinator/pom.xml
@@ -20,18 +20,18 @@
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <parent>
- <artifactId>pinot-grigio</artifactId>
- <groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
- </parent>
- <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <artifactId>pinot-grigio</artifactId>
+ <groupId>org.apache.pinot</groupId>
+ <version>0.2.2.5.44-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
- <artifactId>pinot-grigio-coordinator</artifactId>
-
- <properties>
- <pinot.root>${basedir}/../..</pinot.root>
- </properties>
+ <artifactId>pinot-grigio-coordinator</artifactId>
+ <name>Pinot Grigio Coordinator</name>
+ <properties>
+ <pinot.root>${basedir}/../..</pinot.root>
+ </properties>
<dependencies>
<dependency>
@@ -82,7 +82,6 @@
<dependency>
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-grigio-common</artifactId>
- <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
diff --git a/pinot-server/pom.xml b/pinot-server/pom.xml
index c5d394b..c1536cb 100644
--- a/pinot-server/pom.xml
+++ b/pinot-server/pom.xml
@@ -68,7 +68,6 @@
<dependency>
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-grigio-common</artifactId>
- <version>0.2.2.5.44-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
diff --git a/pom.xml b/pom.xml
index 5e4d1d0..d43b23c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -297,6 +297,11 @@
<version>${project.version}</version>
</dependency>
<dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-grigio-coordinator</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
<groupId>nl.jqno.equalsverifier</groupId>
<artifactId>equalsverifier</artifactId>
<version>1.7.2</version>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 04/12: Implement FixedPartitionCountPartitioner
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 70a913cf0821677dbd697dfe8bece675aaae5a84
Author: Bo Zhang <bz...@uber.com>
AuthorDate: Mon Oct 28 22:44:29 2019 -0700
Implement FixedPartitionCountPartitioner
Reviewers: #streaming_pinot, sjames
Reviewed By: #streaming_pinot, sjames
Subscribers: sjames
Maniphest Tasks: T4265291
Differential Revision: https://code.uberinternal.com/D3516373
---
pinot-grigio/pinot-grigio-coordinator/pom.xml | 5 ++
...va => FixedPartitionCountBytesPartitioner.java} | 36 +++++------
...java => FixedPartitionCountIntPartitioner.java} | 30 ++++-----
...er.java => FixedPartitionCountPartitioner.java} | 25 ++++----
.../rpcQueue/KeyCoordinatorQueueProducer.java | 4 +-
.../rpcQueue/LogCoordinatorQueueProducer.java | 4 +-
.../common/rpcQueue/VersionMsgQueueProducer.java | 4 +-
.../keyCoordinator/starter/KeyCoordinatorConf.java | 2 +-
.../FixedPartitionCountBytesPartitionerTest.java | 72 +++++++++++++++++++++
.../FixedPartitionCountIntPartitionerTest.java | 74 ++++++++++++++++++++++
.../org.mockito.plugins.MockMaker | 1 +
11 files changed, 203 insertions(+), 54 deletions(-)
diff --git a/pinot-grigio/pinot-grigio-coordinator/pom.xml b/pinot-grigio/pinot-grigio-coordinator/pom.xml
index 6222a8f..f8f6e48 100644
--- a/pinot-grigio/pinot-grigio-coordinator/pom.xml
+++ b/pinot-grigio/pinot-grigio-coordinator/pom.xml
@@ -93,6 +93,11 @@
<artifactId>testng</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/IntPartitioner.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/FixedPartitionCountBytesPartitioner.java
similarity index 56%
copy from pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/IntPartitioner.java
copy to pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/FixedPartitionCountBytesPartitioner.java
index cd83ee9..9b1de08 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/IntPartitioner.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/FixedPartitionCountBytesPartitioner.java
@@ -18,29 +18,27 @@
*/
package org.apache.pinot.grigio.common;
-import com.google.common.base.Preconditions;
-import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.utils.Utils;
-import java.util.List;
-import java.util.Map;
-public class IntPartitioner implements Partitioner {
- @Override
- public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
- Preconditions.checkState(key instanceof Integer, "expect key to be an integer");
- List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
- int numPartitions = partitions.size();
- return (Integer) key % numPartitions;
- }
-
- @Override
- public void close() {
-
- }
+/**
+ * Fixed partition count partitioner that partition with the bytes of the primary key
+ */
+public class FixedPartitionCountBytesPartitioner extends FixedPartitionCountPartitioner {
@Override
- public void configure(Map<String, ?> configs) {
+ public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
+ if (keyBytes == null) {
+ throw new IllegalArgumentException("Cannot partition without a key");
+ }
+ int numPartitions = cluster.partitionCountForTopic(topic);
+ int partitionCount = getPartitionCount();
+ if (partitionCount > numPartitions) {
+ throw new IllegalArgumentException(String
+ .format("Cannot partition to %d partitions for records in topic %s, which has only %d partitions.",
+ partitionCount, topic, numPartitions));
+ }
+ return Utils.toPositive(Utils.murmur2(keyBytes)) % partitionCount;
}
}
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/IntPartitioner.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/FixedPartitionCountIntPartitioner.java
similarity index 64%
copy from pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/IntPartitioner.java
copy to pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/FixedPartitionCountIntPartitioner.java
index cd83ee9..3969098 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/IntPartitioner.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/FixedPartitionCountIntPartitioner.java
@@ -19,28 +19,24 @@
package org.apache.pinot.grigio.common;
import com.google.common.base.Preconditions;
-import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.PartitionInfo;
-import java.util.List;
-import java.util.Map;
-public class IntPartitioner implements Partitioner {
+/**
+ * Fixed partition count partitioner that assumes that the primary key is integer and use it as the partition directly
+ */
+public class FixedPartitionCountIntPartitioner extends FixedPartitionCountPartitioner {
+
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
Preconditions.checkState(key instanceof Integer, "expect key to be an integer");
- List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
- int numPartitions = partitions.size();
- return (Integer) key % numPartitions;
- }
-
- @Override
- public void close() {
-
- }
-
- @Override
- public void configure(Map<String, ?> configs) {
+ int numPartitions = cluster.partitionCountForTopic(topic);
+ int partitionCount = getPartitionCount();
+ if (partitionCount > numPartitions) {
+ throw new IllegalArgumentException(String
+ .format("Cannot partition to %d partitions for records in topic %s, which has only %d partitions.",
+ partitionCount, topic, numPartitions));
+ }
+ return (Integer) key % partitionCount;
}
}
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/IntPartitioner.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/FixedPartitionCountPartitioner.java
similarity index 65%
rename from pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/IntPartitioner.java
rename to pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/FixedPartitionCountPartitioner.java
index cd83ee9..7e36abb 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/IntPartitioner.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/FixedPartitionCountPartitioner.java
@@ -19,20 +19,21 @@
package org.apache.pinot.grigio.common;
import com.google.common.base.Preconditions;
+import java.util.Map;
import org.apache.kafka.clients.producer.Partitioner;
-import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.PartitionInfo;
-import java.util.List;
-import java.util.Map;
-public class IntPartitioner implements Partitioner {
- @Override
- public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
- Preconditions.checkState(key instanceof Integer, "expect key to be an integer");
- List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
- int numPartitions = partitions.size();
- return (Integer) key % numPartitions;
+/**
+ * Kafka partitioner that partition records to a fixed number of partitions.
+ * i.e., partition results will not change even when more partitions are added to the Kafka topic
+ */
+public abstract class FixedPartitionCountPartitioner implements Partitioner {
+
+ private static final String PARTITION_COUNT = "partition.count";
+ private int partitionCount;
+
+ int getPartitionCount() {
+ return partitionCount;
}
@Override
@@ -42,5 +43,7 @@ public class IntPartitioner implements Partitioner {
@Override
public void configure(Map<String, ?> configs) {
+ partitionCount = Integer.parseInt((String)configs.get(PARTITION_COUNT));
+ Preconditions.checkState(partitionCount > 0, "Partition count must be greater than 0");
}
}
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/KeyCoordinatorQueueProducer.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/KeyCoordinatorQueueProducer.java
index 3cba067..ee417f2 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/KeyCoordinatorQueueProducer.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/KeyCoordinatorQueueProducer.java
@@ -22,10 +22,10 @@ import com.google.common.base.Preconditions;
import org.apache.commons.configuration.Configuration;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.pinot.grigio.common.CoordinatorConfig;
import org.apache.pinot.grigio.common.DistributedCommonUtils;
+import org.apache.pinot.grigio.common.FixedPartitionCountBytesPartitioner;
import org.apache.pinot.grigio.common.config.CommonConfig;
import org.apache.pinot.grigio.common.messages.KeyCoordinatorQueueMsg;
import org.apache.pinot.grigio.common.metrics.GrigioMetrics;
@@ -67,7 +67,7 @@ public class KeyCoordinatorQueueProducer extends KafkaQueueProducer<byte[], KeyC
conf.subset(CoordinatorConfig.KAFKA_CONFIG.KAFKA_CONFIG_KEY));
kafkaProducerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
- kafkaProducerConfig.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DefaultPartitioner.class.getName());
+ kafkaProducerConfig.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, FixedPartitionCountBytesPartitioner.class.getName());
DistributedCommonUtils.setKakfaLosslessProducerConfig(kafkaProducerConfig, hostname);
_kafkaProducer = new KafkaProducer<>(kafkaProducerConfig);
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/LogCoordinatorQueueProducer.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/LogCoordinatorQueueProducer.java
index 368cf35..9d2b0b7 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/LogCoordinatorQueueProducer.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/LogCoordinatorQueueProducer.java
@@ -29,7 +29,7 @@ import org.apache.pinot.grigio.common.metrics.GrigioMetrics;
import org.apache.pinot.grigio.common.utils.CommonUtils;
import org.apache.pinot.grigio.common.CoordinatorConfig;
import org.apache.pinot.grigio.common.DistributedCommonUtils;
-import org.apache.pinot.grigio.common.IntPartitioner;
+import org.apache.pinot.grigio.common.FixedPartitionCountIntPartitioner;
import java.util.Properties;
@@ -65,7 +65,7 @@ public class LogCoordinatorQueueProducer extends KafkaQueueProducer<Integer, Log
_grigioMetrics = grigioMetrics;
kafkaProducerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
- kafkaProducerConfig.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, IntPartitioner.class.getName());
+ kafkaProducerConfig.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, FixedPartitionCountIntPartitioner.class.getName());
DistributedCommonUtils.setKakfaLosslessProducerConfig(kafkaProducerConfig, hostName);
this._kafkaProducer = new KafkaProducer<>(kafkaProducerConfig);
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/VersionMsgQueueProducer.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/VersionMsgQueueProducer.java
index 4d1e620..edacd33 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/VersionMsgQueueProducer.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/VersionMsgQueueProducer.java
@@ -25,7 +25,7 @@ import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.pinot.grigio.common.CoordinatorConfig;
import org.apache.pinot.grigio.common.DistributedCommonUtils;
-import org.apache.pinot.grigio.common.IntPartitioner;
+import org.apache.pinot.grigio.common.FixedPartitionCountIntPartitioner;
import org.apache.pinot.grigio.common.config.CommonConfig;
import org.apache.pinot.grigio.common.messages.KeyCoordinatorQueueMsg;
import org.apache.pinot.grigio.common.metrics.GrigioMetrics;
@@ -66,7 +66,7 @@ public class VersionMsgQueueProducer extends KafkaQueueProducer<Integer, KeyCoor
conf.subset(CoordinatorConfig.KAFKA_CONFIG.KAFKA_CONFIG_KEY));
kafkaProducerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
- kafkaProducerConfig.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, IntPartitioner.class.getName());
+ kafkaProducerConfig.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, FixedPartitionCountIntPartitioner.class.getName());
DistributedCommonUtils.setKakfaLosslessProducerConfig(kafkaProducerConfig, hostname);
_kafkaProducer = new KafkaProducer<>(kafkaProducerConfig);
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/starter/KeyCoordinatorConf.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/starter/KeyCoordinatorConf.java
index a48a335..d4ccf02 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/starter/KeyCoordinatorConf.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/starter/KeyCoordinatorConf.java
@@ -62,7 +62,7 @@ public class KeyCoordinatorConf extends PropertiesConfiguration {
public static final String KAFKA_CONSUMER_GROUP_ID_PREFIX = "pinot_upsert_kc_consumerGroup_";
private static final String KC_MESSAGE_TOPIC = "kc.message.topic";
- private static final String KC_MESSAGE_PARTITION_COUNT = "kc.message.partition.count"; // todo: get partition count from topic
+ private static final String KC_MESSAGE_PARTITION_COUNT = "kc.message.partition.count";
public static final String KC_OUTPUT_TOPIC_PREFIX_KEY = "kc.output.topic.prefix";
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountBytesPartitionerTest.java b/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountBytesPartitionerTest.java
new file mode 100644
index 0000000..9a4b818
--- /dev/null
+++ b/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountBytesPartitionerTest.java
@@ -0,0 +1,72 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.grigio.common;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.common.Cluster;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+
+
+public class FixedPartitionCountBytesPartitionerTest {
+
+ private Map<String, String> configs;
+
+ @BeforeClass
+ public void setUp() {
+ configs = new HashMap<>();
+ configs.put("partition.count", "4");
+ }
+
+ @Test
+ public void testPartition() {
+ FixedPartitionCountBytesPartitioner partitioner = new FixedPartitionCountBytesPartitioner();
+ partitioner.configure(configs);
+
+ String topic1 = "test-topic1";
+ String topic2 = "test-topic2";
+ Cluster cluster = mock(Cluster.class);
+ when(cluster.partitionCountForTopic(topic1)).thenReturn(4);
+ when(cluster.partitionCountForTopic(topic2)).thenReturn(8);
+
+ String key = "test-key";
+ byte[] keyBytes = key.getBytes();
+
+ int partitionResult1 = partitioner.partition(topic1, key, keyBytes, null, null, cluster);
+ int partitionResult2 = partitioner.partition(topic2, key, keyBytes, null, null, cluster);
+ assertEquals(partitionResult1, partitionResult2);
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testPartitionFailed() {
+ FixedPartitionCountBytesPartitioner partitioner = new FixedPartitionCountBytesPartitioner();
+ partitioner.configure(configs);
+
+ String topic = "test-topic";
+ Cluster cluster = mock(Cluster.class);
+ when(cluster.partitionCountForTopic(topic)).thenReturn(2);
+
+ partitioner.partition(topic, null, null, null, null, cluster);
+ }
+}
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountIntPartitionerTest.java b/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountIntPartitionerTest.java
new file mode 100644
index 0000000..6581ebb
--- /dev/null
+++ b/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountIntPartitionerTest.java
@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.grigio.common;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.common.Cluster;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+
+
+public class FixedPartitionCountIntPartitionerTest {
+
+ private Map<String, String> configs;
+
+ @BeforeClass
+ public void setUp() {
+ configs = new HashMap<>();
+ configs.put("partition.count", "4");
+ }
+
+ @Test
+ public void testPartition() {
+ FixedPartitionCountIntPartitioner partitioner = new FixedPartitionCountIntPartitioner();
+ partitioner.configure(configs);
+
+ String topic1 = "test-topic1";
+ String topic2 = "test-topic2";
+ Cluster cluster = mock(Cluster.class);
+ when(cluster.partitionCountForTopic(topic1)).thenReturn(4);
+ when(cluster.partitionCountForTopic(topic2)).thenReturn(8);
+
+ Integer key = 6;
+
+ int partitionResult1 = partitioner.partition(topic1, key, null, null, null, cluster);
+ int partitionResult2 = partitioner.partition(topic2, key, null, null, null, cluster);
+ assertEquals(partitionResult1, 2);
+ assertEquals(partitionResult2, 2);
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testPartitionFailed() {
+ FixedPartitionCountIntPartitioner partitioner = new FixedPartitionCountIntPartitioner();
+ partitioner.configure(configs);
+
+ String topic = "test-topic";
+ Cluster cluster = mock(Cluster.class);
+ when(cluster.partitionCountForTopic(topic)).thenReturn(2);
+
+ Integer key = 6;
+
+ partitioner.partition(topic, key, null, null, null, cluster);
+ }
+}
\ No newline at end of file
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker b/pinot-grigio/pinot-grigio-coordinator/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker
new file mode 100644
index 0000000..ca6ee9c
--- /dev/null
+++ b/pinot-grigio/pinot-grigio-coordinator/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker
@@ -0,0 +1 @@
+mock-maker-inline
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 12/12: add one more lag metrics to kc
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 5f33a29351367b86d0bd29716610286de97fbd3b
Author: james Shao <sj...@uber.com>
AuthorDate: Fri Nov 8 15:01:23 2019 -0800
add one more lag metrics to kc
Reviewers: tingchen, bzzhang, #streaming_pinot
Reviewed By: bzzhang, #streaming_pinot
Differential Revision: https://code.uberinternal.com/D3574189
---
.../java/org/apache/pinot/grigio/common/metrics/GrigioGauge.java | 1 +
.../apache/pinot/grigio/common/rpcQueue/QueueConsumerRecord.java | 8 +++++++-
.../pinot/grigio/common/rpcQueue/QueueConsumerRecordTest.java | 3 ++-
.../apache/pinot/grigio/common/rpcQueue/KafkaQueueConsumer.java | 3 ++-
.../keyCoordinator/internal/DistributedKeyCoordinatorCore.java | 2 ++
5 files changed, 14 insertions(+), 3 deletions(-)
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/metrics/GrigioGauge.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/metrics/GrigioGauge.java
index 00c5bfc..f9838c1 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/metrics/GrigioGauge.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/metrics/GrigioGauge.java
@@ -31,6 +31,7 @@ public enum GrigioGauge implements AbstractMetrics.Gauge {
VERSION_PRODUCED("versions", MetricsType.KC_ONLY),
KC_VERSION_CONSUMED("versions", MetricsType.KC_ONLY),
SERVER_VERSION_CONSUMED("versions", MetricsType.SERVER_ONLY),
+ KC_INPUT_MESSAGE_LAG_MS("milliseconds", MetricsType.KC_ONLY)
;
private final String _gaugeName;
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/rpcQueue/QueueConsumerRecord.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/rpcQueue/QueueConsumerRecord.java
index e3fa109..7e86937 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/rpcQueue/QueueConsumerRecord.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/rpcQueue/QueueConsumerRecord.java
@@ -29,13 +29,15 @@ public class QueueConsumerRecord<K, V> {
private final long _offset;
private final K _key;
private final V _record;
+ private final long _timestamp;
- public QueueConsumerRecord(String topic, int partition, long offset, K key, V record) {
+ public QueueConsumerRecord(String topic, int partition, long offset, K key, V record, long timestamp) {
this._topic = topic;
this._partition = partition;
this._offset = offset;
this._key = key;
this._record = record;
+ this._timestamp = timestamp;
}
public String getTopic() {
@@ -57,4 +59,8 @@ public class QueueConsumerRecord<K, V> {
public V getRecord() {
return _record;
}
+
+ public long getTimestamp() {
+ return _timestamp;
+ }
}
diff --git a/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/rpcQueue/QueueConsumerRecordTest.java b/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/rpcQueue/QueueConsumerRecordTest.java
index 3d7f006..b129d08 100644
--- a/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/rpcQueue/QueueConsumerRecordTest.java
+++ b/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/rpcQueue/QueueConsumerRecordTest.java
@@ -25,12 +25,13 @@ public class QueueConsumerRecordTest {
@Test
public void testGets() {
- QueueConsumerRecord<String, String> record = new QueueConsumerRecord<>("topic", 1, 2, "key", "record");
+ QueueConsumerRecord<String, String> record = new QueueConsumerRecord<>("topic", 1, 2, "key", "record", 123);
Assert.assertEquals(record.getTopic(), "topic");
Assert.assertEquals(record.getPartition(), 1);
Assert.assertEquals(record.getOffset(), 2);
Assert.assertEquals(record.getKey(), "key");
Assert.assertEquals(record.getRecord(), "record");
+ Assert.assertEquals(record.getTimestamp(), 123);
}
}
\ No newline at end of file
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/KafkaQueueConsumer.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/KafkaQueueConsumer.java
index 1391c22..ab00e39 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/KafkaQueueConsumer.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/rpcQueue/KafkaQueueConsumer.java
@@ -122,7 +122,8 @@ public abstract class KafkaQueueConsumer<K, V> implements QueueConsumer<K, V> {
msgList = new ArrayList<>(records.count());
for (ConsumerRecord<K, V> record : records) {
msgList.add(
- new QueueConsumerRecord<>(record.topic(), record.partition(), record.offset(), record.key(), record.value()));
+ new QueueConsumerRecord<>(record.topic(), record.partition(), record.offset(), record.key(), record.value(),
+ record.timestamp()));
}
}
getMetrics().addMeteredGlobalValue(GrigioMeter.MESSAGE_INGEST_COUNT_PER_BATCH, msgList.size());
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/internal/DistributedKeyCoordinatorCore.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/internal/DistributedKeyCoordinatorCore.java
index 1cdef42..044673f 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/internal/DistributedKeyCoordinatorCore.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/internal/DistributedKeyCoordinatorCore.java
@@ -228,6 +228,8 @@ public class DistributedKeyCoordinatorCore {
LOGGER.warn("exception while trying to put message to queue", e);
}
});
+ _metrics.setValueOfGlobalGauge(GrigioGauge.KC_INPUT_MESSAGE_LAG_MS,
+ System.currentTimeMillis() - records.get(records.size() - 1).getTimestamp());
}
} catch (Exception ex) {
LOGGER.error("encountered exception in consumer ingest loop, will retry", ex);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 07/12: change validFrom/validUntil column data
type to int
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 11adb6e970bbb2230a9155931dff5b414aecce69
Author: james Shao <sj...@uber.com>
AuthorDate: Tue Nov 5 15:15:12 2019 -0800
change validFrom/validUntil column data type to int
Summary:
change validFrom/until data type from long to int, change memory usage per document from
16 bytes to 8 bytes
Reviewers: bzzhang, csoman, tingchen
Reviewed By: bzzhang
Differential Revision: https://code.uberinternal.com/D3534725
---
pinot-api/pom.xml | 2 +-
pinot-azure-filesystem/pom.xml | 2 +-
pinot-broker/pom.xml | 2 +-
pinot-common/pom.xml | 2 +-
pinot-connectors/pinot-connector-kafka-0.11/pom.xml | 2 +-
pinot-connectors/pinot-connector-kafka-0.9/pom.xml | 2 +-
pinot-connectors/pom.xml | 2 +-
pinot-controller/pom.xml | 2 +-
pinot-core/pom.xml | 2 +-
.../immutable/ImmutableUpsertSegmentImpl.java | 2 +-
.../indexsegment/mutable/MutableUpsertSegmentImpl.java | 1 +
.../mutable/ValidFromInMemoryVirtualColumnProvider.java | 7 ++++++-
.../mutable/ValidUntilInMemoryVirtualColumnProvider.java | 7 ++++++-
.../mutable/VirtualColumnLongValueReaderWriter.java | 16 +++++++++++-----
pinot-distribution/pom.xml | 2 +-
pinot-grigio/pinot-grigio-common/pom.xml | 2 +-
.../retentionManager/UpdateLogTableRetentionManager.java | 3 ++-
pinot-grigio/pinot-grigio-coordinator/pom.xml | 2 +-
pinot-grigio/pom.xml | 2 +-
pinot-hadoop-filesystem/pom.xml | 2 +-
pinot-hadoop/pom.xml | 2 +-
pinot-integration-tests/pom.xml | 2 +-
pinot-minion/pom.xml | 2 +-
pinot-orc/pom.xml | 2 +-
pinot-parquet/pom.xml | 2 +-
pinot-perf/pom.xml | 2 +-
pinot-server/pom.xml | 2 +-
pinot-tools/pom.xml | 2 +-
pinot-transport/pom.xml | 2 +-
pom.xml | 2 +-
30 files changed, 51 insertions(+), 33 deletions(-)
diff --git a/pinot-api/pom.xml b/pinot-api/pom.xml
index 6d6e96b..8763b4b 100644
--- a/pinot-api/pom.xml
+++ b/pinot-api/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-api</artifactId>
<name>Pinot API</name>
diff --git a/pinot-azure-filesystem/pom.xml b/pinot-azure-filesystem/pom.xml
index fe6c1c3..e42b355 100644
--- a/pinot-azure-filesystem/pom.xml
+++ b/pinot-azure-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-azure-filesystem</artifactId>
<name>Pinot Azure Filesystem</name>
diff --git a/pinot-broker/pom.xml b/pinot-broker/pom.xml
index fbeb628..d625e2d 100644
--- a/pinot-broker/pom.xml
+++ b/pinot-broker/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-broker</artifactId>
<name>Pinot Broker</name>
diff --git a/pinot-common/pom.xml b/pinot-common/pom.xml
index 91df4fe..cde3078 100644
--- a/pinot-common/pom.xml
+++ b/pinot-common/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-common</artifactId>
<name>Pinot Common</name>
diff --git a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
index 50ef1ad..0b32fe0 100644
--- a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
index 409e3cf..f409a8f 100644
--- a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pom.xml b/pinot-connectors/pom.xml
index 749ec5c..b2cfd90 100644
--- a/pinot-connectors/pom.xml
+++ b/pinot-connectors/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-connectors</artifactId>
diff --git a/pinot-controller/pom.xml b/pinot-controller/pom.xml
index 0e0af54..cf1e1f3 100644
--- a/pinot-controller/pom.xml
+++ b/pinot-controller/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-controller</artifactId>
<name>Pinot Controller</name>
diff --git a/pinot-core/pom.xml b/pinot-core/pom.xml
index a4e6ca8..141249d 100644
--- a/pinot-core/pom.xml
+++ b/pinot-core/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-core</artifactId>
<name>Pinot Core</name>
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
index f42255f..80723b4 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
@@ -150,7 +150,7 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
int docId = getDocIdFromSourceOffset(offset);
StringBuilder result = new StringBuilder("matched: ");
for (VirtualColumnLongValueReaderWriter readerWriter : _virtualColumnsReaderWriter) {
- result.append(readerWriter.getLong(docId)).append("; ");
+ result.append(readerWriter.getInt(docId)).append("; ");
}
return result.toString();
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java
index b4d2e15..b69221d 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java
@@ -44,6 +44,7 @@ public class MutableUpsertSegmentImpl extends MutableSegmentImpl implements Upse
private final String _kafkaOffsetColumnName;
+
private final List<VirtualColumnLongValueReaderWriter> _mutableSegmentReaderWriters = new ArrayList<>();
// use map for mapping between kafka offset and docId because we at-most have 1 mutable segment per consumer
// will use more memory, but we can update this later
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidFromInMemoryVirtualColumnProvider.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidFromInMemoryVirtualColumnProvider.java
index c8ea4a1..2dedc8e 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidFromInMemoryVirtualColumnProvider.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidFromInMemoryVirtualColumnProvider.java
@@ -28,12 +28,17 @@ public class ValidFromInMemoryVirtualColumnProvider extends BaseLongVirtualColum
public DataFileReader buildReader(VirtualColumnContext context) {
return new VirtualColumnLongValueReaderWriter(context) {
@Override
- public boolean update(int docId, long value, LogEventType eventType) {
+ public boolean update(int docId, int value, LogEventType eventType) {
if (eventType == LogEventType.INSERT) {
return updateValue(docId, value);
}
return false;
}
+
+ @Override
+ public boolean update(int docId, long value, LogEventType eventType) {
+ return update(docId, Math.toIntExact(value), eventType);
+ }
};
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidUntilInMemoryVirtualColumnProvider.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidUntilInMemoryVirtualColumnProvider.java
index e6e38c0..1a0856e 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidUntilInMemoryVirtualColumnProvider.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidUntilInMemoryVirtualColumnProvider.java
@@ -28,12 +28,17 @@ public class ValidUntilInMemoryVirtualColumnProvider extends BaseLongVirtualColu
public DataFileReader buildReader(VirtualColumnContext context) {
return new VirtualColumnLongValueReaderWriter(context) {
@Override
- public boolean update(int docId, long offset, LogEventType eventType) {
+ public boolean update(int docId, int offset, LogEventType eventType) {
if (eventType == LogEventType.DELETE) {
return this.updateValue(docId, offset);
}
return false;
}
+
+ @Override
+ public boolean update(int docId, long offset, LogEventType eventType) {
+ return update(docId, Math.toIntExact(offset), eventType);
+ }
};
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/VirtualColumnLongValueReaderWriter.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/VirtualColumnLongValueReaderWriter.java
index e9df80d..75d1b74 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/VirtualColumnLongValueReaderWriter.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/VirtualColumnLongValueReaderWriter.java
@@ -30,18 +30,17 @@ import java.util.Arrays;
public abstract class VirtualColumnLongValueReaderWriter extends BaseVirtualColumnSingleValueReaderWriter<ChunkReaderContext> {
private static final Logger LOGGER = LoggerFactory.getLogger(VirtualColumnLongValueReaderWriter.class);
- private static final long DEFAULT_VALUE = Long.MIN_VALUE;
private final VirtualColumnContext _context;
private int _totalDocSize;
private int _currentMaxDocId;
- private final long[] _values;
- private final long DEFAULT_NEW_VALUE = -1;
+ private final int [] _values;
+ private final int DEFAULT_NEW_VALUE = -1;
public VirtualColumnLongValueReaderWriter(VirtualColumnContext context) {
_context = context;
_totalDocSize = context.getTotalDocCount();
- _values = new long[_totalDocSize];
+ _values = new int[_totalDocSize];
if (!_context.isMutableSegment()) {
Arrays.fill(_values, -1);
}
@@ -84,7 +83,7 @@ public abstract class VirtualColumnLongValueReaderWriter extends BaseVirtualColu
* method to update the internal data to value at a given location
* synchronized to ensure the we will not modify the internal array at the same time from multiple threads
*/
- protected synchronized boolean updateValue(int docId, long value) {
+ protected synchronized boolean updateValue(int docId, int value) {
if (docId >= _totalDocSize) {
throw new RuntimeException(String.format("new record docId %s is larger than capacity %s", docId, _totalDocSize));
}
@@ -100,5 +99,12 @@ public abstract class VirtualColumnLongValueReaderWriter extends BaseVirtualColu
}
}
+ // ensure backward compatibility
+ protected boolean updateValue(int docId, long value) {
+ return updateValue(docId, Math.toIntExact(value));
+ }
+
public abstract boolean update(int docId, long value, LogEventType eventType);
+
+ public abstract boolean update(int docId, int value, LogEventType eventType);
}
diff --git a/pinot-distribution/pom.xml b/pinot-distribution/pom.xml
index fb652d0..ebf0146 100644
--- a/pinot-distribution/pom.xml
+++ b/pinot-distribution/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-distribution</artifactId>
<name>Pinot Distribution</name>
diff --git a/pinot-grigio/pinot-grigio-common/pom.xml b/pinot-grigio/pinot-grigio-common/pom.xml
index 6d9d4bb..42e07c0 100644
--- a/pinot-grigio/pinot-grigio-common/pom.xml
+++ b/pinot-grigio/pinot-grigio-common/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManager.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManager.java
index 62215b8..85efb47 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManager.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManager.java
@@ -78,7 +78,7 @@ public abstract class UpdateLogTableRetentionManager {
int partition = llcSegmentName.getPartitionId();
if (!partitionLastSeg.containsKey(partition)) {
partitionLastSeg.put(partition, llcSegmentName);
- } else if (partitionLastSeg.get(partition).getSequenceNumber() < llcSegmentName.getSequenceNumber()) {
+ } else if (compareSegment(llcSegmentName, partitionLastSeg.get(partition))) {
partitionLastSeg.put(partition, llcSegmentName);
}
}
@@ -110,6 +110,7 @@ public abstract class UpdateLogTableRetentionManager {
}
// we most probably got a segment that is from a deleted table or segment assigned to another table
// assume we don't do re-balance, we won't do refresh
+ LOGGER.warn("adding segment {} to blacklist", segmentName);
_blacklistedSegments.put(segmentName, segmentName);
return false;
}
diff --git a/pinot-grigio/pinot-grigio-coordinator/pom.xml b/pinot-grigio/pinot-grigio-coordinator/pom.xml
index bc85f0f..19499d1 100644
--- a/pinot-grigio/pinot-grigio-coordinator/pom.xml
+++ b/pinot-grigio/pinot-grigio-coordinator/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pinot-grigio-coordinator</artifactId>
diff --git a/pinot-grigio/pom.xml b/pinot-grigio/pom.xml
index 2766d3a..3cdad0c 100644
--- a/pinot-grigio/pom.xml
+++ b/pinot-grigio/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pinot-grigio</artifactId>
diff --git a/pinot-hadoop-filesystem/pom.xml b/pinot-hadoop-filesystem/pom.xml
index 6d44587..49b0a8a 100644
--- a/pinot-hadoop-filesystem/pom.xml
+++ b/pinot-hadoop-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-hadoop-filesystem</artifactId>
<name>Pinot Hadoop Filesystem</name>
diff --git a/pinot-hadoop/pom.xml b/pinot-hadoop/pom.xml
index 2272b24..928a11b 100644
--- a/pinot-hadoop/pom.xml
+++ b/pinot-hadoop/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-hadoop</artifactId>
<name>Pinot Hadoop</name>
diff --git a/pinot-integration-tests/pom.xml b/pinot-integration-tests/pom.xml
index 1095763..9d7cb12 100644
--- a/pinot-integration-tests/pom.xml
+++ b/pinot-integration-tests/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-integration-tests</artifactId>
<name>Pinot Integration Tests</name>
diff --git a/pinot-minion/pom.xml b/pinot-minion/pom.xml
index a7352ee..c057094 100644
--- a/pinot-minion/pom.xml
+++ b/pinot-minion/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-minion</artifactId>
<name>Pinot Minion</name>
diff --git a/pinot-orc/pom.xml b/pinot-orc/pom.xml
index e046ca8..34a8740 100644
--- a/pinot-orc/pom.xml
+++ b/pinot-orc/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-parquet/pom.xml b/pinot-parquet/pom.xml
index 5c5b9b0..2d7dbb0 100644
--- a/pinot-parquet/pom.xml
+++ b/pinot-parquet/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-perf/pom.xml b/pinot-perf/pom.xml
index 3312026..060ebaf 100644
--- a/pinot-perf/pom.xml
+++ b/pinot-perf/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-perf</artifactId>
<name>Pinot Perf</name>
diff --git a/pinot-server/pom.xml b/pinot-server/pom.xml
index 9463c64..ac77f20 100644
--- a/pinot-server/pom.xml
+++ b/pinot-server/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-server</artifactId>
<name>Pinot Server</name>
diff --git a/pinot-tools/pom.xml b/pinot-tools/pom.xml
index 7aa523f..4b4e686 100644
--- a/pinot-tools/pom.xml
+++ b/pinot-tools/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-tools</artifactId>
<name>Pinot Tools</name>
diff --git a/pinot-transport/pom.xml b/pinot-transport/pom.xml
index dc748c7..b977921 100644
--- a/pinot-transport/pom.xml
+++ b/pinot-transport/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
</parent>
<artifactId>pinot-transport</artifactId>
<name>Pinot Transport</name>
diff --git a/pom.xml b/pom.xml
index c3d0cf0..77e8647 100644
--- a/pom.xml
+++ b/pom.xml
@@ -30,7 +30,7 @@
<groupId>org.apache.pinot</groupId>
<artifactId>pinot</artifactId>
- <version>0.2.2.5.45-SNAPSHOT</version>
+ <version>0.2.2.5.49-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Pinot</name>
<description>A realtime distributed OLAP datastore</description>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 10/12: improve pinot server upsert segment
loading time
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 7c8cc38a0aec511e2afb41d7ef9d1cdf1c3174d6
Author: james Shao <sj...@uber.com>
AuthorDate: Mon Nov 11 13:47:16 2019 -0800
improve pinot server upsert segment loading time
Summary:
previously pinot server spent too much time on loading virtual column for upsert segment (10+ seconds for segment with 5M updates), we did some investigation and found issues related to how we load the virtual columns. The major issues boils down to:
1. for each message, we provide an message to Preconditions.checkState(...) with String.format(..). This cause java to construct a new string with String.format for every update events and it is very costly
2. various synchronization that are no longer necessary
3. various minor improvement by picking correct data types/optimization
after optimization the time for init virtual column should be < 0.5 seconds for segments with 5M updates. The overall segments loading time should be within 1-2 seconds (including loading virtual column file from disk and others related work)
Reviewers: bzzhang, tingchen, #streaming_pinot
Reviewed By: bzzhang, #streaming_pinot
Maniphest Tasks: T4441685
Differential Revision: https://code.uberinternal.com/D3571549
---
pinot-api/pom.xml | 2 +-
pinot-azure-filesystem/pom.xml | 2 +-
pinot-broker/pom.xml | 2 +-
pinot-common/pom.xml | 2 +-
.../pinot-connector-kafka-0.11/pom.xml | 2 +-
pinot-connectors/pinot-connector-kafka-0.9/pom.xml | 2 +-
pinot-connectors/pom.xml | 2 +-
pinot-controller/pom.xml | 2 +-
pinot-core/pom.xml | 2 +-
.../immutable/ImmutableUpsertSegmentImpl.java | 119 +++++++++++++--------
.../segment/updater/UpsertWaterMarkManager.java | 13 ++-
.../ValidFromInMemoryVirtualColumnProvider.java | 3 +-
.../ValidUntilInMemoryVirtualColumnProvider.java | 3 +-
.../VirtualColumnLongValueReaderWriter.java | 28 +++--
.../immutable/ImmutableUpsertSegmentImplTest.java | 118 ++++++++++++++++++++
pinot-distribution/pom.xml | 2 +-
pinot-grigio/pinot-grigio-common/pom.xml | 2 +-
.../pinot/grigio/common/metrics/GrigioMeter.java | 1 -
.../SegmentUpdateLogStorageProvider.java | 7 +-
.../storageProvider/UpdateLogStorageProvider.java | 1 -
.../SegmentUpdateLogStorageProviderTest.java | 15 +++
pinot-grigio/pinot-grigio-coordinator/pom.xml | 2 +-
pinot-grigio/pom.xml | 2 +-
pinot-hadoop-filesystem/pom.xml | 2 +-
pinot-hadoop/pom.xml | 2 +-
pinot-integration-tests/pom.xml | 2 +-
pinot-minion/pom.xml | 2 +-
pinot-orc/pom.xml | 2 +-
pinot-parquet/pom.xml | 2 +-
pinot-perf/pom.xml | 2 +-
pinot-server/pom.xml | 2 +-
pinot-tools/pom.xml | 2 +-
pinot-transport/pom.xml | 2 +-
pom.xml | 2 +-
34 files changed, 262 insertions(+), 94 deletions(-)
diff --git a/pinot-api/pom.xml b/pinot-api/pom.xml
index 8763b4b..41391eb 100644
--- a/pinot-api/pom.xml
+++ b/pinot-api/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-api</artifactId>
<name>Pinot API</name>
diff --git a/pinot-azure-filesystem/pom.xml b/pinot-azure-filesystem/pom.xml
index e42b355..ce07b85 100644
--- a/pinot-azure-filesystem/pom.xml
+++ b/pinot-azure-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-azure-filesystem</artifactId>
<name>Pinot Azure Filesystem</name>
diff --git a/pinot-broker/pom.xml b/pinot-broker/pom.xml
index d625e2d..4e92b3f 100644
--- a/pinot-broker/pom.xml
+++ b/pinot-broker/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-broker</artifactId>
<name>Pinot Broker</name>
diff --git a/pinot-common/pom.xml b/pinot-common/pom.xml
index cde3078..4ce2e32 100644
--- a/pinot-common/pom.xml
+++ b/pinot-common/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-common</artifactId>
<name>Pinot Common</name>
diff --git a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
index 0b32fe0..5e384d5 100644
--- a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
index f409a8f..daf74ad 100644
--- a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pom.xml b/pinot-connectors/pom.xml
index b2cfd90..f4e8763 100644
--- a/pinot-connectors/pom.xml
+++ b/pinot-connectors/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-connectors</artifactId>
diff --git a/pinot-controller/pom.xml b/pinot-controller/pom.xml
index cf1e1f3..259300d 100644
--- a/pinot-controller/pom.xml
+++ b/pinot-controller/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-controller</artifactId>
<name>Pinot Controller</name>
diff --git a/pinot-core/pom.xml b/pinot-core/pom.xml
index 141249d..06309cc 100644
--- a/pinot-core/pom.xml
+++ b/pinot-core/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-core</artifactId>
<name>Pinot Core</name>
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
index 80723b4..1537689 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
@@ -19,10 +19,9 @@
package org.apache.pinot.core.indexsegment.immutable;
import com.clearspring.analytics.util.Preconditions;
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.Multimap;
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.pinot.common.Utils;
import org.apache.pinot.common.config.TableNameBuilder;
-import org.apache.pinot.common.data.Schema;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.core.indexsegment.UpsertSegment;
import org.apache.pinot.core.io.reader.BaseSingleColumnSingleValueReader;
@@ -42,7 +41,6 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -51,19 +49,19 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
private static final Logger LOGGER = LoggerFactory.getLogger(ImmutableUpsertSegmentImpl.class);
- private final List<VirtualColumnLongValueReaderWriter> _virtualColumnsReaderWriter = new ArrayList<>();
-
+ private final List<VirtualColumnLongValueReaderWriter> _virtualColumnsReaderWriter;
private final String _tableNameWithType;
private final String _segmentName;
- private final Schema _schema;
private final int _totalDoc;
- private final ColumnIndexContainer _offsetColumnIndexContainer;
- private final UpsertWaterMarkManager _upsertWaterMarkManager;
private long _minSourceOffset;
+ private final UpsertWaterMarkManager _upsertWaterMarkManager;
+ private final UpdateLogStorageProvider _updateLogStorageProvider;
// use array for mapping bewteen offset to docId, where actual offset = min_offset + array_index
// use 4 bytes per record
private int[] _sourceOffsetToDocIdArray;
+ private static final int DEFAULT_DOC_ID_FOR_MISSING_ENTRY = -1;
+
public ImmutableUpsertSegmentImpl(SegmentDirectory segmentDirectory,
SegmentMetadataImpl segmentMetadata,
Map<String, ColumnIndexContainer> columnIndexContainerMap,
@@ -73,23 +71,44 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
_tableNameWithType = TableNameBuilder.ensureTableNameWithType(segmentMetadata.getTableName(),
CommonConstants.Helix.TableType.REALTIME);
_segmentName = segmentMetadata.getName();
- _schema = segmentMetadata.getSchema();
_totalDoc = segmentMetadata.getTotalDocs();
- _offsetColumnIndexContainer = columnIndexContainerMap.get(_schema.getOffsetKey());
_upsertWaterMarkManager = UpsertWaterMarkManager.getInstance();
+ _updateLogStorageProvider = UpdateLogStorageProvider.getInstance();
+ _virtualColumnsReaderWriter = new ArrayList<>();
for (Map.Entry<String, ColumnIndexContainer> entry: columnIndexContainerMap.entrySet()) {
String columnName = entry.getKey();
ColumnIndexContainer container = entry.getValue();
- if (_schema.isVirtualColumn(columnName) && (container.getForwardIndex() instanceof VirtualColumnLongValueReaderWriter)) {
+ if (segmentMetadata.getSchema().isVirtualColumn(columnName) && (container.getForwardIndex() instanceof VirtualColumnLongValueReaderWriter)) {
_virtualColumnsReaderWriter.add((VirtualColumnLongValueReaderWriter) container.getForwardIndex());
}
}
- buildOffsetToDocIdMap();
+ buildOffsetToDocIdMap(columnIndexContainerMap.get(segmentMetadata.getSchema().getOffsetKey()));
}
- private void buildOffsetToDocIdMap() {
- final DataFileReader reader = _offsetColumnIndexContainer.getForwardIndex();
- final Dictionary dictionary = _offsetColumnIndexContainer.getDictionary();
+ /** constructor used for creating instance in test cases
+ * should not be used for creating regular segment
+ */
+ @VisibleForTesting
+ protected ImmutableUpsertSegmentImpl(List<VirtualColumnLongValueReaderWriter> readerWriters,
+ int totalDoc, UpsertWaterMarkManager manager,
+ UpdateLogStorageProvider updateLogStorageProvider,
+ long minSourceOffset, int[] offsetToDocId) {
+ super(null, null, null, null);
+ _tableNameWithType = "testTable";
+ _segmentName = "testSegment";
+ _virtualColumnsReaderWriter = readerWriters;
+ _totalDoc = totalDoc;
+ _upsertWaterMarkManager = manager;
+ _updateLogStorageProvider = updateLogStorageProvider;
+ _minSourceOffset = minSourceOffset;
+ _sourceOffsetToDocIdArray = offsetToDocId;
+
+ }
+
+ private void buildOffsetToDocIdMap(ColumnIndexContainer offsetColumnIndexContainer) {
+ long start = System.currentTimeMillis();
+ final DataFileReader reader = offsetColumnIndexContainer.getForwardIndex();
+ final Dictionary dictionary = offsetColumnIndexContainer.getDictionary();
Map<Long, Integer> kafkaOffsetToDocIdMap = new HashMap<>();
long minOffset = Long.MAX_VALUE;
long maxOffset = 0;
@@ -114,15 +133,13 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
int size = Math.toIntExact(maxOffset - minOffset + 1);
_sourceOffsetToDocIdArray = new int[size];
for (int i = 0; i < size; i++) {
- if (kafkaOffsetToDocIdMap.containsKey(i + minOffset)) {
- _sourceOffsetToDocIdArray[i] = kafkaOffsetToDocIdMap.get(i + minOffset);
- } else {
- _sourceOffsetToDocIdArray[i] = -1;
- }
+ _sourceOffsetToDocIdArray[i] = kafkaOffsetToDocIdMap.getOrDefault(i + minOffset,
+ DEFAULT_DOC_ID_FOR_MISSING_ENTRY);
}
} else {
throw new RuntimeException("unexpected forward reader type for kafka offset column " + reader.getClass());
}
+ LOGGER.info("built offset to DocId map for segment {} with {} documents in {} ms", _segmentName, _totalDoc, System.currentTimeMillis() - start);
}
public static ImmutableUpsertSegmentImpl copyOf(ImmutableSegmentImpl immutableSegment) {
@@ -163,32 +180,48 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
@Override
public void initVirtualColumn() throws IOException {
long start = System.currentTimeMillis();
- List<UpdateLogEntry> updateLogEntries = UpdateLogStorageProvider.getInstance().getAllMessages(_tableNameWithType, _segmentName);
- Multimap<Long, UpdateLogEntry> updateLogEntryMap = ArrayListMultimap.create();
- for (UpdateLogEntry logEntry: updateLogEntries) {
- updateLogEntryMap.put(logEntry.getOffset(), logEntry);
- }
- for (int i = 0; i < _sourceOffsetToDocIdArray.length; i++) {
- if (_sourceOffsetToDocIdArray[i] != -1) {
- final long offset = i + _minSourceOffset;
- final int docId = _sourceOffsetToDocIdArray[i];
- if (updateLogEntryMap.containsKey(offset)) {
- boolean updated = false;
- Collection<UpdateLogEntry> entries = updateLogEntryMap.get(offset);
- UpdateLogEntry lastEntry = null;
- for (UpdateLogEntry entry : entries) {
- lastEntry = entry;
- for (VirtualColumnLongValueReaderWriter readerWriter : _virtualColumnsReaderWriter) {
- updated = readerWriter.update(docId, entry.getValue(), entry.getType()) || updated;
+ final List<UpdateLogEntry> updateLogEntries = _updateLogStorageProvider.getAllMessages(_tableNameWithType, _segmentName);
+ LOGGER.info("load {} update log entry from update log storage provider for segment {} in {} ms",
+ updateLogEntries.size(), _segmentName, System.currentTimeMillis() - start);
+
+ start = System.currentTimeMillis();
+ final long maxOffset = _totalDoc + _minSourceOffset;
+ int unmatchedLogEntryCount = 0;
+ try {
+ Map<Integer, Long> partitionToHighestWatermark = new HashMap<>();
+ int readerWriteCount = _virtualColumnsReaderWriter.size();
+ for (UpdateLogEntry logEntry: updateLogEntries) {
+ final int partition = logEntry.getPartition();
+ final long offset = logEntry.getOffset();
+ if (offset >= _minSourceOffset && offset < maxOffset) {
+ final int docId = _sourceOffsetToDocIdArray[Math.toIntExact(offset - _minSourceOffset)];
+ if (docId != DEFAULT_DOC_ID_FOR_MISSING_ENTRY) {
+ // use traditional for loop over int instead of foreach to give hints for JIT to do loop unroll in byte code
+ for (int i = 0; i < readerWriteCount; i++) {
+ _virtualColumnsReaderWriter.get(i).update(docId, logEntry.getValue(), logEntry.getType());
}
+ if (logEntry.getValue() > partitionToHighestWatermark.getOrDefault(partition, -1L)) {
+ partitionToHighestWatermark.put(partition, logEntry.getValue());
+ }
+ } else {
+ LOGGER.error("segment {} got in-range update log at offset {} but no matching docId", _segmentName, offset);
}
- if (updated) {
- _upsertWaterMarkManager.processMessage(_tableNameWithType, _segmentName, lastEntry);
- }
+ } else {
+ unmatchedLogEntryCount++;
}
}
+ if (unmatchedLogEntryCount > 0) {
+ LOGGER.info("segment {} encountered {} update logs that are outside of its range", _segmentName,
+ unmatchedLogEntryCount);
+ }
+ partitionToHighestWatermark.forEach((partition, value) ->
+ _upsertWaterMarkManager.processVersionUpdate(_tableNameWithType, partition, value));
+
+ } catch (Exception e) {
+ LOGGER.error("failed to load the offset with thread pool");
+ Utils.rethrowException(e);
}
- LOGGER.info("loaded {} update log entries for current immutable segment {} in {} ms", updateLogEntries.size(),
+ LOGGER.info("populated all log entries to virtual columns for current immutable segment {} in {} ms",
_segmentName, System.currentTimeMillis() - start);
}
@@ -205,7 +238,7 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
throw new RuntimeException("offset outside range");
} else {
int position = Math.toIntExact(offset - _minSourceOffset);
- if (_sourceOffsetToDocIdArray[position] == -1) {
+ if (_sourceOffsetToDocIdArray[position] == DEFAULT_DOC_ID_FOR_MISSING_ENTRY) {
LOGGER.error("no docId associated with offset {} for segment {}", offset, _segmentName);
throw new RuntimeException("docId not found");
} else {
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/updater/UpsertWaterMarkManager.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/updater/UpsertWaterMarkManager.java
index 6d37322..bbea037 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/updater/UpsertWaterMarkManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/updater/UpsertWaterMarkManager.java
@@ -22,7 +22,6 @@ import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import org.apache.pinot.grigio.common.metrics.GrigioGauge;
-import org.apache.pinot.grigio.common.metrics.GrigioMeter;
import org.apache.pinot.grigio.common.metrics.GrigioMetrics;
import org.apache.pinot.grigio.common.storageProvider.UpdateLogEntry;
import org.slf4j.Logger;
@@ -62,15 +61,15 @@ public class UpsertWaterMarkManager {
}
long version = logEntry.getValue();
int partition = logEntry.getPartition();
- Preconditions.checkState(partition >= 0, "logEntry has no partition info {} for table ", logEntry.toString(), table);
+ processVersionUpdate(table, partition, version);
+ }
+ public void processVersionUpdate(String table, int partition, long version) {
+ Preconditions.checkState(partition >= 0, "logEntry has invalid version {} for table {}",
+ version, table);
Map<Integer, Long> partitionToHighWaterMark = _highWaterMarkTablePartitionMap.computeIfAbsent(table, t -> new ConcurrentHashMap<>());
long currentVersion = partitionToHighWaterMark.getOrDefault(partition, -1L); // assumes that valid version is non-negative
- if (version < currentVersion) {
- // We expect the version number to increase monotonically unless we are reprocessing previous seen messages.
- _metrics.addMeteredGlobalValue(GrigioMeter.VERSION_LOWER_THAN_CURRENT, 1);
- LOGGER.debug("The latest record {} has lower version than the current one for the table {} ", logEntry, table);
- } else if (version > currentVersion) {
+ if (version > currentVersion) {
partitionToHighWaterMark.put(partition, version);
_metrics.setValueOfTableGauge(String.valueOf(partition), GrigioGauge.SERVER_VERSION_CONSUMED, version);
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidFromInMemoryVirtualColumnProvider.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidFromInMemoryVirtualColumnProvider.java
index 2dedc8e..ebb45e7 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidFromInMemoryVirtualColumnProvider.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidFromInMemoryVirtualColumnProvider.java
@@ -30,7 +30,8 @@ public class ValidFromInMemoryVirtualColumnProvider extends BaseLongVirtualColum
@Override
public boolean update(int docId, int value, LogEventType eventType) {
if (eventType == LogEventType.INSERT) {
- return updateValue(docId, value);
+ updateValue(docId, value);
+ return true;
}
return false;
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidUntilInMemoryVirtualColumnProvider.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidUntilInMemoryVirtualColumnProvider.java
index 1a0856e..9aef94c 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidUntilInMemoryVirtualColumnProvider.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/ValidUntilInMemoryVirtualColumnProvider.java
@@ -30,7 +30,8 @@ public class ValidUntilInMemoryVirtualColumnProvider extends BaseLongVirtualColu
@Override
public boolean update(int docId, int offset, LogEventType eventType) {
if (eventType == LogEventType.DELETE) {
- return this.updateValue(docId, offset);
+ this.updateValue(docId, offset);
+ return true;
}
return false;
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/VirtualColumnLongValueReaderWriter.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/VirtualColumnLongValueReaderWriter.java
index 75d1b74..78f3bda 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/VirtualColumnLongValueReaderWriter.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/virtualcolumn/mutable/VirtualColumnLongValueReaderWriter.java
@@ -31,17 +31,19 @@ public abstract class VirtualColumnLongValueReaderWriter extends BaseVirtualColu
private static final Logger LOGGER = LoggerFactory.getLogger(VirtualColumnLongValueReaderWriter.class);
- private final VirtualColumnContext _context;
private int _totalDocSize;
private int _currentMaxDocId;
private final int [] _values;
private final int DEFAULT_NEW_VALUE = -1;
public VirtualColumnLongValueReaderWriter(VirtualColumnContext context) {
- _context = context;
- _totalDocSize = context.getTotalDocCount();
+ this(context.getTotalDocCount(), context.isMutableSegment());
+ }
+
+ public VirtualColumnLongValueReaderWriter(int totalDocSize, boolean isMutableSegment) {
+ _totalDocSize = totalDocSize;
_values = new int[_totalDocSize];
- if (!_context.isMutableSegment()) {
+ if (!isMutableSegment) {
Arrays.fill(_values, -1);
}
}
@@ -80,10 +82,11 @@ public abstract class VirtualColumnLongValueReaderWriter extends BaseVirtualColu
}
/**
- * method to update the internal data to value at a given location
- * synchronized to ensure the we will not modify the internal array at the same time from multiple threads
+ * update the internal data to value at a given location
+ * assume the update will be idempotent and we won't check if the existing data have been set
+ *
*/
- protected synchronized boolean updateValue(int docId, int value) {
+ protected void updateValue(int docId, int value) {
if (docId >= _totalDocSize) {
throw new RuntimeException(String.format("new record docId %s is larger than capacity %s", docId, _totalDocSize));
}
@@ -91,17 +94,12 @@ public abstract class VirtualColumnLongValueReaderWriter extends BaseVirtualColu
throw new RuntimeException(String.format("failed to update virtual column: with value %s:%s we are trying to " +
"update a value that has not been ingested yet, max doc id %s", docId, value, _currentMaxDocId));
}
- if (_values[docId] == value) {
- return false;
- } else {
- _values[docId] = value;
- return true;
- }
+ _values[docId] = value;
}
// ensure backward compatibility
- protected boolean updateValue(int docId, long value) {
- return updateValue(docId, Math.toIntExact(value));
+ protected void updateValue(int docId, long value) {
+ updateValue(docId, Math.toIntExact(value));
}
public abstract boolean update(int docId, long value, LogEventType eventType);
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImplTest.java b/pinot-core/src/test/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImplTest.java
new file mode 100644
index 0000000..4712625
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImplTest.java
@@ -0,0 +1,118 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.indexsegment.immutable;
+
+import org.apache.pinot.core.segment.updater.UpsertWaterMarkManager;
+import org.apache.pinot.core.segment.virtualcolumn.mutable.VirtualColumnLongValueReaderWriter;
+import org.apache.pinot.grigio.common.messages.LogEventType;
+import org.apache.pinot.grigio.common.storageProvider.UpdateLogEntry;
+import org.apache.pinot.grigio.common.storageProvider.UpdateLogStorageProvider;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+public class ImmutableUpsertSegmentImplTest {
+
+ UpdateLogStorageProvider _mockProvider;
+ UpsertWaterMarkManager _mockUpsertWaterMarkManager;
+ List<VirtualColumnLongValueReaderWriter> _readerWriters = new ArrayList<>();
+
+ @BeforeMethod
+ public void init() {
+ _mockProvider = mock(UpdateLogStorageProvider.class);
+ _mockUpsertWaterMarkManager = mock(UpsertWaterMarkManager.class);
+ }
+
+ @Test
+ public void testInitVirtualColumn() throws IOException {
+ long start = System.currentTimeMillis();
+ long minOffset = 5000_000l;
+ int totalDocs = 5_000_000;
+ _readerWriters.add(new VirtualColumnLongValueReaderWriter(totalDocs, false) {
+ @Override
+ public boolean update(int docId, long value, LogEventType eventType) {
+ if (eventType == LogEventType.INSERT) {
+ updateValue(docId, value);
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public boolean update(int docId, int value, LogEventType eventType) {
+ return update(docId, value, eventType);
+ }
+ });
+ _readerWriters.add(new VirtualColumnLongValueReaderWriter(totalDocs, false) {
+ @Override
+ public boolean update(int docId, long value, LogEventType eventType) {
+ if (eventType == LogEventType.DELETE) {
+ updateValue(docId, value);
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public boolean update(int docId, int value, LogEventType eventType) {
+ return update(docId, value, eventType);
+ }
+ });
+ int[] offsetToDocId = new int[totalDocs];
+ for (int i = 0; i < totalDocs; i++) {
+ offsetToDocId[i] = i;
+ }
+ List<UpdateLogEntry> updateLogEntries = new ArrayList<>(totalDocs * 2);
+ for (int i = 0; i < totalDocs; i++) {
+ updateLogEntries.add(new UpdateLogEntry(minOffset + i, 50, LogEventType.INSERT, i%8));
+ updateLogEntries.add(new UpdateLogEntry(minOffset + i, 100, LogEventType.DELETE, i%8));
+ }
+ when(_mockProvider.getAllMessages(anyString(), anyString())).thenReturn(updateLogEntries);
+ System.out.println("run time for set up: " + (System.currentTimeMillis() - start));
+
+ start = System.currentTimeMillis();
+
+ ImmutableUpsertSegmentImpl immutableUpsertSegment = new ImmutableUpsertSegmentImpl(_readerWriters, totalDocs,
+ _mockUpsertWaterMarkManager, _mockProvider, minOffset, offsetToDocId);
+
+ immutableUpsertSegment.initVirtualColumn();
+ long runtime = System.currentTimeMillis() - start;
+ System.out.println("run time is " + runtime);
+ Assert.assertTrue(runtime < 1_000L, "run time should be less than 1 second");
+
+ VirtualColumnLongValueReaderWriter insertReaderWrite = _readerWriters.get(0);
+ VirtualColumnLongValueReaderWriter deleteReaderWrite = _readerWriters.get(1);
+ for (int i = 0; i < totalDocs; i++) {
+ if (insertReaderWrite.getLong(i) != 50 || deleteReaderWrite.getLong(i) != 100) {
+ System.out.println(String.format("position %d has value %d/%d", i, insertReaderWrite.getLong(i),
+ deleteReaderWrite.getLong(i)));
+ Assert.fail("no correct value");
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/pinot-distribution/pom.xml b/pinot-distribution/pom.xml
index ebf0146..f55e1a7 100644
--- a/pinot-distribution/pom.xml
+++ b/pinot-distribution/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-distribution</artifactId>
<name>Pinot Distribution</name>
diff --git a/pinot-grigio/pinot-grigio-common/pom.xml b/pinot-grigio/pinot-grigio-common/pom.xml
index 42e07c0..0c98e85 100644
--- a/pinot-grigio/pinot-grigio-common/pom.xml
+++ b/pinot-grigio/pinot-grigio-common/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/metrics/GrigioMeter.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/metrics/GrigioMeter.java
index ab04b6e..8a738da 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/metrics/GrigioMeter.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/metrics/GrigioMeter.java
@@ -29,7 +29,6 @@ public enum GrigioMeter implements AbstractMetrics.Meter {
// segment updater metrics
MESSAGE_FETCH_PER_ROUND("messages", MetricsType.SERVER_ONLY),
- VERSION_LOWER_THAN_CURRENT("messages", MetricsType.SERVER_ONLY),
// key coordinator related metrics
MESSAGE_PRODUCE_FAILED_COUNT("message", MetricsType.KC_ONLY),
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProvider.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProvider.java
index 2972473..02f2fba 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProvider.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProvider.java
@@ -58,6 +58,7 @@ public class SegmentUpdateLogStorageProvider {
}
public synchronized List<UpdateLogEntry> readAllMessagesFromFile() throws IOException {
+ long start = System.currentTimeMillis();
int insertMessageCount = 0;
int deleteMessageCount = 0;
int fileLength = (int) _file.length();
@@ -76,7 +77,8 @@ public class SegmentUpdateLogStorageProvider {
logs.add(logEntry);
}
buffer.clear();
- LOGGER.info("loaded {} message from file, {} insert and {} delete", messageCount, insertMessageCount, deleteMessageCount);
+ LOGGER.info("loaded {} message from file, {} insert and {} delete in {} ms", messageCount, insertMessageCount,
+ deleteMessageCount, System.currentTimeMillis() - start);
return logs;
} else {
return ImmutableList.of();
@@ -123,6 +125,7 @@ public class SegmentUpdateLogStorageProvider {
}
private synchronized void readFullyFromBeginning(File segmentUpdateFile, ByteBuffer buffer) throws IOException {
+ long start = System.currentTimeMillis();
FileChannel channel = new RandomAccessFile(segmentUpdateFile, "r").getChannel();
channel.position(0);
long position = 0;
@@ -132,6 +135,8 @@ public class SegmentUpdateLogStorageProvider {
position += byteRead;
} while (byteRead != -1 && buffer.hasRemaining());
buffer.flip();
+ LOGGER.info("read all data from segment update file {} to buffer in {} ms", segmentUpdateFile.getName(),
+ System.currentTimeMillis() - start);
}
/**
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageProvider.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageProvider.java
index 73ba01f..d04b1a2 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageProvider.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageProvider.java
@@ -143,7 +143,6 @@ public class UpdateLogStorageProvider {
}
public List<UpdateLogEntry> getAllMessages(String tableName, String segmentName) throws IOException {
- LOGGER.info("loading all message for table {} segment {}", tableName, segmentName);
if (_virtualColumnStorage.containsKey(tableName)) {
SegmentUpdateLogStorageProvider provider = _virtualColumnStorage.get(tableName).get(segmentName);
if (provider != null) {
diff --git a/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProviderTest.java b/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProviderTest.java
index 69d30bc..64d4d3b 100644
--- a/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProviderTest.java
+++ b/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProviderTest.java
@@ -115,6 +115,21 @@ public class SegmentUpdateLogStorageProviderTest {
Assert.assertEquals(updateLogEntries.get(i * inputDataList.size() + 1), inputDataList.get(1));
Assert.assertEquals(updateLogEntries.get(i * inputDataList.size() + 2), inputDataList.get(2));
}
+ }
+ @Test
+ public void testReadMesssagePerf() throws IOException {
+ int totalMessageCount = 5_000_000;
+ List<UpdateLogEntry> inputMessages = new ArrayList<>(totalMessageCount * 2);
+ for (int i = 0; i < totalMessageCount; i++) {
+ inputMessages.add(new UpdateLogEntry(i, 50, LogEventType.INSERT, i%8));
+ inputMessages.add(new UpdateLogEntry(i, 100, LogEventType.DELETE, i%8));
+ }
+ long start = System.currentTimeMillis();
+ provider.addData(inputMessages);
+ System.out.println("write data takes ms: " + (System.currentTimeMillis() - start));
+ start = System.currentTimeMillis();
+ provider.readAllMessagesFromFile();
+ System.out.println("read data takes ms: " + (System.currentTimeMillis() - start));
}
}
\ No newline at end of file
diff --git a/pinot-grigio/pinot-grigio-coordinator/pom.xml b/pinot-grigio/pinot-grigio-coordinator/pom.xml
index f5b9527..7ef66ce 100644
--- a/pinot-grigio/pinot-grigio-coordinator/pom.xml
+++ b/pinot-grigio/pinot-grigio-coordinator/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pinot-grigio-coordinator</artifactId>
diff --git a/pinot-grigio/pom.xml b/pinot-grigio/pom.xml
index 3cdad0c..2d3a6a7 100644
--- a/pinot-grigio/pom.xml
+++ b/pinot-grigio/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pinot-grigio</artifactId>
diff --git a/pinot-hadoop-filesystem/pom.xml b/pinot-hadoop-filesystem/pom.xml
index 49b0a8a..0e0f03d 100644
--- a/pinot-hadoop-filesystem/pom.xml
+++ b/pinot-hadoop-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-hadoop-filesystem</artifactId>
<name>Pinot Hadoop Filesystem</name>
diff --git a/pinot-hadoop/pom.xml b/pinot-hadoop/pom.xml
index 928a11b..13691d8 100644
--- a/pinot-hadoop/pom.xml
+++ b/pinot-hadoop/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-hadoop</artifactId>
<name>Pinot Hadoop</name>
diff --git a/pinot-integration-tests/pom.xml b/pinot-integration-tests/pom.xml
index 9d7cb12..8ea6a8d 100644
--- a/pinot-integration-tests/pom.xml
+++ b/pinot-integration-tests/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-integration-tests</artifactId>
<name>Pinot Integration Tests</name>
diff --git a/pinot-minion/pom.xml b/pinot-minion/pom.xml
index c057094..72b1d85 100644
--- a/pinot-minion/pom.xml
+++ b/pinot-minion/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-minion</artifactId>
<name>Pinot Minion</name>
diff --git a/pinot-orc/pom.xml b/pinot-orc/pom.xml
index 34a8740..9508747 100644
--- a/pinot-orc/pom.xml
+++ b/pinot-orc/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-parquet/pom.xml b/pinot-parquet/pom.xml
index 2d7dbb0..577e283 100644
--- a/pinot-parquet/pom.xml
+++ b/pinot-parquet/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-perf/pom.xml b/pinot-perf/pom.xml
index 060ebaf..704f659 100644
--- a/pinot-perf/pom.xml
+++ b/pinot-perf/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-perf</artifactId>
<name>Pinot Perf</name>
diff --git a/pinot-server/pom.xml b/pinot-server/pom.xml
index ac77f20..0503db3 100644
--- a/pinot-server/pom.xml
+++ b/pinot-server/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-server</artifactId>
<name>Pinot Server</name>
diff --git a/pinot-tools/pom.xml b/pinot-tools/pom.xml
index 4b4e686..383cc1f 100644
--- a/pinot-tools/pom.xml
+++ b/pinot-tools/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-tools</artifactId>
<name>Pinot Tools</name>
diff --git a/pinot-transport/pom.xml b/pinot-transport/pom.xml
index b977921..5359b71 100644
--- a/pinot-transport/pom.xml
+++ b/pinot-transport/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
</parent>
<artifactId>pinot-transport</artifactId>
<name>Pinot Transport</name>
diff --git a/pom.xml b/pom.xml
index 77e8647..66af82b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -30,7 +30,7 @@
<groupId>org.apache.pinot</groupId>
<artifactId>pinot</artifactId>
- <version>0.2.2.5.49-SNAPSHOT</version>
+ <version>0.2.2.5.57-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Pinot</name>
<description>A realtime distributed OLAP datastore</description>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 09/12: Fix FixedPartitionCountBytesPartitionerTest
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 47081392f736daca45dfb9bf857fb9dad804b99d
Author: Bo Zhang <bz...@uber.com>
AuthorDate: Mon Nov 4 13:22:11 2019 -0800
Fix FixedPartitionCountBytesPartitionerTest
Reviewers: #streaming_pinot, sjames
Reviewed By: #streaming_pinot, sjames
Differential Revision: https://code.uberinternal.com/D3546187
---
.../pinot/grigio/common/FixedPartitionCountBytesPartitionerTest.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountBytesPartitionerTest.java b/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountBytesPartitionerTest.java
index 9a4b818..567f5df 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountBytesPartitionerTest.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/FixedPartitionCountBytesPartitionerTest.java
@@ -67,6 +67,9 @@ public class FixedPartitionCountBytesPartitionerTest {
Cluster cluster = mock(Cluster.class);
when(cluster.partitionCountForTopic(topic)).thenReturn(2);
- partitioner.partition(topic, null, null, null, null, cluster);
+ String key = "test-key";
+ byte[] keyBytes = key.getBytes();
+
+ partitioner.partition(topic, key, keyBytes, null, null, cluster);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 02/12: [maven-release-plugin] prepare for next
development iteration
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit df8e3a8e5f536e431c05b0931c85eb0050d4b93a
Author: Uber Jenkins <pa...@uber.com>
AuthorDate: Wed Oct 16 00:30:37 2019 +0000
[maven-release-plugin] prepare for next development iteration
---
pinot-api/pom.xml | 2 +-
pinot-azure-filesystem/pom.xml | 2 +-
pinot-broker/pom.xml | 2 +-
pinot-common/pom.xml | 2 +-
pinot-connectors/pinot-connector-kafka-0.11/pom.xml | 2 +-
pinot-connectors/pinot-connector-kafka-0.9/pom.xml | 2 +-
pinot-connectors/pom.xml | 2 +-
pinot-controller/pom.xml | 2 +-
pinot-core/pom.xml | 2 +-
pinot-distribution/pom.xml | 2 +-
pinot-grigio/pinot-grigio-common/pom.xml | 2 +-
pinot-grigio/pinot-grigio-coordinator/pom.xml | 2 +-
pinot-grigio/pom.xml | 2 +-
pinot-hadoop-filesystem/pom.xml | 2 +-
pinot-hadoop/pom.xml | 2 +-
pinot-integration-tests/pom.xml | 2 +-
pinot-minion/pom.xml | 2 +-
pinot-orc/pom.xml | 2 +-
pinot-parquet/pom.xml | 2 +-
pinot-perf/pom.xml | 2 +-
pinot-server/pom.xml | 4 ++--
pinot-tools/pom.xml | 2 +-
pinot-transport/pom.xml | 2 +-
pom.xml | 4 ++--
24 files changed, 26 insertions(+), 26 deletions(-)
diff --git a/pinot-api/pom.xml b/pinot-api/pom.xml
index 8dac9f6..f26d749 100644
--- a/pinot-api/pom.xml
+++ b/pinot-api/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-api</artifactId>
<name>Pinot API</name>
diff --git a/pinot-azure-filesystem/pom.xml b/pinot-azure-filesystem/pom.xml
index b01f8e4..7f3e432 100644
--- a/pinot-azure-filesystem/pom.xml
+++ b/pinot-azure-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-azure-filesystem</artifactId>
<name>Pinot Azure Filesystem</name>
diff --git a/pinot-broker/pom.xml b/pinot-broker/pom.xml
index d78368a..36d6eb1 100644
--- a/pinot-broker/pom.xml
+++ b/pinot-broker/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-broker</artifactId>
<name>Pinot Broker</name>
diff --git a/pinot-common/pom.xml b/pinot-common/pom.xml
index 9886aa5..1939448 100644
--- a/pinot-common/pom.xml
+++ b/pinot-common/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-common</artifactId>
<name>Pinot Common</name>
diff --git a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
index cac22df..2e9b3f6 100644
--- a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
index 7218380..f63166c 100644
--- a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pom.xml b/pinot-connectors/pom.xml
index 3bd9cc2..4784801 100644
--- a/pinot-connectors/pom.xml
+++ b/pinot-connectors/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-connectors</artifactId>
diff --git a/pinot-controller/pom.xml b/pinot-controller/pom.xml
index 68358c0..9a326a3 100644
--- a/pinot-controller/pom.xml
+++ b/pinot-controller/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-controller</artifactId>
<name>Pinot Controller</name>
diff --git a/pinot-core/pom.xml b/pinot-core/pom.xml
index 0d1ae93..b949eaf 100644
--- a/pinot-core/pom.xml
+++ b/pinot-core/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-core</artifactId>
<name>Pinot Core</name>
diff --git a/pinot-distribution/pom.xml b/pinot-distribution/pom.xml
index 87d674b..49e3d9c 100644
--- a/pinot-distribution/pom.xml
+++ b/pinot-distribution/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-distribution</artifactId>
<name>Pinot Distribution</name>
diff --git a/pinot-grigio/pinot-grigio-common/pom.xml b/pinot-grigio/pinot-grigio-common/pom.xml
index 107853e..82f3570 100644
--- a/pinot-grigio/pinot-grigio-common/pom.xml
+++ b/pinot-grigio/pinot-grigio-common/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-grigio/pinot-grigio-coordinator/pom.xml b/pinot-grigio/pinot-grigio-coordinator/pom.xml
index 26d4f90..3857d92 100644
--- a/pinot-grigio/pinot-grigio-coordinator/pom.xml
+++ b/pinot-grigio/pinot-grigio-coordinator/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-grigio/pom.xml b/pinot-grigio/pom.xml
index 874606e..5ac6c5a 100644
--- a/pinot-grigio/pom.xml
+++ b/pinot-grigio/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pinot-grigio</artifactId>
diff --git a/pinot-hadoop-filesystem/pom.xml b/pinot-hadoop-filesystem/pom.xml
index ff45d60..e130a69 100644
--- a/pinot-hadoop-filesystem/pom.xml
+++ b/pinot-hadoop-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-hadoop-filesystem</artifactId>
<name>Pinot Hadoop Filesystem</name>
diff --git a/pinot-hadoop/pom.xml b/pinot-hadoop/pom.xml
index 8503e22..2be8498 100644
--- a/pinot-hadoop/pom.xml
+++ b/pinot-hadoop/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-hadoop</artifactId>
<name>Pinot Hadoop</name>
diff --git a/pinot-integration-tests/pom.xml b/pinot-integration-tests/pom.xml
index c87e0e7..46ac371 100644
--- a/pinot-integration-tests/pom.xml
+++ b/pinot-integration-tests/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-integration-tests</artifactId>
<name>Pinot Integration Tests</name>
diff --git a/pinot-minion/pom.xml b/pinot-minion/pom.xml
index e7b0f64..4d37500 100644
--- a/pinot-minion/pom.xml
+++ b/pinot-minion/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-minion</artifactId>
<name>Pinot Minion</name>
diff --git a/pinot-orc/pom.xml b/pinot-orc/pom.xml
index 08ba6a7..2e8672a 100644
--- a/pinot-orc/pom.xml
+++ b/pinot-orc/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-parquet/pom.xml b/pinot-parquet/pom.xml
index e44ae00..690924a 100644
--- a/pinot-parquet/pom.xml
+++ b/pinot-parquet/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-perf/pom.xml b/pinot-perf/pom.xml
index 43ffc02..113d974 100644
--- a/pinot-perf/pom.xml
+++ b/pinot-perf/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-perf</artifactId>
<name>Pinot Perf</name>
diff --git a/pinot-server/pom.xml b/pinot-server/pom.xml
index 2ae24d7..c5d394b 100644
--- a/pinot-server/pom.xml
+++ b/pinot-server/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-server</artifactId>
<name>Pinot Server</name>
@@ -68,7 +68,7 @@
<dependency>
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-grigio-common</artifactId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
diff --git a/pinot-tools/pom.xml b/pinot-tools/pom.xml
index e49075e..4078c18 100644
--- a/pinot-tools/pom.xml
+++ b/pinot-tools/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-tools</artifactId>
<name>Pinot Tools</name>
diff --git a/pinot-transport/pom.xml b/pinot-transport/pom.xml
index 14954aa..7b31dd0 100644
--- a/pinot-transport/pom.xml
+++ b/pinot-transport/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
</parent>
<artifactId>pinot-transport</artifactId>
<name>Pinot Transport</name>
diff --git a/pom.xml b/pom.xml
index 5c1e255..5e4d1d0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -30,7 +30,7 @@
<groupId>org.apache.pinot</groupId>
<artifactId>pinot</artifactId>
- <version>0.2.2.5.43</version>
+ <version>0.2.2.5.44-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Pinot</name>
<description>A realtime distributed OLAP datastore</description>
@@ -96,7 +96,7 @@
<scm>
<developerConnection>scm:git:gitolite@code.uber.internal:data/pinot</developerConnection>
- <tag>pinot-0.2.2.5.43</tag>
+ <tag>pinot-0.1.076-SNAPSHOT</tag>
</scm>
<distributionManagement>
<repository>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 05/12: fix issue in retention management and
refactor
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 9ac12f38bcb30eed228737a1ae85445eac8a4d18
Author: james Shao <sj...@uber.com>
AuthorDate: Wed Oct 23 16:50:35 2019 -0700
fix issue in retention management and refactor
Summary:
right now if we delete and recreate the table, the seq number of segments got reset to start from 0.
this cause issue in newer segment are considered as older than the last segment we have seen.
this diff would solve such issue
we also refactor a bunch of stuff to base class so the kc retention
manager and server retention manager are mostly using the same logics
Reviewers: tingchen, csoman, bzzhang
Reviewed By: bzzhang
Differential Revision: https://code.uberinternal.com/D3528221
---
pinot-grigio/pinot-grigio-common/pom.xml | 5 +
.../UpdateLogRetentionManagerImpl.java | 2 +-
.../UpdateLogTableRetentionManager.java | 98 ++++++++++----
.../UpdateLogTableRetentionManagerImpl.java | 79 +----------
.../grigio}/common/utils/IdealStateHelper.java | 2 +-
.../UpdateLogTableRetentionManagerImplTest.java | 119 +++++++++++++++++
.../grigio/common/utils/IdealStateHelperTest.java | 74 +++++++++++
pinot-grigio/pinot-grigio-coordinator/pom.xml | 5 +
.../KCUpdateLogRetentionManagerImpl.java | 2 +-
.../KCUpdateLogTableRetentionManagerImpl.java | 58 ++------
.../starter/KeyCoordinatorStarter.java | 2 +-
.../KCUpdateLogRetentionManagerImplTest.java | 114 ++++++++++++++++
.../KCUpdateLogTableRetentionManagerImplTest.java | 148 +++++++++++++++++++++
.../pinot/server/starter/ServerInstance.java | 2 +-
14 files changed, 560 insertions(+), 150 deletions(-)
diff --git a/pinot-grigio/pinot-grigio-common/pom.xml b/pinot-grigio/pinot-grigio-common/pom.xml
index 251cb5a..53adbbf 100644
--- a/pinot-grigio/pinot-grigio-common/pom.xml
+++ b/pinot-grigio/pinot-grigio-common/pom.xml
@@ -60,6 +60,11 @@
<artifactId>guava</artifactId>
</dependency>
<dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.testng</groupId>
<artifactId>testng</artifactId>
<scope>test</scope>
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogRetentionManagerImpl.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogRetentionManagerImpl.java
index 6699a64..3852f99 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogRetentionManagerImpl.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogRetentionManagerImpl.java
@@ -18,7 +18,7 @@
*/
package org.apache.pinot.grigio.common.storageProvider.retentionManager;
-import org.apache.pinot.common.utils.IdealStateHelper;
+import org.apache.pinot.grigio.common.utils.IdealStateHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManager.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManager.java
index afb42bc..62215b8 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManager.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManager.java
@@ -20,54 +20,99 @@ package org.apache.pinot.grigio.common.storageProvider.retentionManager;
import com.google.common.collect.ImmutableMap;
import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.grigio.common.utils.IdealStateHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
-import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
/**
* class to help decide whether to keep update log for a particular segment or not
*/
-public interface UpdateLogTableRetentionManager {
+public abstract class UpdateLogTableRetentionManager {
- /**
- * decide whether we should ingest the update log for a given segment name
- * @param segmentName
- * @return true if we should keep the update log for a particular segment, false otherwise
- */
- boolean shouldIngestForSegment(String segmentName);
+ protected static final Logger LOGGER = LoggerFactory.getLogger(UpdateLogTableRetentionManagerImpl.class);
+
+ protected String _tableName;
+ protected IdealStateHelper _idealStateHelper;
+ // use the concurrent hashmap as concurrent hashset
+ protected Map<Integer, LLCSegmentName> _partitionToLastSegment = new ConcurrentHashMap<>();
+ protected Map<String, String> _blacklistedSegments = new ConcurrentHashMap<>();
+ protected Map<String, Map<String, String>> _segmentsToInstanceMap;
+
+ public UpdateLogTableRetentionManager(IdealStateHelper idealStateHelper, String tableName) {
+ _idealStateHelper = idealStateHelper;
+ _tableName = tableName;
+ }
/**
* for external components to notify retention manager that physical data for this segment has been deleted
* @param segmentName the name of the segment that we are deleting from local storage
*/
- void notifySegmentDeletion(String segmentName);
+ public abstract void notifySegmentDeletion(String segmentName);
/**
* for external components to notify us that segments for this tables has been change and we should refresh it
*/
- void notifySegmentsChange();
+ public abstract void notifySegmentsChange();
+
+ /**
+ * check if the the given segment is assigned to current server
+ */
+ protected abstract boolean isSegmentAssignedToCurrentServer(String segmentName);
/**
- * get the mapping between the partition to the last segment (based on timestamp)
- * @param segments the list of segment we are comparing
- * @return the mapping between the partition id and the last segment of that partition
+ * update state from helix
+ */
+ protected void updateStateFromHelix() {
+ long start = System.currentTimeMillis();
+ _segmentsToInstanceMap = ImmutableMap.copyOf(_idealStateHelper.getSegmentToInstanceMap(_tableName));
+ if (_segmentsToInstanceMap.size() == 0) {
+ LOGGER.error("failed to get any segment for the current table {}", _tableName);
+ }
+ Map<Integer, LLCSegmentName> partitionLastSeg = new HashMap<>();
+ for (String segmentStr: _segmentsToInstanceMap.keySet()) {
+ LLCSegmentName llcSegmentName = new LLCSegmentName(segmentStr);
+ int partition = llcSegmentName.getPartitionId();
+ if (!partitionLastSeg.containsKey(partition)) {
+ partitionLastSeg.put(partition, llcSegmentName);
+ } else if (partitionLastSeg.get(partition).getSequenceNumber() < llcSegmentName.getSequenceNumber()) {
+ partitionLastSeg.put(partition, llcSegmentName);
+ }
+ }
+ _partitionToLastSegment = ImmutableMap.copyOf(partitionLastSeg);
+ LOGGER.info("updated table {} state from helix in {} ms", _tableName, System.currentTimeMillis() - start);
+ }
+
+ /**
+ * decide whether we should ingest the update log for a given segment name
+ * @param segmentName
+ * @return true if we should keep the update log for a particular segment, false otherwise
*/
- static Map<Integer, LLCSegmentName> getPartitionToLastSegment(Set<String> segments) {
- Map<Integer, LLCSegmentName> partitionToLastSegment = new HashMap<>();
- for (String segment : segments) {
- LLCSegmentName llcSegmentName = new LLCSegmentName(segment);
+ public boolean shouldIngestForSegment(String segmentName) {
+ if (_segmentsToInstanceMap.containsKey(segmentName)) {
+ return isSegmentAssignedToCurrentServer(segmentName);
+ } else if (_blacklistedSegments.containsKey(segmentName)) {
+ return false;
+ } else {
+ LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
int partition = llcSegmentName.getPartitionId();
- if (!partitionToLastSegment.containsKey(partition)) {
- partitionToLastSegment.put(partition, llcSegmentName);
- } else {
- LLCSegmentName currLastSegment = partitionToLastSegment.get(partition);
- if (llcSegmentName.getCreationTimeStamp() > currLastSegment.getCreationTimeStamp()) {
- partitionToLastSegment.put(partition, llcSegmentName);
+ if (!_partitionToLastSegment.containsKey(llcSegmentName.getPartitionId())
+ || compareSegment(llcSegmentName, _partitionToLastSegment.get(partition))) {
+ // assume our idealState is out of date
+ updateStateFromHelix();
+ if (_segmentsToInstanceMap.containsKey(segmentName) && isSegmentAssignedToCurrentServer(segmentName)) {
+ LOGGER.info("segment {} matched in ideal state after refresh", segmentName);
+ return true;
}
}
+ // we most probably got a segment that is from a deleted table or segment assigned to another table
+ // assume we don't do re-balance, we won't do refresh
+ _blacklistedSegments.put(segmentName, segmentName);
+ return false;
}
- return ImmutableMap.copyOf(partitionToLastSegment);
}
/**
@@ -76,9 +121,8 @@ public interface UpdateLogTableRetentionManager {
* @param segment2
* @return true if the segment1 is a "newer" segment
*/
- static boolean compareSegment(LLCSegmentName segment1, LLCSegmentName segment2) {
- return segment1.getSequenceNumber() > segment2.getSequenceNumber() &&
- segment1.getCreationTimeStamp() > segment2.getCreationTimeStamp();
+ protected boolean compareSegment(LLCSegmentName segment1, LLCSegmentName segment2) {
+ return segment1.getCreationTimeStamp() > segment2.getCreationTimeStamp();
}
}
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManagerImpl.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManagerImpl.java
index f145eb5..91c098a 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManagerImpl.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManagerImpl.java
@@ -18,88 +18,20 @@
*/
package org.apache.pinot.grigio.common.storageProvider.retentionManager;
-import com.google.common.collect.ImmutableMap;
-import org.apache.pinot.common.utils.IdealStateHelper;
-import org.apache.pinot.common.utils.LLCSegmentName;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.pinot.grigio.common.utils.IdealStateHelper;
import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-public class UpdateLogTableRetentionManagerImpl implements UpdateLogTableRetentionManager {
+public class UpdateLogTableRetentionManagerImpl extends UpdateLogTableRetentionManager {
- private static final Logger LOGGER = LoggerFactory.getLogger(UpdateLogTableRetentionManagerImpl.class);
-
- protected IdealStateHelper _idealStateHelper;
- protected String _tableName;
-
- private Map<Integer, LLCSegmentName> _partitionLastSeg = new ConcurrentHashMap<>();
- private Map<String, Map<String, String>> _segmentsToInstanceMap;
- // use the concurrent hashmap as concurrent hashset
- private Map<String, String> _blacklistedSegmentNames = new ConcurrentHashMap<>();
- private String _instanceId;
+ protected String _instanceId;
public UpdateLogTableRetentionManagerImpl(IdealStateHelper idealStateHelper, String tableName, String instanceId) {
- _idealStateHelper = idealStateHelper;
- _tableName = tableName;
+ super(idealStateHelper, tableName);
_instanceId = instanceId;
updateStateFromHelix();
}
- private void updateStateFromHelix() {
- long start = System.currentTimeMillis();
- _segmentsToInstanceMap = ImmutableMap.copyOf(_idealStateHelper.getSegmentToInstanceMap(_tableName));
- if (_segmentsToInstanceMap.size() == 0) {
- LOGGER.error("failed to get any segment for the current table {}", _tableName);
- }
- Map<Integer, LLCSegmentName> partitionLastSeg = new HashMap<>();
- for (String segmentStr: _segmentsToInstanceMap.keySet()) {
- LLCSegmentName llcSegmentName = new LLCSegmentName(segmentStr);
- int partition = llcSegmentName.getPartitionId();
- if (!partitionLastSeg.containsKey(partition)) {
- partitionLastSeg.put(partition, llcSegmentName);
- } else if (partitionLastSeg.get(partition).getSequenceNumber() < llcSegmentName.getSequenceNumber()) {
- partitionLastSeg.put(partition, llcSegmentName);
- }
- }
- _partitionLastSeg = ImmutableMap.copyOf(partitionLastSeg);
- LOGGER.info("updated table {} state from helix in {} ms", _tableName, System.currentTimeMillis() - start);
- }
-
- @Override
- public synchronized boolean shouldIngestForSegment(String segmentName) {
- if (_segmentsToInstanceMap.containsKey(segmentName)) {
- return isSegmentAssignedToCurrentServer(segmentName);
- } else if (_blacklistedSegmentNames.containsKey(segmentName)) {
- return false;
- } else {
- LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
- int partitionId = llcSegmentName.getPartitionId();
- int seq = llcSegmentName.getSequenceNumber();
- long ts = llcSegmentName.getCreationTimeStamp();
- if (_partitionLastSeg.containsKey(partitionId)) {
- LLCSegmentName latestSegment = _partitionLastSeg.get(partitionId);
- if (seq > latestSegment.getSequenceNumber() && ts > latestSegment.getCreationTimeStamp()) {
- // assume our idealState is out of date
- updateStateFromHelix();
- if (isSegmentAssignedToCurrentServer(segmentName)) {
- LOGGER.info("segment {} matched in ideal state after refresh", segmentName);
- return true;
- }
- } else {
- // we most probably got a segment that is from a deleted table or segment assigned to another table
- // assume we don't do re-balance, we won't do refresh
- LOGGER.info("adding segment {} to blacklist segment list", segmentName);
- _blacklistedSegmentNames.put(segmentName, segmentName);
- }
- }
- }
- return false;
- }
-
@Override
public synchronized void notifySegmentDeletion(String segmentName) {
LOGGER.info("handling segment deletion in retention manager");
@@ -115,7 +47,8 @@ public class UpdateLogTableRetentionManagerImpl implements UpdateLogTableRetenti
updateStateFromHelix();
}
- private boolean isSegmentAssignedToCurrentServer(String segmentName) {
+ @Override
+ protected boolean isSegmentAssignedToCurrentServer(String segmentName) {
return _segmentsToInstanceMap.containsKey(segmentName)
&& _segmentsToInstanceMap.get(segmentName).containsKey(_instanceId)
&& !"DROPPED".equals(_segmentsToInstanceMap.get(segmentName).get(_instanceId));
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/IdealStateHelper.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/utils/IdealStateHelper.java
similarity index 98%
rename from pinot-common/src/main/java/org/apache/pinot/common/utils/IdealStateHelper.java
rename to pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/utils/IdealStateHelper.java
index 9ebc72d..b0c193f 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/IdealStateHelper.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/utils/IdealStateHelper.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pinot.common.utils;
+package org.apache.pinot.grigio.common.utils;
import org.apache.helix.HelixAdmin;
import org.apache.helix.model.IdealState;
diff --git a/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManagerImplTest.java b/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManagerImplTest.java
new file mode 100644
index 0000000..83fba68
--- /dev/null
+++ b/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/UpdateLogTableRetentionManagerImplTest.java
@@ -0,0 +1,119 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.grigio.common.storageProvider.retentionManager;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.pinot.grigio.common.utils.IdealStateHelper;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.internal.verification.VerificationModeFactory.times;
+
+public class UpdateLogTableRetentionManagerImplTest {
+
+ private IdealStateHelper mockIdealStateHelper;
+ private Map<String, Map<String, String>> segmentsInstanceMap;
+ private UpdateLogTableRetentionManagerImpl updateLogTableRetentionManager;
+
+ @BeforeMethod
+ public void init() {
+ segmentsInstanceMap = new HashMap<>();
+
+ segmentsInstanceMap.put("table__0__10__20191027T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server2", "ONLINE"));
+ segmentsInstanceMap.put("table__0__11__20191028T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server3", "ONLINE"));
+ segmentsInstanceMap.put("table__1__10__20191027T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server2", "ONLINE"));
+ segmentsInstanceMap.put("table__2__10__20191027T2041Z",
+ ImmutableMap.of("server2", "ONLINE", "server3", "ONLINE"));
+
+ mockIdealStateHelper = mock(IdealStateHelper.class);
+ when(mockIdealStateHelper.getSegmentToInstanceMap("table")).thenReturn(segmentsInstanceMap);
+
+ updateLogTableRetentionManager = new UpdateLogTableRetentionManagerImpl(mockIdealStateHelper, "table",
+ "server1");
+ }
+
+ @Test
+ public void testShouldIngestForSegment() {
+ // test the happy paths
+ Assert.assertTrue(updateLogTableRetentionManager.shouldIngestForSegment("table__0__10__20191027T2041Z"));
+ Assert.assertTrue(updateLogTableRetentionManager.shouldIngestForSegment("table__1__10__20191027T2041Z"));
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+
+ // test for segments that has lower seq
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__1__5__20191027T2041Z"));
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+
+ //test for segment that are older and not in existing segment list
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__1__20__20191027T2041Z"));
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+
+ // test for partition that has no assignment on host
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__2__10__20191027T2041Z"));
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+
+ //test for older timestamp but higher seq
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__0__20__20191027T2041Z"));
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+
+ // test for newer segments, and there is update in newer ideal state
+ segmentsInstanceMap.put("table__0__12__20191029T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server3", "ONLINE"));
+ Assert.assertTrue(updateLogTableRetentionManager.shouldIngestForSegment("table__0__12__20191029T2041Z"));
+ verify(mockIdealStateHelper, times(2)).getSegmentToInstanceMap("table");
+
+ // test for newer segments and there is no update in newer ideal state
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__0__13__20191029T2141Z"));
+ verify(mockIdealStateHelper, times(3)).getSegmentToInstanceMap("table");
+
+ // multiple attempts trying to fetch for segment should not recheck the ideal state often
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__0__13__20191029T2141Z"));
+ verify(mockIdealStateHelper, times(3)).getSegmentToInstanceMap("table");
+
+ // for unknown partition, we should also refresh data
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__2__13__20191029T2141Z"));
+ verify(mockIdealStateHelper, times(4)).getSegmentToInstanceMap("table");
+
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__3__13__20191029T2141Z"));
+ verify(mockIdealStateHelper, times(5)).getSegmentToInstanceMap("table");
+ }
+
+ @Test
+ public void testNotifySegmentDeletion() {
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+ updateLogTableRetentionManager.notifySegmentDeletion("table__0__10__20191027T2041Z");
+ verify(mockIdealStateHelper, times(2)).getSegmentToInstanceMap("table");
+ }
+
+ @Test
+ public void testNotifySegmentsChange() {
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+ updateLogTableRetentionManager.notifySegmentDeletion("table__0__10__20191027T2041Z");
+ verify(mockIdealStateHelper, times(2)).getSegmentToInstanceMap("table");
+ }
+}
\ No newline at end of file
diff --git a/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/utils/IdealStateHelperTest.java b/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/utils/IdealStateHelperTest.java
new file mode 100644
index 0000000..4245e5b
--- /dev/null
+++ b/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/utils/IdealStateHelperTest.java
@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.grigio.common.utils;
+
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.model.IdealState;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import java.util.Map;
+import java.util.Set;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class IdealStateHelperTest {
+
+ private HelixAdmin mockAmin;
+ private IdealStateHelper idealStateHelper;
+ private IdealState mockIdealState;
+
+ @BeforeClass
+ public void init() {
+ mockIdealState = new IdealState("resource");
+ mockIdealState.enable(true);
+ mockIdealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
+ mockIdealState.setPartitionState("seg1", "host1", "ONLINE");
+ mockIdealState.setPartitionState("seg1", "host2", "ONLINE");
+ mockIdealState.setPartitionState("seg2", "host2", "ONLINE");
+ mockIdealState.setPartitionState("seg2", "host3", "OFFLINE");
+
+ mockAmin = mock(HelixAdmin.class);
+ when(mockAmin.getResourceIdealState(anyString(), anyString())).thenReturn(mockIdealState);
+ idealStateHelper = new IdealStateHelper(mockAmin, "clusterName");
+ }
+
+ @Test
+ public void testGetAllSegmentsForTable() {
+ Set<String> segments = idealStateHelper.getAllSegmentsForTable("table");
+ Assert.assertEquals(segments.size(), 2);
+ Assert.assertTrue(segments.contains("seg1"));
+ Assert.assertTrue(segments.contains("seg2"));
+ }
+
+ @Test
+ public void testGetSegmentToInstanceMap() {
+ Map<String, Map<String, String>> resultMap = idealStateHelper.getSegmentToInstanceMap("table");
+ Assert.assertEquals(resultMap.size(), 2);
+ Assert.assertEquals(resultMap.get("seg1").size(), 2);
+ Assert.assertEquals(resultMap.get("seg1").get("host1"), "ONLINE");
+ Assert.assertEquals(resultMap.get("seg1").get("host2"), "ONLINE");
+ Assert.assertEquals(resultMap.get("seg2").size(), 2);
+ Assert.assertEquals(resultMap.get("seg2").get("host2"), "ONLINE");
+ Assert.assertEquals(resultMap.get("seg2").get("host3"), "OFFLINE");
+ }
+}
\ No newline at end of file
diff --git a/pinot-grigio/pinot-grigio-coordinator/pom.xml b/pinot-grigio/pinot-grigio-coordinator/pom.xml
index f8f6e48..a7d6db2 100644
--- a/pinot-grigio/pinot-grigio-coordinator/pom.xml
+++ b/pinot-grigio/pinot-grigio-coordinator/pom.xml
@@ -89,6 +89,11 @@
<version>${project.version}</version>
</dependency>
<dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.testng</groupId>
<artifactId>testng</artifactId>
<scope>test</scope>
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogRetentionManagerImpl.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogRetentionManagerImpl.java
index ef13300..8c846b3 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogRetentionManagerImpl.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogRetentionManagerImpl.java
@@ -22,7 +22,7 @@ import org.apache.helix.NotificationContext;
import org.apache.helix.api.listeners.IdealStateChangeListener;
import org.apache.helix.model.IdealState;
import org.apache.pinot.common.Utils;
-import org.apache.pinot.common.utils.IdealStateHelper;
+import org.apache.pinot.grigio.common.utils.IdealStateHelper;
import org.apache.pinot.grigio.common.storageProvider.UpdateLogStorageProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogTableRetentionManagerImpl.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogTableRetentionManagerImpl.java
index cfb2f27..8e89fb9 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogTableRetentionManagerImpl.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogTableRetentionManagerImpl.java
@@ -18,40 +18,31 @@
*/
package org.apache.pinot.grigio.common.storageProvider.retentionManager;
-import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
-import org.apache.pinot.common.utils.IdealStateHelper;
-import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.grigio.common.storageProvider.UpdateLogStorageProvider;
+import org.apache.pinot.grigio.common.utils.IdealStateHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Arrays;
-import java.util.Map;
import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-public class KCUpdateLogTableRetentionManagerImpl implements UpdateLogTableRetentionManager {
+public class KCUpdateLogTableRetentionManagerImpl extends UpdateLogTableRetentionManager {
private static final Logger LOGGER = LoggerFactory.getLogger(KCUpdateLogTableRetentionManagerImpl.class);
private Set<String> _segments;
- private Map<Integer, LLCSegmentName> _partitionToLastSegment;
- private final Map<String, String> _blackListedSegments;
- private final String _tableName;
private final UpdateLogStorageProvider _provider;
- private final IdealStateHelper _idealStateHelper;
public KCUpdateLogTableRetentionManagerImpl(IdealStateHelper idealStateHelper, String tableName, UpdateLogStorageProvider provider) throws IOException {
-
+ super(idealStateHelper, tableName);
_provider = provider;
- _tableName = tableName;
- _idealStateHelper = idealStateHelper;
- _blackListedSegments = new ConcurrentHashMap<>();
+ // load the current update log on this server and match it with helix stored state,
+ // so we can remove any unused update logs
_provider.loadTable(tableName);
_segments = _provider.getAllSegments(tableName);
- _partitionToLastSegment = UpdateLogTableRetentionManager.getPartitionToLastSegment(_segments);
+ updateStateFromHelix();
}
public synchronized void updateSegmentsAndRemoveOldFiles(Set<String> newSegmentSet) {
@@ -69,40 +60,17 @@ public class KCUpdateLogTableRetentionManagerImpl implements UpdateLogTableReten
}
}
- private synchronized void updateSegments(Set<String> newSegmentSet) {
- _segments = newSegmentSet;
- _partitionToLastSegment = UpdateLogTableRetentionManager.getPartitionToLastSegment(_segments);
- }
-
- public Set<String> getAllSegments() {
- return ImmutableSet.copyOf(_segments);
- }
-
@Override
- public synchronized boolean shouldIngestForSegment(String segmentName) {
- if (_segments.contains(segmentName)) {
- return true;
- } else if (_blackListedSegments.containsKey(segmentName)) {
- return false;
- } else {
- LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
- int partition = llcSegmentName.getPartitionId();
- if (!_partitionToLastSegment.containsKey(llcSegmentName.getPartitionId())
- || UpdateLogTableRetentionManager.compareSegment(llcSegmentName, _partitionToLastSegment.get(partition))) {
- updateStateFromHelix();
- if (_segments.contains(segmentName)) {
- LOGGER.info("segment {} matched in ideal state after refresh", segmentName);
- return true;
- }
- }
- _blackListedSegments.put(segmentName, segmentName);
- return false;
- }
+ protected boolean isSegmentAssignedToCurrentServer(String segmentName) {
+ // always return true as key coordinator should store whatever segment sent to it
+ return true;
}
- private void updateStateFromHelix() {
+ @Override
+ protected void updateStateFromHelix() {
long start = System.currentTimeMillis();
- updateSegmentsAndRemoveOldFiles(_idealStateHelper.getSegmentToInstanceMap(_tableName).keySet());
+ super.updateStateFromHelix();
+ updateSegmentsAndRemoveOldFiles(_segmentsToInstanceMap.keySet());
LOGGER.info("updated table {} state from helix in {} ms", _tableName, System.currentTimeMillis() - start);
}
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/starter/KeyCoordinatorStarter.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/starter/KeyCoordinatorStarter.java
index 60012fc..c7d9628 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/starter/KeyCoordinatorStarter.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/keyCoordinator/starter/KeyCoordinatorStarter.java
@@ -24,7 +24,7 @@ import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.StringUtils;
import org.apache.pinot.common.metrics.MetricsHelper;
import org.apache.pinot.common.utils.CommonConstants;
-import org.apache.pinot.common.utils.IdealStateHelper;
+import org.apache.pinot.grigio.common.utils.IdealStateHelper;
import org.apache.pinot.grigio.common.config.CommonConfig;
import org.apache.pinot.grigio.common.rpcQueue.KeyCoordinatorQueueConsumer;
import org.apache.pinot.grigio.common.rpcQueue.LogCoordinatorQueueProducer;
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogRetentionManagerImplTest.java b/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogRetentionManagerImplTest.java
new file mode 100644
index 0000000..9f9f4de
--- /dev/null
+++ b/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogRetentionManagerImplTest.java
@@ -0,0 +1,114 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.grigio.common.storageProvider.retentionManager;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.model.IdealState;
+import org.apache.pinot.grigio.common.storageProvider.UpdateLogStorageProvider;
+import org.apache.pinot.grigio.common.utils.IdealStateHelper;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.internal.verification.VerificationModeFactory.times;
+
+public class KCUpdateLogRetentionManagerImplTest {
+ private KCUpdateLogRetentionManagerImpl kcUpdateLogRetentionManager;
+ private IdealStateHelper mockIdealStateHelper;
+ private UpdateLogStorageProvider mockUpdateLogStorageProvider;
+ private Map<String, Map<String, String>> table1SegmentMap;
+ private Map<String, Map<String, String>> table2SegmentMap;
+
+ @BeforeMethod
+ public void setUp() {
+ table1SegmentMap = new HashMap<>();
+
+ table1SegmentMap.put("table1__0__10__20191027T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server2", "ONLINE"));
+ table1SegmentMap.put("table1__0__11__20191028T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server3", "ONLINE"));
+ table1SegmentMap.put("table1__1__10__20191027T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server2", "ONLINE"));
+ table1SegmentMap.put("table1__2__10__20191027T2041Z",
+ ImmutableMap.of("server2", "ONLINE", "server3", "ONLINE"));
+
+ table2SegmentMap = new HashMap<>();
+ table2SegmentMap.put("table2__0__10__20191027T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server2", "ONLINE"));
+
+ mockIdealStateHelper = mock(IdealStateHelper.class);
+ when(mockIdealStateHelper.getSegmentToInstanceMap("table1")).thenReturn(table1SegmentMap);
+ when(mockIdealStateHelper.getSegmentToInstanceMap("table2")).thenReturn(table1SegmentMap);
+
+ mockUpdateLogStorageProvider = mock(UpdateLogStorageProvider.class);
+ kcUpdateLogRetentionManager = new KCUpdateLogRetentionManagerImpl(mockIdealStateHelper, mockUpdateLogStorageProvider,
+ "kc1");
+ }
+
+ @Test
+ public void testOnIdealStateChange() throws InterruptedException {
+ IdealState table1MockIdealState = getMockIdealState("table1_REALTIME",
+ ImmutableList.of("seg1", "host1"), ImmutableList.of("seg1", "host2"));
+
+ IdealState table2MockIdealState = getMockIdealState("table2_REALTIME",
+ ImmutableList.of("seg1", "host1"), ImmutableList.of("seg1", "host2"));
+
+ UpdateLogTableRetentionManager table1RetentionManager = kcUpdateLogRetentionManager.
+ getRetentionManagerForTable("table1_REALTIME");
+ UpdateLogTableRetentionManager table2RetentionManager = kcUpdateLogRetentionManager.
+ getRetentionManagerForTable("table2_REALTIME");
+
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table1_REALTIME");
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table2_REALTIME");
+
+ kcUpdateLogRetentionManager.onIdealStateChange(ImmutableList.of(table1MockIdealState), mock(NotificationContext.class));
+ verify(mockIdealStateHelper, times(2)).getSegmentToInstanceMap("table1_REALTIME");
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table2_REALTIME");
+
+ // table1 idealstate is the same as last time
+ kcUpdateLogRetentionManager.onIdealStateChange(ImmutableList.of(table1MockIdealState, table2MockIdealState), mock(NotificationContext.class));
+ verify(mockIdealStateHelper, times(2)).getSegmentToInstanceMap("table1_REALTIME");
+ verify(mockIdealStateHelper, times(2)).getSegmentToInstanceMap("table2_REALTIME");
+
+ kcUpdateLogRetentionManager.onIdealStateChange(ImmutableList.of(
+ getMockIdealState("table1_REALTIME", ImmutableList.of("seg1", "host1"), ImmutableList.of("seg2", "host2")),
+ getMockIdealState("table2_REALTIME", ImmutableList.of("seg1", "host1"), ImmutableList.of("seg2", "host2"))
+ ), mock(NotificationContext.class));
+ verify(mockIdealStateHelper, times(3)).getSegmentToInstanceMap("table1_REALTIME");
+ verify(mockIdealStateHelper, times(3)).getSegmentToInstanceMap("table2_REALTIME");
+
+ }
+
+ private IdealState getMockIdealState(String resourceName, List<String>... partitionInstance) {
+ IdealState mockIdealState = new IdealState(resourceName);
+ mockIdealState.enable(true);
+ mockIdealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
+ Arrays.stream(partitionInstance).forEach(partition -> mockIdealState.setPartitionState(partition.get(0), partition.get(1), "ONLINE"));
+ return mockIdealState;
+ }
+}
\ No newline at end of file
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogTableRetentionManagerImplTest.java b/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogTableRetentionManagerImplTest.java
new file mode 100644
index 0000000..c36ff1f
--- /dev/null
+++ b/pinot-grigio/pinot-grigio-coordinator/src/test/java/org/apache/pinot/grigio/common/storageProvider/retentionManager/KCUpdateLogTableRetentionManagerImplTest.java
@@ -0,0 +1,148 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.grigio.common.storageProvider.retentionManager;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.pinot.grigio.common.storageProvider.UpdateLogStorageProvider;
+import org.apache.pinot.grigio.common.utils.IdealStateHelper;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class KCUpdateLogTableRetentionManagerImplTest {
+ private IdealStateHelper mockIdealStateHelper;
+ private UpdateLogStorageProvider mockStorageProvider;
+ private Map<String, Map<String, String>> segmentsInstanceMap;
+ private Set<String> updateLogStorageSegments;
+ private KCUpdateLogTableRetentionManagerImpl updateLogTableRetentionManager;
+
+ @BeforeMethod
+ public void setUp() throws IOException {
+ segmentsInstanceMap = new HashMap<>();
+
+ segmentsInstanceMap.put("table__0__10__20191027T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server2", "ONLINE"));
+ segmentsInstanceMap.put("table__0__11__20191028T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server3", "ONLINE"));
+ segmentsInstanceMap.put("table__1__10__20191027T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server2", "ONLINE"));
+ segmentsInstanceMap.put("table__2__10__20191027T2041Z",
+ ImmutableMap.of("server2", "ONLINE", "server3", "ONLINE"));
+
+ updateLogStorageSegments = new HashSet<>();
+ updateLogStorageSegments.add("table__0__10__20191027T2041Z");
+ updateLogStorageSegments.add("table__0__11__20191028T2041Z");
+ updateLogStorageSegments.add("table__1__10__20191027T2041Z");
+ updateLogStorageSegments.add("table__2__10__20191027T2041Z");
+
+ mockIdealStateHelper = mock(IdealStateHelper.class);
+ when(mockIdealStateHelper.getSegmentToInstanceMap("table")).thenReturn(segmentsInstanceMap);
+
+ mockStorageProvider = mock(UpdateLogStorageProvider.class);
+ when(mockStorageProvider.getAllSegments(anyString())).thenReturn(updateLogStorageSegments);
+
+ }
+
+ @Test
+ public void testInit() throws IOException {
+ updateLogStorageSegments.add("table__3__10__20191027T2041Z");
+ updateLogTableRetentionManager = new KCUpdateLogTableRetentionManagerImpl(mockIdealStateHelper, "table",
+ mockStorageProvider);
+
+ // verify that we update helix and performed deletion for expired segments
+ verify(mockStorageProvider, times(1))
+ .removeSegment("table", "table__3__10__20191027T2041Z");
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+ }
+
+ @Test
+ public void testShouldIngestForSegment() throws IOException {
+ updateLogTableRetentionManager = new KCUpdateLogTableRetentionManagerImpl(mockIdealStateHelper, "table",
+ mockStorageProvider);
+
+ // test the happy paths
+ Assert.assertTrue(updateLogTableRetentionManager.shouldIngestForSegment("table__0__10__20191027T2041Z"));
+ Assert.assertTrue(updateLogTableRetentionManager.shouldIngestForSegment("table__0__11__20191028T2041Z"));
+ Assert.assertTrue(updateLogTableRetentionManager.shouldIngestForSegment("table__1__10__20191027T2041Z"));
+ Assert.assertTrue(updateLogTableRetentionManager.shouldIngestForSegment("table__2__10__20191027T2041Z"));
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+
+ // test for segments that has lower seq
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__1__5__20191027T2041Z"));
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+
+ //test for segment that are older and not in existing segment list
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__1__20__20191027T2041Z"));
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+
+ //test for older timestamp but higher seq
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__0__20__20191027T2041Z"));
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+
+ // test for newer segments, and there is update in newer ideal state
+ segmentsInstanceMap.put("table__0__12__20191029T2041Z",
+ ImmutableMap.of("server1", "ONLINE", "server3", "ONLINE"));
+ Assert.assertTrue(updateLogTableRetentionManager.shouldIngestForSegment("table__0__12__20191029T2041Z"));
+ verify(mockIdealStateHelper, times(2)).getSegmentToInstanceMap("table");
+
+ // test for newer segments and there is no update in newer ideal state
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__0__13__20191029T2141Z"));
+ verify(mockIdealStateHelper, times(3)).getSegmentToInstanceMap("table");
+
+ // multiple attempts trying to fetch for segment should not recheck the ideal state often
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__0__13__20191029T2141Z"));
+ verify(mockIdealStateHelper, times(3)).getSegmentToInstanceMap("table");
+
+ // for unknown partition, we should also refresh data
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__2__13__20191029T2141Z"));
+ verify(mockIdealStateHelper, times(4)).getSegmentToInstanceMap("table");
+
+ Assert.assertFalse(updateLogTableRetentionManager.shouldIngestForSegment("table__3__13__20191029T2141Z"));
+ verify(mockIdealStateHelper, times(5)).getSegmentToInstanceMap("table");
+
+ }
+
+ @Test
+ public void testNotifySegmentsChange() throws IOException {
+ updateLogTableRetentionManager = new KCUpdateLogTableRetentionManagerImpl(mockIdealStateHelper, "table",
+ mockStorageProvider);
+ verify(mockIdealStateHelper, times(1)).getSegmentToInstanceMap("table");
+ updateLogTableRetentionManager.notifySegmentsChange();
+ verify(mockIdealStateHelper, times(2)).getSegmentToInstanceMap("table");
+ verify(mockStorageProvider, never()).removeSegment(anyString(), anyString());
+
+ segmentsInstanceMap.remove("table__0__10__20191027T2041Z");
+ updateLogTableRetentionManager.notifySegmentsChange();
+ verify(mockIdealStateHelper, times(3)).getSegmentToInstanceMap("table");
+ verify(mockStorageProvider, times(1)).removeSegment("table", "table__0__10__20191027T2041Z");
+ }
+}
\ No newline at end of file
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/ServerInstance.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/ServerInstance.java
index c5c6886..d215cbd 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/ServerInstance.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/ServerInstance.java
@@ -24,7 +24,7 @@ import org.apache.helix.HelixManager;
import org.apache.helix.ZNRecord;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.pinot.common.metrics.ServerMetrics;
-import org.apache.pinot.common.utils.IdealStateHelper;
+import org.apache.pinot.grigio.common.utils.IdealStateHelper;
import org.apache.pinot.core.data.manager.InstanceDataManager;
import org.apache.pinot.core.query.executor.QueryExecutor;
import org.apache.pinot.core.query.scheduler.QueryScheduler;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 11/12: small improve on loading data for virtual
column
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 51684f7ff2d8779b9afba6080b4e19447fb2751e
Author: james Shao <sj...@uber.com>
AuthorDate: Fri Nov 8 13:30:49 2019 -0800
small improve on loading data for virtual column
Summary: reduce data loading time and GC pressure by removing intermediate arrayList that stores all update event, use iterable that generate new short-live object as needed
Reviewers: tingchen, bzzhang, #streaming_pinot
Reviewed By: bzzhang, #streaming_pinot
Maniphest Tasks: T4441685
Differential Revision: https://code.uberinternal.com/D3572409
---
pinot-api/pom.xml | 2 +-
pinot-azure-filesystem/pom.xml | 2 +-
pinot-broker/pom.xml | 2 +-
pinot-common/pom.xml | 2 +-
.../pinot-connector-kafka-0.11/pom.xml | 2 +-
pinot-connectors/pinot-connector-kafka-0.9/pom.xml | 2 +-
pinot-connectors/pom.xml | 2 +-
pinot-controller/pom.xml | 2 +-
pinot-core/pom.xml | 2 +-
.../pinot/core/indexsegment/UpsertSegment.java | 3 +-
.../immutable/ImmutableUpsertSegmentImpl.java | 7 ++-
.../mutable/MutableUpsertSegmentImpl.java | 8 +--
.../immutable/ImmutableUpsertSegmentImplTest.java | 10 +++-
pinot-distribution/pom.xml | 2 +-
pinot-grigio/pinot-grigio-common/pom.xml | 2 +-
.../pinot/grigio/common/messages/LogEventType.java | 8 +--
.../SegmentUpdateLogStorageProvider.java | 25 +-------
.../common/storageProvider/UpdateLogEntry.java | 3 +
.../common/storageProvider/UpdateLogEntrySet.java | 69 ++++++++++++++++++++++
.../storageProvider/UpdateLogStorageExplorer.java | 7 +--
.../storageProvider/UpdateLogStorageProvider.java | 7 +--
.../SegmentUpdateLogStorageProviderTest.java | 62 +++++++++++++------
pinot-grigio/pinot-grigio-coordinator/pom.xml | 2 +-
pinot-grigio/pom.xml | 2 +-
pinot-hadoop-filesystem/pom.xml | 2 +-
pinot-hadoop/pom.xml | 2 +-
pinot-integration-tests/pom.xml | 2 +-
pinot-minion/pom.xml | 2 +-
pinot-orc/pom.xml | 2 +-
pinot-parquet/pom.xml | 2 +-
pinot-perf/pom.xml | 2 +-
pinot-server/pom.xml | 2 +-
pinot-tools/pom.xml | 2 +-
pinot-transport/pom.xml | 2 +-
pom.xml | 2 +-
35 files changed, 169 insertions(+), 88 deletions(-)
diff --git a/pinot-api/pom.xml b/pinot-api/pom.xml
index 41391eb..a1b43aa 100644
--- a/pinot-api/pom.xml
+++ b/pinot-api/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-api</artifactId>
<name>Pinot API</name>
diff --git a/pinot-azure-filesystem/pom.xml b/pinot-azure-filesystem/pom.xml
index ce07b85..4f6b42d 100644
--- a/pinot-azure-filesystem/pom.xml
+++ b/pinot-azure-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-azure-filesystem</artifactId>
<name>Pinot Azure Filesystem</name>
diff --git a/pinot-broker/pom.xml b/pinot-broker/pom.xml
index 4e92b3f..47402c4 100644
--- a/pinot-broker/pom.xml
+++ b/pinot-broker/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-broker</artifactId>
<name>Pinot Broker</name>
diff --git a/pinot-common/pom.xml b/pinot-common/pom.xml
index 4ce2e32..339905e 100644
--- a/pinot-common/pom.xml
+++ b/pinot-common/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-common</artifactId>
<name>Pinot Common</name>
diff --git a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
index 5e384d5..aa658ef 100644
--- a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
index daf74ad..c674d7a 100644
--- a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pom.xml b/pinot-connectors/pom.xml
index f4e8763..44b5673 100644
--- a/pinot-connectors/pom.xml
+++ b/pinot-connectors/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-connectors</artifactId>
diff --git a/pinot-controller/pom.xml b/pinot-controller/pom.xml
index 259300d..be511e3 100644
--- a/pinot-controller/pom.xml
+++ b/pinot-controller/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-controller</artifactId>
<name>Pinot Controller</name>
diff --git a/pinot-core/pom.xml b/pinot-core/pom.xml
index 06309cc..0dbdd63 100644
--- a/pinot-core/pom.xml
+++ b/pinot-core/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-core</artifactId>
<name>Pinot Core</name>
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/UpsertSegment.java b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/UpsertSegment.java
index 45fe425..125cf06 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/UpsertSegment.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/UpsertSegment.java
@@ -21,7 +21,6 @@ package org.apache.pinot.core.indexsegment;
import org.apache.pinot.grigio.common.storageProvider.UpdateLogEntry;
import java.io.IOException;
-import java.util.List;
public interface UpsertSegment {
@@ -29,7 +28,7 @@ public interface UpsertSegment {
* update the upsert-related virtual columns with the new values in this list of update logs
* @param messages list of updates logs to update the virtual columns
*/
- void updateVirtualColumn(List<UpdateLogEntry> messages);
+ void updateVirtualColumn(Iterable<UpdateLogEntry> messages);
/**
* get the upsert related virtual column debug info given an offset
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
index 1537689..5ab42f5 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
@@ -34,6 +34,7 @@ import org.apache.pinot.core.segment.updater.UpsertWaterMarkManager;
import org.apache.pinot.core.segment.virtualcolumn.mutable.VirtualColumnLongValueReaderWriter;
import org.apache.pinot.core.startree.v2.store.StarTreeIndexContainer;
import org.apache.pinot.grigio.common.storageProvider.UpdateLogEntry;
+import org.apache.pinot.grigio.common.storageProvider.UpdateLogEntrySet;
import org.apache.pinot.grigio.common.storageProvider.UpdateLogStorageProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -149,8 +150,8 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
}
@Override
- public void updateVirtualColumn(List<UpdateLogEntry> logEntryList) {
- for (UpdateLogEntry logEntry: logEntryList) {
+ public void updateVirtualColumn(Iterable<UpdateLogEntry> logEntries) {
+ for (UpdateLogEntry logEntry: logEntries) {
boolean updated = false;
int docId = getDocIdFromSourceOffset(logEntry.getOffset());
for (VirtualColumnLongValueReaderWriter readerWriter : _virtualColumnsReaderWriter) {
@@ -180,7 +181,7 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
@Override
public void initVirtualColumn() throws IOException {
long start = System.currentTimeMillis();
- final List<UpdateLogEntry> updateLogEntries = _updateLogStorageProvider.getAllMessages(_tableNameWithType, _segmentName);
+ final UpdateLogEntrySet updateLogEntries = _updateLogStorageProvider.getAllMessages(_tableNameWithType, _segmentName);
LOGGER.info("load {} update log entry from update log storage provider for segment {} in {} ms",
updateLogEntries.size(), _segmentName, System.currentTimeMillis() - start);
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java
index b69221d..ba6b6b5 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java
@@ -27,6 +27,7 @@ import org.apache.pinot.core.segment.updater.UpsertWaterMarkManager;
import org.apache.pinot.core.segment.virtualcolumn.mutable.VirtualColumnLongValueReaderWriter;
import org.apache.pinot.grigio.common.messages.LogEventType;
import org.apache.pinot.grigio.common.storageProvider.UpdateLogEntry;
+import org.apache.pinot.grigio.common.storageProvider.UpdateLogEntrySet;
import org.apache.pinot.grigio.common.storageProvider.UpdateLogStorageProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,7 +45,6 @@ public class MutableUpsertSegmentImpl extends MutableSegmentImpl implements Upse
private final String _kafkaOffsetColumnName;
-
private final List<VirtualColumnLongValueReaderWriter> _mutableSegmentReaderWriters = new ArrayList<>();
// use map for mapping between kafka offset and docId because we at-most have 1 mutable segment per consumer
// will use more memory, but we can update this later
@@ -71,8 +71,8 @@ public class MutableUpsertSegmentImpl extends MutableSegmentImpl implements Upse
}
@Override
- public synchronized void updateVirtualColumn(List<UpdateLogEntry> logEntryList) {
- for (UpdateLogEntry logEntry: logEntryList) {
+ public synchronized void updateVirtualColumn(Iterable<UpdateLogEntry> logEntries) {
+ for (UpdateLogEntry logEntry: logEntries) {
boolean updated = false;
boolean offsetFound = false;
Integer docId = _sourceOffsetToDocId.get(logEntry.getOffset());
@@ -126,7 +126,7 @@ public class MutableUpsertSegmentImpl extends MutableSegmentImpl implements Upse
@Override
public void initVirtualColumn() throws IOException {
Preconditions.checkState(_numDocsIndexed == 0, "should init virtual column before ingestion");
- List<UpdateLogEntry> updateLogEntries = UpdateLogStorageProvider.getInstance().getAllMessages(_tableName, _segmentName);
+ UpdateLogEntrySet updateLogEntries = UpdateLogStorageProvider.getInstance().getAllMessages(_tableName, _segmentName);
LOGGER.info("got {} update log entries for current segment {}", updateLogEntries.size(), _segmentName);
// some physical data might have been ingested when we init virtual column, we will go through the normal update
// flow to ensure we wont miss records
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImplTest.java b/pinot-core/src/test/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImplTest.java
index 4712625..19bef14 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImplTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImplTest.java
@@ -22,6 +22,7 @@ import org.apache.pinot.core.segment.updater.UpsertWaterMarkManager;
import org.apache.pinot.core.segment.virtualcolumn.mutable.VirtualColumnLongValueReaderWriter;
import org.apache.pinot.grigio.common.messages.LogEventType;
import org.apache.pinot.grigio.common.storageProvider.UpdateLogEntry;
+import org.apache.pinot.grigio.common.storageProvider.UpdateLogEntrySet;
import org.apache.pinot.grigio.common.storageProvider.UpdateLogStorageProvider;
import org.testng.Assert;
import org.testng.annotations.BeforeMethod;
@@ -29,6 +30,7 @@ import org.testng.annotations.Test;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
import static org.mockito.ArgumentMatchers.anyString;
@@ -92,7 +94,13 @@ public class ImmutableUpsertSegmentImplTest {
updateLogEntries.add(new UpdateLogEntry(minOffset + i, 50, LogEventType.INSERT, i%8));
updateLogEntries.add(new UpdateLogEntry(minOffset + i, 100, LogEventType.DELETE, i%8));
}
- when(_mockProvider.getAllMessages(anyString(), anyString())).thenReturn(updateLogEntries);
+ UpdateLogEntrySet entrySet = new UpdateLogEntrySet(null, 2) {
+ @Override
+ public Iterator<UpdateLogEntry> iterator() {
+ return updateLogEntries.iterator();
+ }
+ };
+ when(_mockProvider.getAllMessages(anyString(), anyString())).thenReturn(entrySet);
System.out.println("run time for set up: " + (System.currentTimeMillis() - start));
start = System.currentTimeMillis();
diff --git a/pinot-distribution/pom.xml b/pinot-distribution/pom.xml
index f55e1a7..12dcd75 100644
--- a/pinot-distribution/pom.xml
+++ b/pinot-distribution/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-distribution</artifactId>
<name>Pinot Distribution</name>
diff --git a/pinot-grigio/pinot-grigio-common/pom.xml b/pinot-grigio/pinot-grigio-common/pom.xml
index 0c98e85..a5d4152 100644
--- a/pinot-grigio/pinot-grigio-common/pom.xml
+++ b/pinot-grigio/pinot-grigio-common/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/messages/LogEventType.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/messages/LogEventType.java
index fa0db46..b4e9605 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/messages/LogEventType.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/messages/LogEventType.java
@@ -18,8 +18,6 @@
*/
package org.apache.pinot.grigio.common.messages;
-import com.google.common.base.Preconditions;
-
import java.util.HashMap;
import java.util.Map;
@@ -44,9 +42,7 @@ public enum LogEventType {
return this._uuid;
}
- public static LogEventType getEventType(long uuid) {
- int uuidInt = Math.toIntExact(uuid);
- Preconditions.checkState(UUID_MAP.containsKey(uuidInt));
- return UUID_MAP.get(uuidInt);
+ public static LogEventType getEventType(int uuid) {
+ return UUID_MAP.get(uuid);
}
}
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProvider.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProvider.java
index 02f2fba..e7b035f 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProvider.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProvider.java
@@ -20,8 +20,6 @@ package org.apache.pinot.grigio.common.storageProvider;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
-import org.apache.pinot.grigio.common.messages.LogEventType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,7 +31,6 @@ import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Paths;
-import java.util.ArrayList;
import java.util.List;
/**
@@ -57,31 +54,15 @@ public class SegmentUpdateLogStorageProvider {
_outputStream = new FileOutputStream(_file, true);
}
- public synchronized List<UpdateLogEntry> readAllMessagesFromFile() throws IOException {
- long start = System.currentTimeMillis();
- int insertMessageCount = 0;
- int deleteMessageCount = 0;
+ public synchronized UpdateLogEntrySet readAllMessagesFromFile() throws IOException {
int fileLength = (int) _file.length();
if (fileLength > 0) {
ByteBuffer buffer = ByteBuffer.allocate(fileLength);
readFullyFromBeginning(_file, buffer);
int messageCount = fileLength / UpdateLogEntry.SIZE;
- List<UpdateLogEntry> logs = new ArrayList<>(messageCount);
- for (int i = 0; i < messageCount; i++) {
- UpdateLogEntry logEntry = UpdateLogEntry.fromBytesBuffer(buffer);
- if ((logEntry.getType() == LogEventType.INSERT)) {
- insertMessageCount++;
- } else {
- deleteMessageCount++;
- }
- logs.add(logEntry);
- }
- buffer.clear();
- LOGGER.info("loaded {} message from file, {} insert and {} delete in {} ms", messageCount, insertMessageCount,
- deleteMessageCount, System.currentTimeMillis() - start);
- return logs;
+ return new UpdateLogEntrySet(buffer, messageCount);
} else {
- return ImmutableList.of();
+ return UpdateLogEntrySet.getEmptySet();
}
}
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogEntry.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogEntry.java
index c46f163..2ecc547 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogEntry.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogEntry.java
@@ -74,6 +74,9 @@ public class UpdateLogEntry implements Serializable {
}
public static UpdateLogEntry fromBytesBuffer(ByteBuffer buffer) {
+ if (buffer == null) {
+ throw new RuntimeException("trying to get update log event from null buffer");
+ }
return new UpdateLogEntry(buffer.getLong(), buffer.getLong(), LogEventType.getEventType(buffer.getInt()), buffer.getInt());
}
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogEntrySet.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogEntrySet.java
new file mode 100644
index 0000000..f756d73
--- /dev/null
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogEntrySet.java
@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.grigio.common.storageProvider;
+
+import java.nio.ByteBuffer;
+import java.util.Iterator;
+
+/**
+ * class for holding the list of update logs we read from files
+ * provide iterator interface for more efficient memory access
+ */
+public class UpdateLogEntrySet implements Iterable<UpdateLogEntry> {
+
+ private final ByteBuffer _buffer;
+ private final int _messageCount;
+ private static final UpdateLogEntrySet EMPTY_LOG_ENTRY_SET = new UpdateLogEntrySet(ByteBuffer.allocate(0),
+ 0);
+
+ public UpdateLogEntrySet(ByteBuffer buffer, int messageCount) {
+ _buffer = buffer;
+ _messageCount = messageCount;
+ }
+
+ public int size() {
+ return _messageCount;
+ }
+
+ @Override
+ public Iterator<UpdateLogEntry> iterator() {
+ return new Iterator<UpdateLogEntry>() {
+ @Override
+ public boolean hasNext() {
+ return _buffer != null && _buffer.hasRemaining();
+ }
+
+ @Override
+ public UpdateLogEntry next() {
+ if (!hasNext()) {
+ throw new RuntimeException("no more entries in buffer");
+ }
+ return UpdateLogEntry.fromBytesBuffer(_buffer);
+ }
+ };
+ }
+
+ /**
+ * helper method to create a default empty set in case of invalid/missing input
+ * @return an empty entry set has no data
+ */
+ public static UpdateLogEntrySet getEmptySet() {
+ return EMPTY_LOG_ENTRY_SET;
+ }
+}
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageExplorer.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageExplorer.java
index 3c9685d..0e38deb 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageExplorer.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageExplorer.java
@@ -26,7 +26,6 @@ import org.apache.commons.configuration.PropertiesConfiguration;
import java.io.IOException;
import java.util.Collection;
-import java.util.List;
import java.util.Scanner;
/**
@@ -56,10 +55,10 @@ public class UpdateLogStorageExplorer {
String segmentName = inputSplits[1];
provider.loadTable(tableName);
- List<UpdateLogEntry> updateLogEntryList = provider.getAllMessages(tableName, segmentName);
+ UpdateLogEntrySet updateLogEntrySet = provider.getAllMessages(tableName, segmentName);
Multimap<Long, UpdateLogEntry> map = ArrayListMultimap.create();
- System.out.println("update log size: " + updateLogEntryList.size());
- updateLogEntryList.forEach(u -> {
+ System.out.println("update log size: " + updateLogEntrySet.size());
+ updateLogEntrySet.forEach(u -> {
map.put(u.getOffset(), u);
});
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageProvider.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageProvider.java
index d04b1a2..b9802c2 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageProvider.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/storageProvider/UpdateLogStorageProvider.java
@@ -20,7 +20,6 @@ package org.apache.pinot.grigio.common.storageProvider;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.StringUtils;
@@ -142,18 +141,18 @@ public class UpdateLogStorageProvider {
}
}
- public List<UpdateLogEntry> getAllMessages(String tableName, String segmentName) throws IOException {
+ public UpdateLogEntrySet getAllMessages(String tableName, String segmentName) throws IOException {
if (_virtualColumnStorage.containsKey(tableName)) {
SegmentUpdateLogStorageProvider provider = _virtualColumnStorage.get(tableName).get(segmentName);
if (provider != null) {
return provider.readAllMessagesFromFile();
} else {
LOGGER.warn("don't have data for segment {}", segmentName);
- return ImmutableList.of();
+ return UpdateLogEntrySet.getEmptySet();
}
} else {
LOGGER.error("don't have data for table {}", tableName);
- return ImmutableList.of();
+ return UpdateLogEntrySet.getEmptySet();
}
}
diff --git a/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProviderTest.java b/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProviderTest.java
index 64d4d3b..219fbc8 100644
--- a/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProviderTest.java
+++ b/pinot-grigio/pinot-grigio-common/src/test/java/org/apache/pinot/grigio/common/storageProvider/SegmentUpdateLogStorageProviderTest.java
@@ -27,6 +27,7 @@ import org.testng.annotations.Test;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
@@ -34,6 +35,7 @@ import java.util.concurrent.Executors;
public class SegmentUpdateLogStorageProviderTest {
+ protected volatile UpdateLogEntry entryHolder;
SegmentUpdateLogStorageProvider provider;
List<UpdateLogEntry> inputDataList = ImmutableList.of(
new UpdateLogEntry(1, 2, LogEventType.INSERT, 0),
@@ -49,17 +51,19 @@ public class SegmentUpdateLogStorageProviderTest {
@Test
public void testWriteAndReadData() throws IOException {
- List<UpdateLogEntry> logEntryList = provider.readAllMessagesFromFile();
+ UpdateLogEntrySet logEntrySet = provider.readAllMessagesFromFile();
// new file should have no data
- Assert.assertEquals(logEntryList.size(), 0);
+ Assert.assertEquals(logEntrySet.size(), 0);
provider.addData(inputDataList);
SegmentUpdateLogStorageProvider provider1= new SegmentUpdateLogStorageProvider(provider._file);
- logEntryList = provider1.readAllMessagesFromFile();
- Assert.assertEquals(logEntryList.size(), inputDataList.size());
- Assert.assertEquals(logEntryList.get(0), inputDataList.get(0));
- Assert.assertEquals(logEntryList.get(1), inputDataList.get(1));
- Assert.assertEquals(logEntryList.get(2), inputDataList.get(2));
+ logEntrySet = provider1.readAllMessagesFromFile();
+ Iterator<UpdateLogEntry> it = logEntrySet.iterator();
+ Assert.assertEquals(logEntrySet.size(), inputDataList.size());
+ Assert.assertEquals(it.next(), inputDataList.get(0));
+ Assert.assertEquals(it.next(), inputDataList.get(1));
+ Assert.assertEquals(it.next(), inputDataList.get(2));
+ Assert.assertFalse(it.hasNext());
provider.addData(inputDataList);
Assert.assertEquals(provider.readAllMessagesFromFile().size(), inputDataList.size() * 2);
}
@@ -72,11 +76,13 @@ public class SegmentUpdateLogStorageProviderTest {
provider._outputStream.flush();
SegmentUpdateLogStorageProvider provider1 = new SegmentUpdateLogStorageProvider(provider._file);
- List<UpdateLogEntry> logEntryList = provider1.readAllMessagesFromFile();
- Assert.assertEquals(logEntryList.size(), inputDataList.size());
- Assert.assertEquals(logEntryList.get(0), inputDataList.get(0));
- Assert.assertEquals(logEntryList.get(1), inputDataList.get(1));
- Assert.assertEquals(logEntryList.get(2), inputDataList.get(2));
+ UpdateLogEntrySet logEntrySet = provider1.readAllMessagesFromFile();
+ Iterator<UpdateLogEntry> it = logEntrySet.iterator();
+ Assert.assertEquals(logEntrySet.size(), inputDataList.size());
+ Assert.assertEquals(it.next(), inputDataList.get(0));
+ Assert.assertEquals(it.next(), inputDataList.get(1));
+ Assert.assertEquals(it.next(), inputDataList.get(2));
+ Assert.assertFalse(it.hasNext());
}
@Test
@@ -108,18 +114,21 @@ public class SegmentUpdateLogStorageProviderTest {
});
service.invokeAll(tasks);
service.shutdownNow();
- List<UpdateLogEntry> updateLogEntries = provider.readAllMessagesFromFile();
+ UpdateLogEntrySet updateLogEntries = provider.readAllMessagesFromFile();
Assert.assertEquals(updateLogEntries.size(), writeIterationCount * inputDataList.size());
+ Iterator<UpdateLogEntry> it = updateLogEntries.iterator();
for (int i = 0; i < writeIterationCount; i++) {
- Assert.assertEquals(updateLogEntries.get(i * inputDataList.size()), inputDataList.get(0));
- Assert.assertEquals(updateLogEntries.get(i * inputDataList.size() + 1), inputDataList.get(1));
- Assert.assertEquals(updateLogEntries.get(i * inputDataList.size() + 2), inputDataList.get(2));
+ Assert.assertEquals(it.next(), inputDataList.get(0));
+ Assert.assertEquals(it.next(), inputDataList.get(1));
+ Assert.assertEquals(it.next(), inputDataList.get(2));
}
+ Assert.assertFalse(it.hasNext());
}
@Test
public void testReadMesssagePerf() throws IOException {
int totalMessageCount = 5_000_000;
+ // write a lot of data to file
List<UpdateLogEntry> inputMessages = new ArrayList<>(totalMessageCount * 2);
for (int i = 0; i < totalMessageCount; i++) {
inputMessages.add(new UpdateLogEntry(i, 50, LogEventType.INSERT, i%8));
@@ -128,8 +137,25 @@ public class SegmentUpdateLogStorageProviderTest {
long start = System.currentTimeMillis();
provider.addData(inputMessages);
System.out.println("write data takes ms: " + (System.currentTimeMillis() - start));
+
+ // load data from file to temp object, we don't measure this performance as it depends on disk/computer
+ start = System.currentTimeMillis();
+ UpdateLogEntrySet entrySet = provider.readAllMessagesFromFile();
+ long loadTime = System.currentTimeMillis() - start;
+ System.out.println("load data takes ms: " + loadTime);
+ Assert.assertTrue(entrySet.size() == totalMessageCount * 2);
+
+ // old implementation where we hold the data in array list will take 1000 - 2000 seconds for the data loading
+ // using iterator (current implementation) should make this code finished within 300 - 600 ms.
+ // test accessing those object
start = System.currentTimeMillis();
- provider.readAllMessagesFromFile();
- System.out.println("read data takes ms: " + (System.currentTimeMillis() - start));
+ for (UpdateLogEntry entry: entrySet) {
+ // ensure we hold them in volatile member to force JVM allocate the object and
+ // prevent JIT optimize this part of code away
+ entryHolder = entry;
+ }
+ long readTime = System.currentTimeMillis() - start;
+ Assert.assertTrue(readTime < 1_000L); // this should be relatively fast
+ System.out.println("read data takes ms: " + readTime);
}
}
\ No newline at end of file
diff --git a/pinot-grigio/pinot-grigio-coordinator/pom.xml b/pinot-grigio/pinot-grigio-coordinator/pom.xml
index 7ef66ce..8e3c353 100644
--- a/pinot-grigio/pinot-grigio-coordinator/pom.xml
+++ b/pinot-grigio/pinot-grigio-coordinator/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pinot-grigio-coordinator</artifactId>
diff --git a/pinot-grigio/pom.xml b/pinot-grigio/pom.xml
index 2d3a6a7..145912c 100644
--- a/pinot-grigio/pom.xml
+++ b/pinot-grigio/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pinot-grigio</artifactId>
diff --git a/pinot-hadoop-filesystem/pom.xml b/pinot-hadoop-filesystem/pom.xml
index 0e0f03d..fa64d3a 100644
--- a/pinot-hadoop-filesystem/pom.xml
+++ b/pinot-hadoop-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-hadoop-filesystem</artifactId>
<name>Pinot Hadoop Filesystem</name>
diff --git a/pinot-hadoop/pom.xml b/pinot-hadoop/pom.xml
index 13691d8..967264a 100644
--- a/pinot-hadoop/pom.xml
+++ b/pinot-hadoop/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-hadoop</artifactId>
<name>Pinot Hadoop</name>
diff --git a/pinot-integration-tests/pom.xml b/pinot-integration-tests/pom.xml
index 8ea6a8d..76af408 100644
--- a/pinot-integration-tests/pom.xml
+++ b/pinot-integration-tests/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-integration-tests</artifactId>
<name>Pinot Integration Tests</name>
diff --git a/pinot-minion/pom.xml b/pinot-minion/pom.xml
index 72b1d85..82a34f5 100644
--- a/pinot-minion/pom.xml
+++ b/pinot-minion/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-minion</artifactId>
<name>Pinot Minion</name>
diff --git a/pinot-orc/pom.xml b/pinot-orc/pom.xml
index 9508747..fcc5064 100644
--- a/pinot-orc/pom.xml
+++ b/pinot-orc/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-parquet/pom.xml b/pinot-parquet/pom.xml
index 577e283..956ed02 100644
--- a/pinot-parquet/pom.xml
+++ b/pinot-parquet/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-perf/pom.xml b/pinot-perf/pom.xml
index 704f659..0301af5 100644
--- a/pinot-perf/pom.xml
+++ b/pinot-perf/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-perf</artifactId>
<name>Pinot Perf</name>
diff --git a/pinot-server/pom.xml b/pinot-server/pom.xml
index 0503db3..41039c0 100644
--- a/pinot-server/pom.xml
+++ b/pinot-server/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-server</artifactId>
<name>Pinot Server</name>
diff --git a/pinot-tools/pom.xml b/pinot-tools/pom.xml
index 383cc1f..b6c8cda 100644
--- a/pinot-tools/pom.xml
+++ b/pinot-tools/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-tools</artifactId>
<name>Pinot Tools</name>
diff --git a/pinot-transport/pom.xml b/pinot-transport/pom.xml
index 5359b71..e38eda3 100644
--- a/pinot-transport/pom.xml
+++ b/pinot-transport/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
</parent>
<artifactId>pinot-transport</artifactId>
<name>Pinot Transport</name>
diff --git a/pom.xml b/pom.xml
index 66af82b..9d81542 100644
--- a/pom.xml
+++ b/pom.xml
@@ -30,7 +30,7 @@
<groupId>org.apache.pinot</groupId>
<artifactId>pinot</artifactId>
- <version>0.2.2.5.57-SNAPSHOT</version>
+ <version>0.2.2.5.58-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Pinot</name>
<description>A realtime distributed OLAP datastore</description>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 01/12: [maven-release-plugin] prepare release
pinot-0.2.2.5.43
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit f35f1d7fab3340a1bf53e806ffaf647dd2084c4b
Author: Uber Jenkins <pa...@uber.com>
AuthorDate: Wed Oct 16 00:30:22 2019 +0000
[maven-release-plugin] prepare release pinot-0.2.2.5.43
---
pinot-api/pom.xml | 2 +-
pinot-azure-filesystem/pom.xml | 2 +-
pinot-broker/pom.xml | 2 +-
pinot-common/pom.xml | 2 +-
pinot-connectors/pinot-connector-kafka-0.11/pom.xml | 2 +-
pinot-connectors/pinot-connector-kafka-0.9/pom.xml | 2 +-
pinot-connectors/pom.xml | 2 +-
pinot-controller/pom.xml | 2 +-
pinot-core/pom.xml | 2 +-
pinot-distribution/pom.xml | 2 +-
pinot-grigio/pinot-grigio-common/pom.xml | 2 +-
pinot-grigio/pinot-grigio-coordinator/pom.xml | 2 +-
pinot-grigio/pom.xml | 2 +-
pinot-hadoop-filesystem/pom.xml | 2 +-
pinot-hadoop/pom.xml | 2 +-
pinot-integration-tests/pom.xml | 2 +-
pinot-minion/pom.xml | 2 +-
pinot-orc/pom.xml | 2 +-
pinot-parquet/pom.xml | 2 +-
pinot-perf/pom.xml | 2 +-
pinot-server/pom.xml | 4 ++--
pinot-tools/pom.xml | 2 +-
pinot-transport/pom.xml | 2 +-
pom.xml | 4 ++--
24 files changed, 26 insertions(+), 26 deletions(-)
diff --git a/pinot-api/pom.xml b/pinot-api/pom.xml
index 2a5a295..8dac9f6 100644
--- a/pinot-api/pom.xml
+++ b/pinot-api/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-api</artifactId>
<name>Pinot API</name>
diff --git a/pinot-azure-filesystem/pom.xml b/pinot-azure-filesystem/pom.xml
index c536ffb..b01f8e4 100644
--- a/pinot-azure-filesystem/pom.xml
+++ b/pinot-azure-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-azure-filesystem</artifactId>
<name>Pinot Azure Filesystem</name>
diff --git a/pinot-broker/pom.xml b/pinot-broker/pom.xml
index 7aa652a..d78368a 100644
--- a/pinot-broker/pom.xml
+++ b/pinot-broker/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-broker</artifactId>
<name>Pinot Broker</name>
diff --git a/pinot-common/pom.xml b/pinot-common/pom.xml
index 4afb60a..9886aa5 100644
--- a/pinot-common/pom.xml
+++ b/pinot-common/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-common</artifactId>
<name>Pinot Common</name>
diff --git a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
index 0e033e6..cac22df 100644
--- a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
index 620cfd5..7218380 100644
--- a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pom.xml b/pinot-connectors/pom.xml
index 4a0f3ec..3bd9cc2 100644
--- a/pinot-connectors/pom.xml
+++ b/pinot-connectors/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-connectors</artifactId>
diff --git a/pinot-controller/pom.xml b/pinot-controller/pom.xml
index 01d33ac..68358c0 100644
--- a/pinot-controller/pom.xml
+++ b/pinot-controller/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-controller</artifactId>
<name>Pinot Controller</name>
diff --git a/pinot-core/pom.xml b/pinot-core/pom.xml
index d4fa501..0d1ae93 100644
--- a/pinot-core/pom.xml
+++ b/pinot-core/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-core</artifactId>
<name>Pinot Core</name>
diff --git a/pinot-distribution/pom.xml b/pinot-distribution/pom.xml
index 1b837af..87d674b 100644
--- a/pinot-distribution/pom.xml
+++ b/pinot-distribution/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-distribution</artifactId>
<name>Pinot Distribution</name>
diff --git a/pinot-grigio/pinot-grigio-common/pom.xml b/pinot-grigio/pinot-grigio-common/pom.xml
index 8d0fe56..107853e 100644
--- a/pinot-grigio/pinot-grigio-common/pom.xml
+++ b/pinot-grigio/pinot-grigio-common/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-grigio/pinot-grigio-coordinator/pom.xml b/pinot-grigio/pinot-grigio-coordinator/pom.xml
index 409b0f6..26d4f90 100644
--- a/pinot-grigio/pinot-grigio-coordinator/pom.xml
+++ b/pinot-grigio/pinot-grigio-coordinator/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-grigio/pom.xml b/pinot-grigio/pom.xml
index 4323a86..874606e 100644
--- a/pinot-grigio/pom.xml
+++ b/pinot-grigio/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pinot-grigio</artifactId>
diff --git a/pinot-hadoop-filesystem/pom.xml b/pinot-hadoop-filesystem/pom.xml
index 3bea4ff..ff45d60 100644
--- a/pinot-hadoop-filesystem/pom.xml
+++ b/pinot-hadoop-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-hadoop-filesystem</artifactId>
<name>Pinot Hadoop Filesystem</name>
diff --git a/pinot-hadoop/pom.xml b/pinot-hadoop/pom.xml
index 88edfa8..8503e22 100644
--- a/pinot-hadoop/pom.xml
+++ b/pinot-hadoop/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-hadoop</artifactId>
<name>Pinot Hadoop</name>
diff --git a/pinot-integration-tests/pom.xml b/pinot-integration-tests/pom.xml
index cb16857..c87e0e7 100644
--- a/pinot-integration-tests/pom.xml
+++ b/pinot-integration-tests/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-integration-tests</artifactId>
<name>Pinot Integration Tests</name>
diff --git a/pinot-minion/pom.xml b/pinot-minion/pom.xml
index 2a37cac..e7b0f64 100644
--- a/pinot-minion/pom.xml
+++ b/pinot-minion/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-minion</artifactId>
<name>Pinot Minion</name>
diff --git a/pinot-orc/pom.xml b/pinot-orc/pom.xml
index f3d213a..08ba6a7 100644
--- a/pinot-orc/pom.xml
+++ b/pinot-orc/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-parquet/pom.xml b/pinot-parquet/pom.xml
index d4d3067..e44ae00 100644
--- a/pinot-parquet/pom.xml
+++ b/pinot-parquet/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-perf/pom.xml b/pinot-perf/pom.xml
index 65998ef..43ffc02 100644
--- a/pinot-perf/pom.xml
+++ b/pinot-perf/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-perf</artifactId>
<name>Pinot Perf</name>
diff --git a/pinot-server/pom.xml b/pinot-server/pom.xml
index 4efd35b..2ae24d7 100644
--- a/pinot-server/pom.xml
+++ b/pinot-server/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-server</artifactId>
<name>Pinot Server</name>
@@ -68,7 +68,7 @@
<dependency>
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-grigio-common</artifactId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
diff --git a/pinot-tools/pom.xml b/pinot-tools/pom.xml
index e69b555..e49075e 100644
--- a/pinot-tools/pom.xml
+++ b/pinot-tools/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-tools</artifactId>
<name>Pinot Tools</name>
diff --git a/pinot-transport/pom.xml b/pinot-transport/pom.xml
index 06c3ea8..14954aa 100644
--- a/pinot-transport/pom.xml
+++ b/pinot-transport/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
</parent>
<artifactId>pinot-transport</artifactId>
<name>Pinot Transport</name>
diff --git a/pom.xml b/pom.xml
index 747df60..5c1e255 100644
--- a/pom.xml
+++ b/pom.xml
@@ -30,7 +30,7 @@
<groupId>org.apache.pinot</groupId>
<artifactId>pinot</artifactId>
- <version>0.2.2.5.43-SNAPSHOT</version>
+ <version>0.2.2.5.43</version>
<packaging>pom</packaging>
<name>Pinot</name>
<description>A realtime distributed OLAP datastore</description>
@@ -96,7 +96,7 @@
<scm>
<developerConnection>scm:git:gitolite@code.uber.internal:data/pinot</developerConnection>
- <tag>pinot-0.1.076-SNAPSHOT</tag>
+ <tag>pinot-0.2.2.5.43</tag>
</scm>
<distributionManagement>
<repository>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 08/12: Add retries to RocksDB reads and writes
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit fea13a01fd24b2ab143a397c9a2a8d916759e935
Author: Bo Zhang <bz...@uber.com>
AuthorDate: Mon Nov 4 11:36:24 2019 -0800
Add retries to RocksDB reads and writes
Reviewers: #streaming_pinot, sjames
Reviewed By: #streaming_pinot, sjames
Subscribers: sjames
Maniphest Tasks: T4289467
Differential Revision: https://code.uberinternal.com/D3545295
---
.../common/keyValueStore/KeyValueStoreTable.java | 2 -
pinot-grigio/pinot-grigio-coordinator/pom.xml | 3 --
.../common/keyValueStore/RocksDBBatchReader.java | 62 ++++++++++++++++++++++
.../common/keyValueStore/RocksDBBatchWriter.java | 56 +++++++++++++++++++
.../keyValueStore/RocksDBKeyValueStoreTable.java | 31 +++--------
5 files changed, 126 insertions(+), 28 deletions(-)
diff --git a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/keyValueStore/KeyValueStoreTable.java b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/keyValueStore/KeyValueStoreTable.java
index 5145fd8..d39eae7 100644
--- a/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/keyValueStore/KeyValueStoreTable.java
+++ b/pinot-grigio/pinot-grigio-common/src/main/java/org/apache/pinot/grigio/common/keyValueStore/KeyValueStoreTable.java
@@ -26,8 +26,6 @@ public interface KeyValueStoreTable<K, V> {
Map<K, V> multiGet(List<K> keys) throws IOException;
- void multiPut(List<K> keys, List<V> values) throws IOException;
-
void multiPut(Map<K, V> keyValuePairs) throws IOException;
void deleteTable() throws IOException;
diff --git a/pinot-grigio/pinot-grigio-coordinator/pom.xml b/pinot-grigio/pinot-grigio-coordinator/pom.xml
index 19499d1..f5b9527 100644
--- a/pinot-grigio/pinot-grigio-coordinator/pom.xml
+++ b/pinot-grigio/pinot-grigio-coordinator/pom.xml
@@ -63,7 +63,6 @@
<dependency>
<groupId>commons-configuration</groupId>
<artifactId>commons-configuration</artifactId>
- <version>1.6</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
@@ -102,6 +101,4 @@
<scope>test</scope>
</dependency>
</dependencies>
-
-
</project>
\ No newline at end of file
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBBatchReader.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBBatchReader.java
new file mode 100644
index 0000000..3512414
--- /dev/null
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBBatchReader.java
@@ -0,0 +1,62 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.grigio.common.keyValueStore;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class RocksDBBatchReader implements Callable<Boolean> {
+ static final int MAX_RETRY_ATTEMPTS = 3;
+ static final long RETRY_WAIT_MS = 1000L;
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(RocksDBBatchReader.class);
+
+ private final RocksDB _rocksDB;
+ private final List<byte[]> _byteKeys;
+ private Map<byte[], byte[]> _result;
+
+ RocksDBBatchReader(RocksDB rocksDB, List<byte[]> byteKeys) {
+ _rocksDB = rocksDB;
+ _byteKeys = byteKeys;
+ }
+
+ Map<byte[], byte[]> getResult() {
+ if (_result == null) {
+ throw new RuntimeException("No data got from RocksDB yet!");
+ }
+ return _result;
+ }
+
+ @Override
+ public Boolean call() {
+ try {
+ _result = _rocksDB.multiGet(_byteKeys);
+ return true;
+ } catch (RocksDBException e) {
+ LOGGER.warn("Failed to read from RocksDB: ", e);
+ return false;
+ }
+ }
+}
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBBatchWriter.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBBatchWriter.java
new file mode 100644
index 0000000..a910c02
--- /dev/null
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBBatchWriter.java
@@ -0,0 +1,56 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.grigio.common.keyValueStore;
+
+import java.util.concurrent.Callable;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.WriteBatch;
+import org.rocksdb.WriteOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class RocksDBBatchWriter implements Callable<Boolean> {
+ static final int MAX_RETRY_ATTEMPTS = 3;
+ static final long RETRY_WAIT_MS = 1000L;
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(RocksDBBatchWriter.class);
+
+ private final RocksDB _rocksDB;
+ private final WriteOptions _writeOptions;
+ private final WriteBatch _batch;
+
+ public RocksDBBatchWriter(RocksDB rocksDB, WriteOptions writeOptions, WriteBatch batch) {
+ _rocksDB = rocksDB;
+ _writeOptions = writeOptions;
+ _batch = batch;
+ }
+
+ @Override
+ public Boolean call() {
+ try {
+ _rocksDB.write(_writeOptions, _batch);
+ return true;
+ } catch (RocksDBException e) {
+ LOGGER.warn("Failed to write to RocksDB: ", e);
+ return false;
+ }
+ }
+}
diff --git a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBKeyValueStoreTable.java b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBKeyValueStoreTable.java
index 814f392..a712360 100644
--- a/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBKeyValueStoreTable.java
+++ b/pinot-grigio/pinot-grigio-coordinator/src/main/java/org/apache/pinot/grigio/common/keyValueStore/RocksDBKeyValueStoreTable.java
@@ -18,7 +18,7 @@
*/
package org.apache.pinot.grigio.common.keyValueStore;
-import com.google.common.base.Preconditions;
+import org.apache.pinot.common.utils.retry.RetryPolicies;
import org.apache.pinot.grigio.common.messages.KeyCoordinatorMessageContext;
import org.rocksdb.Options;
import org.rocksdb.ReadOptions;
@@ -65,7 +65,9 @@ public class RocksDBKeyValueStoreTable implements KeyValueStoreTable<ByteArrayWr
public Map<ByteArrayWrapper, KeyCoordinatorMessageContext> multiGet(List<ByteArrayWrapper> keys) throws IOException {
try {
List<byte[]> byteKeys = keys.stream().map(ByteArrayWrapper::getData).collect(Collectors.toList());
- Map<byte[], byte[]> rocksdbResult = _db.multiGet(byteKeys);
+ RocksDBBatchReader batchReader = new RocksDBBatchReader(_db, byteKeys);
+ RetryPolicies.fixedDelayRetryPolicy(RocksDBBatchReader.MAX_RETRY_ATTEMPTS, RocksDBBatchReader.RETRY_WAIT_MS).attempt(batchReader);
+ Map<byte[], byte[]> rocksdbResult = batchReader.getResult();
Map<ByteArrayWrapper, KeyCoordinatorMessageContext> result = new HashMap<>(rocksdbResult.size());
for (Map.Entry<byte[], byte[]> entry : rocksdbResult.entrySet()) {
Optional<KeyCoordinatorMessageContext> value = KeyCoordinatorMessageContext.fromBytes(entry.getValue());
@@ -76,30 +78,12 @@ public class RocksDBKeyValueStoreTable implements KeyValueStoreTable<ByteArrayWr
}
}
return result;
- } catch (RocksDBException e) {
+ } catch (Exception e) {
throw new IOException("failed to get keys from rocksdb " + _path, e);
}
}
@Override
- public void multiPut(List<ByteArrayWrapper> keys, List<KeyCoordinatorMessageContext> values) throws IOException {
- Preconditions.checkState(keys.size() == values.size(),
- "keys size {} does not match values size {}", keys.size(), values.size());
- if (keys.size() == 0) {
- return;
- }
- final WriteBatch batch = new WriteBatch();
- try {
- for (int i = 0; i < keys.size(); i++) {
- batch.put(keys.get(i).getData(), values.get(i).toBytes());
- }
- _db.write(_writeOptions, batch);
- } catch (RocksDBException e) {
- throw new IOException("failed to put data to rocksdb table " + _path, e);
- }
- }
-
- @Override
public void multiPut(Map<ByteArrayWrapper, KeyCoordinatorMessageContext> keyValuePairs) throws IOException {
if (keyValuePairs.size() == 0) {
return;
@@ -109,8 +93,9 @@ public class RocksDBKeyValueStoreTable implements KeyValueStoreTable<ByteArrayWr
for (Map.Entry<ByteArrayWrapper, KeyCoordinatorMessageContext> entry: keyValuePairs.entrySet()) {
batch.put(entry.getKey().getData(), entry.getValue().toBytes());
}
- _db.write(_writeOptions, batch);
- } catch (RocksDBException e) {
+ RocksDBBatchWriter batchWriter = new RocksDBBatchWriter(_db, _writeOptions, batch);
+ RetryPolicies.fixedDelayRetryPolicy(RocksDBBatchWriter.MAX_RETRY_ATTEMPTS, RocksDBBatchWriter.RETRY_WAIT_MS).attempt(batchWriter);
+ } catch (Exception e) {
throw new IOException("failed to put data to rocksdb table " + _path, e);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 06/12: change offset mapping strategy for
immutable seg to reduce memory
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jamesshao pushed a commit to branch upsert
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 6df85f3da0259aee319b4360653e54ae61ba2067
Author: james Shao <sj...@uber.com>
AuthorDate: Thu Oct 31 17:14:23 2019 -0700
change offset mapping strategy for immutable seg to reduce memory
Summary:
update offset mapping to docId to use array to reduce memory
right now we use hashmap for such mapping. It is very expensive to do so
changing to array for mapping will use 4 byte/document.
We only do this for immutable segment as it is easier to controll their size,
also immutable segment is the majority where we incur the memory cost
Reviewers: csoman, tingchen, bzzhang
Reviewed By: bzzhang
Differential Revision: https://code.uberinternal.com/D3534691
---
pinot-api/pom.xml | 2 +-
pinot-azure-filesystem/pom.xml | 2 +-
pinot-broker/pom.xml | 2 +-
pinot-common/pom.xml | 2 +-
.../pinot-connector-kafka-0.11/pom.xml | 2 +-
pinot-connectors/pinot-connector-kafka-0.9/pom.xml | 2 +-
pinot-connectors/pom.xml | 2 +-
pinot-controller/pom.xml | 2 +-
pinot-core/pom.xml | 2 +-
.../immutable/ImmutableUpsertSegmentImpl.java | 111 +++++++++++++--------
.../mutable/MutableUpsertSegmentImpl.java | 2 +
pinot-distribution/pom.xml | 2 +-
pinot-grigio/pinot-grigio-common/pom.xml | 2 +-
pinot-grigio/pinot-grigio-coordinator/pom.xml | 4 +-
pinot-grigio/pom.xml | 2 +-
pinot-hadoop-filesystem/pom.xml | 2 +-
pinot-hadoop/pom.xml | 2 +-
pinot-integration-tests/pom.xml | 2 +-
pinot-minion/pom.xml | 2 +-
pinot-orc/pom.xml | 2 +-
pinot-parquet/pom.xml | 2 +-
pinot-perf/pom.xml | 2 +-
pinot-server/pom.xml | 2 +-
pinot-tools/pom.xml | 2 +-
pinot-transport/pom.xml | 2 +-
pom.xml | 2 +-
26 files changed, 97 insertions(+), 66 deletions(-)
diff --git a/pinot-api/pom.xml b/pinot-api/pom.xml
index f26d749..6d6e96b 100644
--- a/pinot-api/pom.xml
+++ b/pinot-api/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-api</artifactId>
<name>Pinot API</name>
diff --git a/pinot-azure-filesystem/pom.xml b/pinot-azure-filesystem/pom.xml
index 7f3e432..fe6c1c3 100644
--- a/pinot-azure-filesystem/pom.xml
+++ b/pinot-azure-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-azure-filesystem</artifactId>
<name>Pinot Azure Filesystem</name>
diff --git a/pinot-broker/pom.xml b/pinot-broker/pom.xml
index 36d6eb1..fbeb628 100644
--- a/pinot-broker/pom.xml
+++ b/pinot-broker/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-broker</artifactId>
<name>Pinot Broker</name>
diff --git a/pinot-common/pom.xml b/pinot-common/pom.xml
index 1939448..91df4fe 100644
--- a/pinot-common/pom.xml
+++ b/pinot-common/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-common</artifactId>
<name>Pinot Common</name>
diff --git a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
index 2e9b3f6..50ef1ad 100644
--- a/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.11/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
index f63166c..409e3cf 100644
--- a/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
+++ b/pinot-connectors/pinot-connector-kafka-0.9/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-connectors</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-connectors/pom.xml b/pinot-connectors/pom.xml
index 4784801..749ec5c 100644
--- a/pinot-connectors/pom.xml
+++ b/pinot-connectors/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-connectors</artifactId>
diff --git a/pinot-controller/pom.xml b/pinot-controller/pom.xml
index 9a326a3..0e0af54 100644
--- a/pinot-controller/pom.xml
+++ b/pinot-controller/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-controller</artifactId>
<name>Pinot Controller</name>
diff --git a/pinot-core/pom.xml b/pinot-core/pom.xml
index 577c235..a4e6ca8 100644
--- a/pinot-core/pom.xml
+++ b/pinot-core/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-core</artifactId>
<name>Pinot Core</name>
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
index f12de6f..f42255f 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/immutable/ImmutableUpsertSegmentImpl.java
@@ -20,7 +20,6 @@ package org.apache.pinot.core.indexsegment.immutable;
import com.clearspring.analytics.util.Preconditions;
import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Multimap;
import org.apache.pinot.common.config.TableNameBuilder;
import org.apache.pinot.common.data.Schema;
@@ -44,6 +43,7 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -59,9 +59,10 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
private final int _totalDoc;
private final ColumnIndexContainer _offsetColumnIndexContainer;
private final UpsertWaterMarkManager _upsertWaterMarkManager;
- // right now we use map for this storage. But it will cost 12 bytes per record. It will translate to 12GB for 1B records
- // will look into reducing this cost later
- private final ImmutableMap<Long, Integer> _sourceOffsetToDocId;
+ private long _minSourceOffset;
+ // use array for mapping bewteen offset to docId, where actual offset = min_offset + array_index
+ // use 4 bytes per record
+ private int[] _sourceOffsetToDocIdArray;
public ImmutableUpsertSegmentImpl(SegmentDirectory segmentDirectory,
SegmentMetadataImpl segmentMetadata,
@@ -83,13 +84,15 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
_virtualColumnsReaderWriter.add((VirtualColumnLongValueReaderWriter) container.getForwardIndex());
}
}
- _sourceOffsetToDocId = buildOffsetToDocIdMap();
+ buildOffsetToDocIdMap();
}
- private ImmutableMap<Long, Integer> buildOffsetToDocIdMap() {
+ private void buildOffsetToDocIdMap() {
final DataFileReader reader = _offsetColumnIndexContainer.getForwardIndex();
final Dictionary dictionary = _offsetColumnIndexContainer.getDictionary();
- ImmutableMap.Builder<Long, Integer> kafkaOffsetToDocIdBuilder = ImmutableMap.builder();
+ Map<Long, Integer> kafkaOffsetToDocIdMap = new HashMap<>();
+ long minOffset = Long.MAX_VALUE;
+ long maxOffset = 0;
if (reader instanceof BaseSingleColumnSingleValueReader) {
BaseSingleColumnSingleValueReader scsvReader = (BaseSingleColumnSingleValueReader) reader;
for (int docId = 0; docId < _totalDoc; docId++) {
@@ -102,10 +105,21 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
if (offset == null) {
LOGGER.error("kafka offset is null at docID {}", docId);
} else {
- kafkaOffsetToDocIdBuilder.put(offset, docId);
+ minOffset = Math.min(offset, minOffset);
+ maxOffset = Math.max(offset, maxOffset);
+ kafkaOffsetToDocIdMap.put(offset, docId);
+ }
+ }
+ _minSourceOffset = minOffset;
+ int size = Math.toIntExact(maxOffset - minOffset + 1);
+ _sourceOffsetToDocIdArray = new int[size];
+ for (int i = 0; i < size; i++) {
+ if (kafkaOffsetToDocIdMap.containsKey(i + minOffset)) {
+ _sourceOffsetToDocIdArray[i] = kafkaOffsetToDocIdMap.get(i + minOffset);
+ } else {
+ _sourceOffsetToDocIdArray[i] = -1;
}
}
- return kafkaOffsetToDocIdBuilder.build();
} else {
throw new RuntimeException("unexpected forward reader type for kafka offset column " + reader.getClass());
}
@@ -121,30 +135,22 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
public void updateVirtualColumn(List<UpdateLogEntry> logEntryList) {
for (UpdateLogEntry logEntry: logEntryList) {
boolean updated = false;
- Integer docId = _sourceOffsetToDocId.get(logEntry.getOffset());
- if (docId == null) {
- LOGGER.warn("segment {} failed to found docId for log update entry {}", _segmentName, logEntry.toString());
- } else {
- for (VirtualColumnLongValueReaderWriter readerWriter : _virtualColumnsReaderWriter) {
- updated = readerWriter.update(docId, logEntry.getValue(), logEntry.getType()) || updated;
- }
- if (updated) {
- _upsertWaterMarkManager.processMessage(_tableNameWithType, _segmentName, logEntry);
- }
+ int docId = getDocIdFromSourceOffset(logEntry.getOffset());
+ for (VirtualColumnLongValueReaderWriter readerWriter : _virtualColumnsReaderWriter) {
+ updated = readerWriter.update(docId, logEntry.getValue(), logEntry.getType()) || updated;
+ }
+ if (updated) {
+ _upsertWaterMarkManager.processMessage(_tableNameWithType, _segmentName, logEntry);
}
}
}
@Override
public String getVirtualColumnInfo(long offset) {
- Integer docId = _sourceOffsetToDocId.get(offset);
+ int docId = getDocIdFromSourceOffset(offset);
StringBuilder result = new StringBuilder("matched: ");
- if (docId == null) {
- result = new StringBuilder("no doc id found ");
- } else {
- for (VirtualColumnLongValueReaderWriter readerWriter : _virtualColumnsReaderWriter) {
- result.append(readerWriter.getLong(docId)).append("; ");
- }
+ for (VirtualColumnLongValueReaderWriter readerWriter : _virtualColumnsReaderWriter) {
+ result.append(readerWriter.getLong(docId)).append("; ");
}
return result.toString();
}
@@ -162,25 +168,50 @@ public class ImmutableUpsertSegmentImpl extends ImmutableSegmentImpl implements
for (UpdateLogEntry logEntry: updateLogEntries) {
updateLogEntryMap.put(logEntry.getOffset(), logEntry);
}
- for (Map.Entry<Long, Integer> mapEntry: _sourceOffsetToDocId.entrySet()) {
- final long offset = mapEntry.getKey();
- final int docId = mapEntry.getValue();
- if (updateLogEntryMap.containsKey(offset)) {
- boolean updated = false;
- Collection<UpdateLogEntry> entries = updateLogEntryMap.get(offset);
- UpdateLogEntry lastEntry = null;
- for (UpdateLogEntry entry : entries) {
- lastEntry = entry;
- for (VirtualColumnLongValueReaderWriter readerWriter : _virtualColumnsReaderWriter) {
- updated = readerWriter.update(docId, entry.getValue(), entry.getType()) || updated;
+ for (int i = 0; i < _sourceOffsetToDocIdArray.length; i++) {
+ if (_sourceOffsetToDocIdArray[i] != -1) {
+ final long offset = i + _minSourceOffset;
+ final int docId = _sourceOffsetToDocIdArray[i];
+ if (updateLogEntryMap.containsKey(offset)) {
+ boolean updated = false;
+ Collection<UpdateLogEntry> entries = updateLogEntryMap.get(offset);
+ UpdateLogEntry lastEntry = null;
+ for (UpdateLogEntry entry : entries) {
+ lastEntry = entry;
+ for (VirtualColumnLongValueReaderWriter readerWriter : _virtualColumnsReaderWriter) {
+ updated = readerWriter.update(docId, entry.getValue(), entry.getType()) || updated;
+ }
+ }
+ if (updated) {
+ _upsertWaterMarkManager.processMessage(_tableNameWithType, _segmentName, lastEntry);
}
- }
- if (updated) {
- _upsertWaterMarkManager.processMessage(_tableNameWithType, _segmentName, lastEntry);
}
}
}
LOGGER.info("loaded {} update log entries for current immutable segment {} in {} ms", updateLogEntries.size(),
_segmentName, System.currentTimeMillis() - start);
}
+
+ /**
+ * given a offset from source kafka topic, return the docId associated with it
+ * throw exception if there is no docId for the associated kafka offset (because kafka offset might not be continuous)
+ * @param offset offset in the source topic
+ * @return the corresponding docId
+ */
+ private int getDocIdFromSourceOffset(long offset) throws RuntimeException {
+ if (offset < _minSourceOffset || offset - _minSourceOffset >= _sourceOffsetToDocIdArray.length) {
+ LOGGER.error("offset {} is outside range for current segment {} start offset {} size {}",
+ offset, _segmentName, _minSourceOffset, _sourceOffsetToDocIdArray.length);
+ throw new RuntimeException("offset outside range");
+ } else {
+ int position = Math.toIntExact(offset - _minSourceOffset);
+ if (_sourceOffsetToDocIdArray[position] == -1) {
+ LOGGER.error("no docId associated with offset {} for segment {}", offset, _segmentName);
+ throw new RuntimeException("docId not found");
+ } else {
+ return _sourceOffsetToDocIdArray[position];
+ }
+ }
+
+ }
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java
index 253c76f..b4d2e15 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableUpsertSegmentImpl.java
@@ -45,6 +45,8 @@ public class MutableUpsertSegmentImpl extends MutableSegmentImpl implements Upse
private final String _kafkaOffsetColumnName;
private final List<VirtualColumnLongValueReaderWriter> _mutableSegmentReaderWriters = new ArrayList<>();
+ // use map for mapping between kafka offset and docId because we at-most have 1 mutable segment per consumer
+ // will use more memory, but we can update this later
private final Map<Long, Integer> _sourceOffsetToDocId = new ConcurrentHashMap<>();
// to store the update event that arrive before my current record
// TODO remove this in the later version of design if necessary
diff --git a/pinot-distribution/pom.xml b/pinot-distribution/pom.xml
index 49e3d9c..fb652d0 100644
--- a/pinot-distribution/pom.xml
+++ b/pinot-distribution/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-distribution</artifactId>
<name>Pinot Distribution</name>
diff --git a/pinot-grigio/pinot-grigio-common/pom.xml b/pinot-grigio/pinot-grigio-common/pom.xml
index 53adbbf..6d9d4bb 100644
--- a/pinot-grigio/pinot-grigio-common/pom.xml
+++ b/pinot-grigio/pinot-grigio-common/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-grigio/pinot-grigio-coordinator/pom.xml b/pinot-grigio/pinot-grigio-coordinator/pom.xml
index a7d6db2..bc85f0f 100644
--- a/pinot-grigio/pinot-grigio-coordinator/pom.xml
+++ b/pinot-grigio/pinot-grigio-coordinator/pom.xml
@@ -23,16 +23,14 @@
<parent>
<artifactId>pinot-grigio</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
-
<artifactId>pinot-grigio-coordinator</artifactId>
<name>Pinot Grigio Coordinator</name>
<properties>
<pinot.root>${basedir}/../..</pinot.root>
</properties>
-
<dependencies>
<dependency>
<groupId>org.glassfish.jersey.containers</groupId>
diff --git a/pinot-grigio/pom.xml b/pinot-grigio/pom.xml
index 5ac6c5a..2766d3a 100644
--- a/pinot-grigio/pom.xml
+++ b/pinot-grigio/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pinot-grigio</artifactId>
diff --git a/pinot-hadoop-filesystem/pom.xml b/pinot-hadoop-filesystem/pom.xml
index e130a69..6d44587 100644
--- a/pinot-hadoop-filesystem/pom.xml
+++ b/pinot-hadoop-filesystem/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-hadoop-filesystem</artifactId>
<name>Pinot Hadoop Filesystem</name>
diff --git a/pinot-hadoop/pom.xml b/pinot-hadoop/pom.xml
index 2be8498..2272b24 100644
--- a/pinot-hadoop/pom.xml
+++ b/pinot-hadoop/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-hadoop</artifactId>
<name>Pinot Hadoop</name>
diff --git a/pinot-integration-tests/pom.xml b/pinot-integration-tests/pom.xml
index 46ac371..1095763 100644
--- a/pinot-integration-tests/pom.xml
+++ b/pinot-integration-tests/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-integration-tests</artifactId>
<name>Pinot Integration Tests</name>
diff --git a/pinot-minion/pom.xml b/pinot-minion/pom.xml
index 4d37500..a7352ee 100644
--- a/pinot-minion/pom.xml
+++ b/pinot-minion/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-minion</artifactId>
<name>Pinot Minion</name>
diff --git a/pinot-orc/pom.xml b/pinot-orc/pom.xml
index 2e8672a..e046ca8 100644
--- a/pinot-orc/pom.xml
+++ b/pinot-orc/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-parquet/pom.xml b/pinot-parquet/pom.xml
index 690924a..5c5b9b0 100644
--- a/pinot-parquet/pom.xml
+++ b/pinot-parquet/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pinot-perf/pom.xml b/pinot-perf/pom.xml
index 113d974..3312026 100644
--- a/pinot-perf/pom.xml
+++ b/pinot-perf/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-perf</artifactId>
<name>Pinot Perf</name>
diff --git a/pinot-server/pom.xml b/pinot-server/pom.xml
index c1536cb..9463c64 100644
--- a/pinot-server/pom.xml
+++ b/pinot-server/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-server</artifactId>
<name>Pinot Server</name>
diff --git a/pinot-tools/pom.xml b/pinot-tools/pom.xml
index 4078c18..7aa523f 100644
--- a/pinot-tools/pom.xml
+++ b/pinot-tools/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-tools</artifactId>
<name>Pinot Tools</name>
diff --git a/pinot-transport/pom.xml b/pinot-transport/pom.xml
index 7b31dd0..dc748c7 100644
--- a/pinot-transport/pom.xml
+++ b/pinot-transport/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pinot</artifactId>
<groupId>org.apache.pinot</groupId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
</parent>
<artifactId>pinot-transport</artifactId>
<name>Pinot Transport</name>
diff --git a/pom.xml b/pom.xml
index d43b23c..c3d0cf0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -30,7 +30,7 @@
<groupId>org.apache.pinot</groupId>
<artifactId>pinot</artifactId>
- <version>0.2.2.5.44-SNAPSHOT</version>
+ <version>0.2.2.5.45-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Pinot</name>
<description>A realtime distributed OLAP datastore</description>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org