You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by GitBox <gi...@apache.org> on 2021/02/09 23:53:33 UTC
[GitHub] [incubator-pinot] chenboat opened a new pull request #6567: Enable segment upload to a realtime table
chenboat opened a new pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567
## Description
Currently segment upload is only applicable for offline Pinot tables. This PR opens the segment upload to Pinot realtime tables too. Segment upload to realtime tables makes bootstrapping of a realtime table much faster because the segments can now be computed outside of Pinot. This features allows Pinot realtime tables with longer retention (e.g., Pinot upsert enable table) to be bootstrapped in a much efficient manner.
Design doc: https://docs.google.com/document/d/1STYxZsUYGcYrzHdmOBymQ1NkpxuEzLa4OTiw4mGHiKk/edit#
## Upgrade Notes
Does this PR prevent a zero down-time upgrade? (Assume upgrade order: Controller, Broker, Server, Minion)
* [ ] Yes (Please label as **<code>backward-incompat</code>**, and complete the section below on Release Notes)
Does this PR fix a zero-downtime upgrade introduced earlier?
* [ ] Yes (Please label this as **<code>backward-incompat</code>**, and complete the section below on Release Notes)
Does this PR otherwise need attention when creating release notes? Things to consider:
- New configuration options
- Deprecation of configurations
- Signature changes to public methods/interfaces
- New plugins added or old plugins removed
* [ ] Yes (Please label this PR as **<code>release-notes</code>** and complete the section on Release Notes)
## Release Notes
If you have tagged this as either backward-incompat or release-notes,
you MUST add text here that you would like to see appear in release notes of the
next release.
If you have a series of commits adding or enabling a feature, then
add this section only in final commit that marks the feature completed.
Refer to earlier release notes to see examples of text
## Documentation
If you have introduced a new feature or configuration, please add it to the documentation as well.
See https://docs.pinot.apache.org/developers/developers-and-contributors/update-document
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r598028114
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUntarSegmentToLocal(String uri, File tableDataDir, String segmentName)
+ throws Exception {
+ File tempDir = new File(tableDataDir, "tmp-" + segmentName + "-" + UUID.randomUUID());
+ FileUtils.forceMkdir(tempDir);
+ File tempTarFile = new File(tempDir, segmentName + TAR_GZ_SUFFIX);
+ File tempSegmentDir = new File(tempDir, segmentName);
+ try {
+ try {
+ SegmentFetcherFactory.fetchSegmentToLocal(uri, tempTarFile);
+ LOGGER.info("Downloaded tarred segment: {} from: {} to: {}, file length: {}", segmentName, uri, tempTarFile,
+ tempTarFile.length());
+ } catch (AttemptsExceededException e) {
+ LOGGER.error("Attempts exceeded when downloading segment: {} from: {} to: {}", segmentName, uri,
+ tempTarFile, e);
+ Utils.rethrowException(e);
+ }
+
+ try {
+ // If an exception is thrown when untarring, it means the tar file is broken OR not found after the retry.
+ // Thus, there's no need to retry again.
+ File tempIndexDir = TarGzCompressionUtils.untar(tempTarFile, tempSegmentDir).get(0);
+ File segmentDir = new File(tableDataDir, segmentName);
+ if (segmentDir.exists()) {
+ LOGGER.info("Deleting existing index directory for segment: {}", segmentName);
+ FileUtils.deleteDirectory(segmentDir);
+ }
+ FileUtils.moveDirectory(tempIndexDir, segmentDir);
+ LOGGER.info("Successfully downloaded segment: {} to: {}", segmentName, segmentDir);
+ } catch (Exception e) {
+ LOGGER.error("Exception when untarring segment: {} for from {} to {}", segmentName, tempTarFile, tempSegmentDir,
+ e);
+ Utils.rethrowException(e);
+ }
+ } finally {
+ FileUtils.deleteQuietly(tempDir);
Review comment:
No as far as I know. This problem is common to many file related operation in Pinot today. We may need to think of a common solution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] yupeng9 commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
yupeng9 commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r596473726
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUntarSegmentToLocal(String uri, File tableDataDir, String segmentName)
Review comment:
@npawar fyi, shall this be part of the `SegmentUploader` implementation?
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java
##########
@@ -172,6 +183,39 @@ private void checkReplication(InstancePartitions instancePartitions) {
}
}
+ private int getSegmentPartitionId(String segmentName) {
+ // A fast path if the segmentName is a LLC segment name and we can get the partition id from the name directly.
+ if (LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
+ return new LLCSegmentName(segmentName).getPartitionGroupId();
+ }
+ // Otherwise, retrieve the partition id from the segment zk metadata. Currently only realtime segments from upsert
+ // enabled tables have partition ids in their segment metadata.
+ RealtimeSegmentZKMetadata segmentZKMetadata = ZKMetadataProvider
+ .getRealtimeSegmentZKMetadata(_helixManager.getHelixPropertyStore(), _realtimeTableName, segmentName);
+ Preconditions
+ .checkState(segmentZKMetadata != null, "Failed to find segment ZK metadata for segment: %s of table: %s",
+ segmentName, _realtimeTableName);
+ return getSegmentPartitionIdFromZkMetaData(segmentZKMetadata);
+ }
+
+ private int getSegmentPartitionIdFromZkMetaData(RealtimeSegmentZKMetadata segmentZKMetadata) {
+ String segmentName = segmentZKMetadata.getSegmentName();
+ Preconditions.checkState(segmentZKMetadata.getPartitionMetadata() != null,
+ "Segment ZK metadata for segment: %s of table: %s does not contain partition metadata", segmentName,
+ _realtimeTableName);
+
+ ColumnPartitionMetadata partitionMetadata =
+ segmentZKMetadata.getPartitionMetadata().getColumnPartitionMap().get(_partitionColumn);
+ Preconditions.checkState(partitionMetadata != null,
+ "Segment ZK metadata for segment: %s of table: %s does not contain partition metadata for column: %s",
Review comment:
shall we make the message more informative, by adding `checking if the table is in upsert mode`?
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUntarSegmentToLocal(String uri, File tableDataDir, String segmentName)
+ throws Exception {
+ File tempDir = new File(tableDataDir, "tmp-" + segmentName + "-" + UUID.randomUUID());
+ FileUtils.forceMkdir(tempDir);
+ File tempTarFile = new File(tempDir, segmentName + TAR_GZ_SUFFIX);
+ File tempSegmentDir = new File(tempDir, segmentName);
+ try {
+ try {
+ SegmentFetcherFactory.fetchSegmentToLocal(uri, tempTarFile);
+ LOGGER.info("Downloaded tarred segment: {} from: {} to: {}, file length: {}", segmentName, uri, tempTarFile,
+ tempTarFile.length());
+ } catch (AttemptsExceededException e) {
+ LOGGER.error("Attempts exceeded when downloading segment: {} from: {} to: {}", segmentName, uri,
+ tempTarFile, e);
+ Utils.rethrowException(e);
+ }
+
+ try {
+ // If an exception is thrown when untarring, it means the tar file is broken OR not found after the retry.
+ // Thus, there's no need to retry again.
+ File tempIndexDir = TarGzCompressionUtils.untar(tempTarFile, tempSegmentDir).get(0);
+ File segmentDir = new File(tableDataDir, segmentName);
+ if (segmentDir.exists()) {
+ LOGGER.info("Deleting existing index directory for segment: {}", segmentName);
+ FileUtils.deleteDirectory(segmentDir);
+ }
+ FileUtils.moveDirectory(tempIndexDir, segmentDir);
+ LOGGER.info("Successfully downloaded segment: {} to: {}", segmentName, segmentDir);
+ } catch (Exception e) {
+ LOGGER.error("Exception when untarring segment: {} for from {} to {}", segmentName, tempTarFile, tempSegmentDir,
Review comment:
no need for `for` in the msg?
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
##########
@@ -543,4 +556,41 @@ private boolean isValid(Schema schema, IndexingConfig indexingConfig) {
}
return isValid;
}
+
+ private int getSegmentPartitionId(String segmentName, String tableName) {
Review comment:
this logic is repeated in `RealtimeSegmentAssignment.java`?
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -358,7 +358,7 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ IN_PROGRESS, DONE, UPLOAD
Review comment:
+1 to @mcvsubbu's suggestion. @chenboat is this status/flow documented somewhere in the design doc?
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUntarSegmentToLocal(String uri, File tableDataDir, String segmentName)
+ throws Exception {
+ File tempDir = new File(tableDataDir, "tmp-" + segmentName + "-" + UUID.randomUUID());
+ FileUtils.forceMkdir(tempDir);
+ File tempTarFile = new File(tempDir, segmentName + TAR_GZ_SUFFIX);
+ File tempSegmentDir = new File(tempDir, segmentName);
+ try {
+ try {
+ SegmentFetcherFactory.fetchSegmentToLocal(uri, tempTarFile);
+ LOGGER.info("Downloaded tarred segment: {} from: {} to: {}, file length: {}", segmentName, uri, tempTarFile,
+ tempTarFile.length());
+ } catch (AttemptsExceededException e) {
+ LOGGER.error("Attempts exceeded when downloading segment: {} from: {} to: {}", segmentName, uri,
+ tempTarFile, e);
+ Utils.rethrowException(e);
+ }
+
+ try {
+ // If an exception is thrown when untarring, it means the tar file is broken OR not found after the retry.
+ // Thus, there's no need to retry again.
+ File tempIndexDir = TarGzCompressionUtils.untar(tempTarFile, tempSegmentDir).get(0);
+ File segmentDir = new File(tableDataDir, segmentName);
+ if (segmentDir.exists()) {
+ LOGGER.info("Deleting existing index directory for segment: {}", segmentName);
+ FileUtils.deleteDirectory(segmentDir);
+ }
+ FileUtils.moveDirectory(tempIndexDir, segmentDir);
+ LOGGER.info("Successfully downloaded segment: {} to: {}", segmentName, segmentDir);
+ } catch (Exception e) {
+ LOGGER.error("Exception when untarring segment: {} for from {} to {}", segmentName, tempTarFile, tempSegmentDir,
+ e);
+ Utils.rethrowException(e);
+ }
+ } finally {
+ FileUtils.deleteQuietly(tempDir);
Review comment:
the tmp files might not be cleaned up in the process crashed before hitting `finally`. Do we have some mechanism to recycle zombie files?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] codecov-io edited a comment on pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-776350585
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=h1) Report
> Merging [#6567](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=desc) (38cca2c) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/1beaab59b73f26c4e35f3b9bc856b03806cddf5a?el=desc) (1beaab5) will **decrease** coverage by `0.55%`.
> The diff coverage is `63.15%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #6567 +/- ##
==========================================
- Coverage 66.44% 65.89% -0.56%
==========================================
Files 1075 1403 +328
Lines 54773 67986 +13213
Branches 8168 9836 +1668
==========================================
+ Hits 36396 44797 +8401
- Misses 15700 19997 +4297
- Partials 2677 3192 +515
```
| Flag | Coverage Δ | |
|---|---|---|
| unittests | `65.89% <63.15%> (?)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...e/pinot/broker/api/resources/PinotBrokerDebug.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYXBpL3Jlc291cmNlcy9QaW5vdEJyb2tlckRlYnVnLmphdmE=) | `0.00% <0.00%> (-79.32%)` | :arrow_down: |
| [...pinot/broker/api/resources/PinotClientRequest.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYXBpL3Jlc291cmNlcy9QaW5vdENsaWVudFJlcXVlc3QuamF2YQ==) | `0.00% <0.00%> (-27.28%)` | :arrow_down: |
| [...ot/broker/broker/AllowAllAccessControlFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL0FsbG93QWxsQWNjZXNzQ29udHJvbEZhY3RvcnkuamF2YQ==) | `85.71% <ø> (-14.29%)` | :arrow_down: |
| [.../helix/BrokerUserDefinedMessageHandlerFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL2hlbGl4L0Jyb2tlclVzZXJEZWZpbmVkTWVzc2FnZUhhbmRsZXJGYWN0b3J5LmphdmE=) | `33.96% <0.00%> (-32.71%)` | :arrow_down: |
| [...ker/routing/instanceselector/InstanceSelector.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9pbnN0YW5jZXNlbGVjdG9yL0luc3RhbmNlU2VsZWN0b3IuamF2YQ==) | `100.00% <ø> (ø)` | |
| [...ava/org/apache/pinot/client/AbstractResultSet.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0Fic3RyYWN0UmVzdWx0U2V0LmphdmE=) | `66.66% <ø> (+9.52%)` | :arrow_up: |
| [...n/java/org/apache/pinot/client/BrokerResponse.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0Jyb2tlclJlc3BvbnNlLmphdmE=) | `100.00% <ø> (ø)` | |
| [.../main/java/org/apache/pinot/client/Connection.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0Nvbm5lY3Rpb24uamF2YQ==) | `35.55% <ø> (-13.29%)` | :arrow_down: |
| [...org/apache/pinot/client/DynamicBrokerSelector.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0R5bmFtaWNCcm9rZXJTZWxlY3Rvci5qYXZh) | `82.85% <ø> (+10.12%)` | :arrow_up: |
| [...n/java/org/apache/pinot/client/ExecutionStats.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0V4ZWN1dGlvblN0YXRzLmphdmE=) | `68.88% <ø> (ø)` | |
| ... and [1290 more](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=footer). Last update [53c2a3c...38cca2c](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r596478568
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1649,63 +1651,100 @@ public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, Str
public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ String tableNameWithType;
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
- OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
- ZKMetadataUtils.updateSegmentMetadata(offlineSegmentZKMetadata, segmentMetadata);
- offlineSegmentZKMetadata.setDownloadUrl(downloadUrl);
- offlineSegmentZKMetadata.setCrypterName(crypter);
- offlineSegmentZKMetadata.setPushTime(System.currentTimeMillis());
+ ZNRecord znRecord;
+
+ if (isUpsertTable(tableName)) {
Review comment:
Added the check.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r598028157
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
##########
@@ -543,4 +556,41 @@ private boolean isValid(Schema schema, IndexingConfig indexingConfig) {
}
return isValid;
}
+
+ private int getSegmentPartitionId(String segmentName, String tableName) {
Review comment:
Refactor to a common util class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] Jackie-Jiang commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r598967441
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUntarSegmentToLocal(String uri, File tableDataDir, String segmentName)
Review comment:
Why do we need this method? There is already an `untarAndMoveSegment()` within the `RealtimeTableDataManager` class. We should be able to use that
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
##########
@@ -57,41 +57,37 @@ public ZKOperator(PinotHelixResourceManager pinotHelixResourceManager, Controlle
_controllerMetrics = controllerMetrics;
}
- public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata,
+ public void completeSegmentOperations(String tableNameWithType, SegmentMetadata segmentMetadata,
URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection,
HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter)
throws Exception {
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
String segmentName = segmentMetadata.getName();
-
- // Brand new segment, not refresh, directly add the segment
- ZNRecord segmentMetadataZnRecord =
- _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName);
+ ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType, segmentName);
if (segmentMetadataZnRecord == null) {
- LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName);
+ LOGGER.info("Adding new segment {} from table {}", segmentName, tableNameWithType);
processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter,
- rawTableName, segmentName, moveSegmentToFinalLocation);
+ tableNameWithType, segmentName, moveSegmentToFinalLocation);
return;
}
- LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, rawTableName);
+ LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, tableNameWithType);
processExistingSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation,
- enableParallelPushProtection, headers, zkDownloadURI, crypter, offlineTableName, segmentName,
+ enableParallelPushProtection, headers, zkDownloadURI, crypter, tableNameWithType, segmentName,
segmentMetadataZnRecord, moveSegmentToFinalLocation);
}
private void processExistingSegment(SegmentMetadata segmentMetadata, URI finalSegmentLocationURI,
File currentSegmentLocation, boolean enableParallelPushProtection, HttpHeaders headers, String zkDownloadURI,
- String crypter, String offlineTableName, String segmentName, ZNRecord znRecord,
+ String crypter, String tableNameWithType, String segmentName, ZNRecord znRecord,
boolean moveSegmentToFinalLocation)
throws Exception {
OfflineSegmentZKMetadata existingSegmentZKMetadata = new OfflineSegmentZKMetadata(znRecord);
Review comment:
(Critical) For real-time segment, this should not be `OfflineSegmentZKMetadata`, where the `status` and `offset` info are lost
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java
##########
@@ -19,6 +19,7 @@
package org.apache.pinot.controller.helix.core.assignment.segment;
import com.google.common.base.Preconditions;
+
Review comment:
(nit) reformat
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -403,10 +414,13 @@ private void decryptFile(String crypterClassName, File tempEncryptedFile, File t
// it keeps it at the downloadURI header that is set. We will not support this endpoint going forward.
public void uploadSegmentAsJson(String segmentJsonStr,
@ApiParam(value = "Name of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_NAME) String tableName,
+ @ApiParam(value = "Type of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_TYPE) @DefaultValue("OFFLINE") String tableType,
@ApiParam(value = "Whether to enable parallel push protection") @DefaultValue("false") @QueryParam(FileUploadDownloadClient.QueryParameters.ENABLE_PARALLEL_PUSH_PROTECTION) boolean enableParallelPushProtection,
@Context HttpHeaders headers, @Context Request request, @Suspended final AsyncResponse asyncResponse) {
try {
- asyncResponse.resume(uploadSegment(tableName, null, enableParallelPushProtection, headers, request, false));
+ asyncResponse.resume(
+ uploadSegment(tableName, "OFFLINE".equalsIgnoreCase(tableType) ? TableType.OFFLINE : TableType.REALTIME, null,
Review comment:
Use `TableType.valueOf(tableType.toUpperCase())`? We don't want to upload segment to real-time table when getting random `tableType`. Same for other places
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r575483440
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
+ if (_pinotHelixResourceManager.isRealtimeOnlyTable(rawTableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+ } else {
+ tableNameWithType = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ }
String clientAddress = InetAddress.getByName(request.getRemoteAddr()).getHostName();
LOGGER.info("Processing upload request for segment: {} of table: {} from client: {}, ingestion descriptor: {}",
- segmentName, offlineTableName, clientAddress, ingestionDescriptor);
+ segmentName, tableNameWithType, clientAddress, ingestionDescriptor);
- // Skip segment validation if upload only segment metadata
- if (uploadType != FileUploadDownloadClient.FileUploadType.METADATA) {
+ // Skip segment validation if upload only segment metadata or it is a realtime table segment.
+ // TODO Perform a validation check for realtime segments too.
Review comment:
Yes. @yupeng9 restricting to upsert table only (by doing table config validation) is a good idea.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] fx19880617 commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
fx19880617 commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r593726121
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUtarSegmentToLocal(String uri, File tableDataDir, String segmentName)
Review comment:
nit: fetchAndUntarSegmentToLocal
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java
##########
@@ -85,15 +92,19 @@
private HelixManager _helixManager;
private String _realtimeTableName;
private int _replication;
+ private String _partitionColumn;
@Override
public void init(HelixManager helixManager, TableConfig tableConfig) {
_helixManager = helixManager;
_realtimeTableName = tableConfig.getTableName();
_replication = tableConfig.getValidationConfig().getReplicasPerPartitionNumber();
+ ReplicaGroupStrategyConfig replicaGroupStrategyConfig =
+ tableConfig.getValidationConfig().getReplicaGroupStrategyConfig();
+ _partitionColumn = replicaGroupStrategyConfig != null ? replicaGroupStrategyConfig.getPartitionColumn() : null;
- LOGGER.info("Initialized RealtimeSegmentAssignment with replication: {} for table: {}", _replication,
- _realtimeTableName);
+ LOGGER.info("Initialized RealtimeSegmentAssignment with replication: {} partitionColumn: {} for table: {}",
Review comment:
nit: `with replication: {}, partitionColumn: {}`
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -358,7 +358,7 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ IN_PROGRESS, DONE, UPLOAD
Review comment:
Do you think if we need to have separated status for `UPLOAD_UPSERT` and `UPLOAD_APPEND`?
##########
File path: pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterIntegrationTestUtils.java
##########
@@ -285,7 +285,7 @@ public static void buildSegmentFromAvro(File avroFile, TableConfig tableConfig,
segmentGeneratorConfig.setOutDir(segmentDir.getPath());
segmentGeneratorConfig.setTableName(tableConfig.getTableName());
// Test segment with space and special character in the file name
- segmentGeneratorConfig.setSegmentNamePostfix(segmentIndex + " %");
+ segmentGeneratorConfig.setSegmentNamePostfix(Integer.toString(segmentIndex));
Review comment:
why change this?
##########
File path: pinot-integration-tests/src/test/resources/upsert_table_test.schema
##########
@@ -0,0 +1,33 @@
+{
+ "dimensionFieldSpecs": [
+ {
+ "dataType": "INT",
+ "singleValueField": true,
+ "name": "clientId"
+ },
+ {
+ "dataType": "STRING",
+ "singleValueField": true,
+ "name": "city"
+ },
+ {
+ "dataType": "STRING",
+ "singleValueField": true,
+ "name": "description"
+ },
+ {
+ "dataType": "INT",
+ "singleValueField": true,
+ "name": "salary"
+ }
+ ],
+ "timeFieldSpec": {
+ "incomingGranularitySpec": {
+ "timeType": "DAYS",
+ "dataType": "INT",
+ "name": "DaysSinceEpoch"
+ }
+ },
+ "primaryKeyColumns": ["clientId"],
+ "schemaName": "upsertSchema"
+}
Review comment:
nit: newline
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
+ if (_pinotHelixResourceManager.isRealtimeOnlyTable(rawTableName)) {
Review comment:
how to upload segments to a real-time table part for a hybrid table?
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUtarSegmentToLocal(String uri, File tableDataDir, String segmentName)
+ throws Exception {
+ File tempDir = new File(tableDataDir, "tmp-" + segmentName + "-" + UUID.randomUUID());
+ FileUtils.forceMkdir(tempDir);
+ File tempTarFile = new File(tempDir, segmentName + TAR_GZ_SUFFIX);
+ File tempSegmentDir = new File(tempDir, segmentName);
+ try {
+ try {
+ SegmentFetcherFactory.fetchSegmentToLocal(uri, tempTarFile);
+ LOGGER.info("Downloaded tarred segment: {} from: {} to: {}, file length: {}", segmentName, uri, tempTarFile,
+ tempTarFile.length());
+ } catch (AttemptsExceededException e) {
+ LOGGER.error("Attempts exceeded when downloading segment: {} : {} from: {} to: {}", segmentName, uri,
Review comment:
Log the error `LOGGER.error(..., e)`?
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
+ if (_pinotHelixResourceManager.isUpsertTable(rawTableName)) {
Review comment:
Shall we throw an exception on the non-upsert realtime table?
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUtarSegmentToLocal(String uri, File tableDataDir, String segmentName)
+ throws Exception {
+ File tempDir = new File(tableDataDir, "tmp-" + segmentName + "-" + UUID.randomUUID());
+ FileUtils.forceMkdir(tempDir);
+ File tempTarFile = new File(tempDir, segmentName + TAR_GZ_SUFFIX);
+ File tempSegmentDir = new File(tempDir, segmentName);
+ try {
+ try {
+ SegmentFetcherFactory.fetchSegmentToLocal(uri, tempTarFile);
+ LOGGER.info("Downloaded tarred segment: {} from: {} to: {}, file length: {}", segmentName, uri, tempTarFile,
+ tempTarFile.length());
+ } catch (AttemptsExceededException e) {
+ LOGGER.error("Attempts exceeded when downloading segment: {} : {} from: {} to: {}", segmentName, uri,
+ tempTarFile);
+ Utils.rethrowException(e);
+ }
+
+ try {
+ // If an exception is thrown when untarring, it means the tar file is broken OR not found after the retry.
+ // Thus, there's no need to retry again.
+ File tempIndexDir = TarGzCompressionUtils.untar(tempTarFile, tempSegmentDir).get(0);
+ File segmentDir = new File(tableDataDir, segmentName);
+ if (segmentDir.exists()) {
+ LOGGER.info("Deleting existing index directory for segment: {}", segmentName);
+ FileUtils.deleteDirectory(segmentDir);
+ }
+ FileUtils.moveDirectory(tempIndexDir, segmentDir);
+ LOGGER.info("Successfully downloaded segment: {} to: {}", segmentName, segmentDir);
+ } catch (Exception e) {
+ LOGGER
+ .error("Exception when untarring segment: {} for from {} to {}", segmentName, tempTarFile, tempSegmentDir);
Review comment:
nit: `LOGGER
.error("Exception when untarring segment: {} from {} to {}", segmentName, tempTarFile, tempSegmentDir, e);`
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
##########
@@ -528,4 +542,41 @@ private boolean isValid(Schema schema, IndexingConfig indexingConfig) {
}
return isValid;
}
+
+ private int getSegmentPartitionId(String segmentName, String tableName) {
+ // A fast path if the segmentName is a LLC segment name and we can get the partition id from the name directly.
+ if (LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
+ return new LLCSegmentName(segmentName).getPartitionId();
+ }
+ // Otherwise, retrieve the partition id from the segment zk metadata. Currently only realtime segments from upsert
+ // enabled tables have partition ids in their segment metadata.
+ RealtimeSegmentZKMetadata segmentZKMetadata =
+ ZKMetadataProvider.getRealtimeSegmentZKMetadata(_helixManager.getHelixPropertyStore(), tableName, segmentName);
+ Preconditions.checkState(isUpsertEnabled(),
+ "Only upsert enabled table has partition ids in its segment metadata: seg %s of table %s", segmentName,
+ tableName);
+ Preconditions
+ .checkState(segmentZKMetadata != null, "Failed to find segment ZK metadata for segment: %s of table: %s",
+ segmentName, tableName);
+ return getSegmentPartitionIdFromZkMetaData(segmentZKMetadata, tableName);
+ }
+
+ private int getSegmentPartitionIdFromZkMetaData(RealtimeSegmentZKMetadata segmentZKMetadata, String tableName) {
+ String segmentName = segmentZKMetadata.getSegmentName();
+ Preconditions.checkState(segmentZKMetadata.getPartitionMetadata() != null,
+ "Segment ZK metadata for segment: %s of table: %s does not contain partition metadata", segmentName,
+ tableName);
+
+ // Use any primary key column to fetch the partition metadata
+ ColumnPartitionMetadata partitionMetadata =
+ segmentZKMetadata.getPartitionMetadata().getColumnPartitionMap().get(_primaryKeyColumns.get(0));
+ Preconditions.checkState(partitionMetadata != null,
+ "Segment ZK metadata for segment: %s of table: %s does not contain partition metadata for column: %s",
+ segmentName, tableName, _primaryKeyColumns.get(0));
+ Set<Integer> partitions = partitionMetadata.getPartitions();
+ Preconditions.checkState(partitions.size() == 1,
+ "Segment ZK metadata for segment: %s of table: %s contains multiple partitions for column: %s", segmentName,
Review comment:
also dump all the partitions in the error log
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1649,63 +1651,100 @@ public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, Str
public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ String tableNameWithType;
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
- OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
- ZKMetadataUtils.updateSegmentMetadata(offlineSegmentZKMetadata, segmentMetadata);
- offlineSegmentZKMetadata.setDownloadUrl(downloadUrl);
- offlineSegmentZKMetadata.setCrypterName(crypter);
- offlineSegmentZKMetadata.setPushTime(System.currentTimeMillis());
+ ZNRecord znRecord;
+
+ if (isUpsertTable(tableName)) {
Review comment:
shall we also add a check and throw an exception for a REALTIME non-upsert table?
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1649,63 +1651,100 @@ public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, Str
public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ String tableNameWithType;
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
- OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
- ZKMetadataUtils.updateSegmentMetadata(offlineSegmentZKMetadata, segmentMetadata);
- offlineSegmentZKMetadata.setDownloadUrl(downloadUrl);
- offlineSegmentZKMetadata.setCrypterName(crypter);
- offlineSegmentZKMetadata.setPushTime(System.currentTimeMillis());
+ ZNRecord znRecord;
+
+ if (isUpsertTable(tableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ instancePartitionsType = InstancePartitionsType.CONSUMING;
+ // Build the realtime segment zk metadata with necessary fields.
+ LLCRealtimeSegmentZKMetadata segmentZKMetadata = new LLCRealtimeSegmentZKMetadata();
+ ZKMetadataUtils
+ .updateSegmentMetadata(segmentZKMetadata, segmentMetadata, CommonConstants.Segment.SegmentType.REALTIME);
+ segmentZKMetadata.setDownloadUrl(downloadUrl);
+ segmentZKMetadata.setCrypterName(crypter);
+ segmentZKMetadata.setStatus(CommonConstants.Segment.Realtime.Status.UPLOAD);
+ znRecord = segmentZKMetadata.toZNRecord();
+ } else {
+ tableNameWithType = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
+ instancePartitionsType = InstancePartitionsType.OFFLINE;
+ // Build the offline segment zk metadata with necessary fields.
+ OfflineSegmentZKMetadata segmentZKMetadata = new OfflineSegmentZKMetadata();
+ ZKMetadataUtils
+ .updateSegmentMetadata(segmentZKMetadata, segmentMetadata, CommonConstants.Segment.SegmentType.OFFLINE);
+ segmentZKMetadata.setDownloadUrl(downloadUrl);
+ segmentZKMetadata.setCrypterName(crypter);
+ segmentZKMetadata.setPushTime(System.currentTimeMillis());
+ znRecord = segmentZKMetadata.toZNRecord();
+ }
String segmentZKMetadataPath =
- ZKMetadataProvider.constructPropertyStorePathForSegment(offlineTableName, segmentName);
+ ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType, segmentName);
Preconditions.checkState(
- _propertyStore.set(segmentZKMetadataPath, offlineSegmentZKMetadata.toZNRecord(), AccessOption.PERSISTENT),
- "Failed to set segment ZK metadata for table: " + offlineTableName + ", segment: " + segmentName);
- LOGGER.info("Added segment: {} of table: {} to property store", segmentName, offlineTableName);
+ _propertyStore.set(segmentZKMetadataPath, znRecord, AccessOption.PERSISTENT),
+ "Failed to set segment ZK metadata for table: " + tableNameWithType + ", segment: " + segmentName);
+ LOGGER.info("Added segment: {} of table: {} to property store", segmentName, tableNameWithType);
+ assignTableSegment(tableNameWithType, segmentName, segmentZKMetadataPath, instancePartitionsType);
+ }
+
+ private void assignTableSegment(String tableNameWithType, String segmentName, String segmentZKMetadataPath,
+ InstancePartitionsType instancePartitionsType) {
// Assign instances for the segment and add it into IdealState
try {
- TableConfig offlineTableConfig = getTableConfig(offlineTableName);
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
Preconditions
- .checkState(offlineTableConfig != null, "Failed to find table config for table: " + offlineTableName);
+ .checkState(tableConfig != null, "Failed to find table config for table: " + tableNameWithType);
SegmentAssignment segmentAssignment =
- SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, offlineTableConfig);
+ SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, tableConfig);
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap = Collections
- .singletonMap(InstancePartitionsType.OFFLINE, InstancePartitionsUtils
- .fetchOrComputeInstancePartitions(_helixZkManager, offlineTableConfig, InstancePartitionsType.OFFLINE));
- synchronized (getTableUpdaterLock(offlineTableName)) {
- HelixHelper.updateIdealState(_helixZkManager, offlineTableName, idealState -> {
+ .singletonMap(instancePartitionsType, InstancePartitionsUtils
+ .fetchOrComputeInstancePartitions(_helixZkManager, tableConfig, instancePartitionsType));
+ synchronized (getTableUpdaterLock(tableNameWithType)) {
+ HelixHelper.updateIdealState(_helixZkManager, tableNameWithType, idealState -> {
assert idealState != null;
Map<String, Map<String, String>> currentAssignment = idealState.getRecord().getMapFields();
if (currentAssignment.containsKey(segmentName)) {
LOGGER.warn("Segment: {} already exists in the IdealState for table: {}, do not update", segmentName,
- offlineTableName);
+ tableNameWithType);
} else {
List<String> assignedInstances = segmentAssignment.assignSegment(segmentName, currentAssignment, instancePartitionsMap);
LOGGER.info("Assigning segment: {} to instances: {} for table: {}", segmentName, assignedInstances,
- offlineTableName);
- currentAssignment.put(segmentName, SegmentAssignmentUtils.getInstanceStateMap(assignedInstances, SegmentStateModel.ONLINE));
+ tableNameWithType);
+ currentAssignment.put(segmentName, SegmentAssignmentUtils
+ .getInstanceStateMap(assignedInstances, SegmentStateModel.ONLINE));
}
return idealState;
});
- LOGGER.info("Added segment: {} to IdealState for table: {}", segmentName, offlineTableName);
+ LOGGER.info("Added segment: {} to IdealState for table: {}", segmentName, tableNameWithType);
}
} catch (Exception e) {
LOGGER
.error("Caught exception while adding segment: {} to IdealState for table: {}, deleting segment ZK metadata",
- segmentName, offlineTableName, e);
+ segmentName, tableNameWithType, e);
if (_propertyStore.remove(segmentZKMetadataPath, AccessOption.PERSISTENT)) {
- LOGGER.info("Deleted segment ZK metadata for segment: {} of table: {}", segmentName, offlineTableName);
+ LOGGER.info("Deleted segment ZK metadata for segment: {} of table: {}", segmentName, tableNameWithType);
} else {
LOGGER
- .error("Failed to deleted segment ZK metadata for segment: {} of table: {}", segmentName, offlineTableName);
+ .error("Failed to deleted segment ZK metadata for segment: {} of table: {}", segmentName, tableNameWithType);
}
throw e;
}
}
+ public boolean isUpsertTable(String tableName) {
+ if (hasOfflineTable(tableName))
+ return false;
+ if (!hasRealtimeTable(tableName))
Review comment:
is this if-condition redundant?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r606493803
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -247,24 +248,35 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
+ if (tableType == TableType.OFFLINE) {
+ tableNameWithType = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ } else {
+ if (!_pinotHelixResourceManager.isUpsertTable(rawTableName)) {
+ throw new UnsupportedOperationException(
+ "Upload segment to non-upsert realtime table is not supported " + rawTableName);
+ }
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+ }
+
String clientAddress = InetAddress.getByName(request.getRemoteAddr()).getHostName();
LOGGER.info("Processing upload request for segment: {} of table: {} from client: {}, ingestion descriptor: {}",
- segmentName, offlineTableName, clientAddress, ingestionDescriptor);
+ segmentName, tableNameWithType, clientAddress, ingestionDescriptor);
- // Skip segment validation if upload only segment metadata
- if (uploadType != FileUploadDownloadClient.FileUploadType.METADATA) {
+ // Skip segment validation if upload is to an offline table and only segment metadata.
Review comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r596474328
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1649,63 +1651,100 @@ public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, Str
public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ String tableNameWithType;
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
- OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
- ZKMetadataUtils.updateSegmentMetadata(offlineSegmentZKMetadata, segmentMetadata);
- offlineSegmentZKMetadata.setDownloadUrl(downloadUrl);
- offlineSegmentZKMetadata.setCrypterName(crypter);
- offlineSegmentZKMetadata.setPushTime(System.currentTimeMillis());
+ ZNRecord znRecord;
+
+ if (isUpsertTable(tableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ instancePartitionsType = InstancePartitionsType.CONSUMING;
+ // Build the realtime segment zk metadata with necessary fields.
+ LLCRealtimeSegmentZKMetadata segmentZKMetadata = new LLCRealtimeSegmentZKMetadata();
+ ZKMetadataUtils
+ .updateSegmentMetadata(segmentZKMetadata, segmentMetadata, CommonConstants.Segment.SegmentType.REALTIME);
+ segmentZKMetadata.setDownloadUrl(downloadUrl);
+ segmentZKMetadata.setCrypterName(crypter);
+ segmentZKMetadata.setStatus(CommonConstants.Segment.Realtime.Status.UPLOAD);
+ znRecord = segmentZKMetadata.toZNRecord();
+ } else {
+ tableNameWithType = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
+ instancePartitionsType = InstancePartitionsType.OFFLINE;
+ // Build the offline segment zk metadata with necessary fields.
+ OfflineSegmentZKMetadata segmentZKMetadata = new OfflineSegmentZKMetadata();
+ ZKMetadataUtils
+ .updateSegmentMetadata(segmentZKMetadata, segmentMetadata, CommonConstants.Segment.SegmentType.OFFLINE);
+ segmentZKMetadata.setDownloadUrl(downloadUrl);
+ segmentZKMetadata.setCrypterName(crypter);
+ segmentZKMetadata.setPushTime(System.currentTimeMillis());
+ znRecord = segmentZKMetadata.toZNRecord();
+ }
String segmentZKMetadataPath =
- ZKMetadataProvider.constructPropertyStorePathForSegment(offlineTableName, segmentName);
+ ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType, segmentName);
Preconditions.checkState(
- _propertyStore.set(segmentZKMetadataPath, offlineSegmentZKMetadata.toZNRecord(), AccessOption.PERSISTENT),
- "Failed to set segment ZK metadata for table: " + offlineTableName + ", segment: " + segmentName);
- LOGGER.info("Added segment: {} of table: {} to property store", segmentName, offlineTableName);
+ _propertyStore.set(segmentZKMetadataPath, znRecord, AccessOption.PERSISTENT),
+ "Failed to set segment ZK metadata for table: " + tableNameWithType + ", segment: " + segmentName);
+ LOGGER.info("Added segment: {} of table: {} to property store", segmentName, tableNameWithType);
+ assignTableSegment(tableNameWithType, segmentName, segmentZKMetadataPath, instancePartitionsType);
+ }
+
+ private void assignTableSegment(String tableNameWithType, String segmentName, String segmentZKMetadataPath,
+ InstancePartitionsType instancePartitionsType) {
// Assign instances for the segment and add it into IdealState
try {
- TableConfig offlineTableConfig = getTableConfig(offlineTableName);
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
Preconditions
- .checkState(offlineTableConfig != null, "Failed to find table config for table: " + offlineTableName);
+ .checkState(tableConfig != null, "Failed to find table config for table: " + tableNameWithType);
SegmentAssignment segmentAssignment =
- SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, offlineTableConfig);
+ SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, tableConfig);
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap = Collections
- .singletonMap(InstancePartitionsType.OFFLINE, InstancePartitionsUtils
- .fetchOrComputeInstancePartitions(_helixZkManager, offlineTableConfig, InstancePartitionsType.OFFLINE));
- synchronized (getTableUpdaterLock(offlineTableName)) {
- HelixHelper.updateIdealState(_helixZkManager, offlineTableName, idealState -> {
+ .singletonMap(instancePartitionsType, InstancePartitionsUtils
+ .fetchOrComputeInstancePartitions(_helixZkManager, tableConfig, instancePartitionsType));
+ synchronized (getTableUpdaterLock(tableNameWithType)) {
+ HelixHelper.updateIdealState(_helixZkManager, tableNameWithType, idealState -> {
assert idealState != null;
Map<String, Map<String, String>> currentAssignment = idealState.getRecord().getMapFields();
if (currentAssignment.containsKey(segmentName)) {
LOGGER.warn("Segment: {} already exists in the IdealState for table: {}, do not update", segmentName,
- offlineTableName);
+ tableNameWithType);
} else {
List<String> assignedInstances = segmentAssignment.assignSegment(segmentName, currentAssignment, instancePartitionsMap);
LOGGER.info("Assigning segment: {} to instances: {} for table: {}", segmentName, assignedInstances,
- offlineTableName);
- currentAssignment.put(segmentName, SegmentAssignmentUtils.getInstanceStateMap(assignedInstances, SegmentStateModel.ONLINE));
+ tableNameWithType);
+ currentAssignment.put(segmentName, SegmentAssignmentUtils
+ .getInstanceStateMap(assignedInstances, SegmentStateModel.ONLINE));
}
return idealState;
});
- LOGGER.info("Added segment: {} to IdealState for table: {}", segmentName, offlineTableName);
+ LOGGER.info("Added segment: {} to IdealState for table: {}", segmentName, tableNameWithType);
}
} catch (Exception e) {
LOGGER
.error("Caught exception while adding segment: {} to IdealState for table: {}, deleting segment ZK metadata",
- segmentName, offlineTableName, e);
+ segmentName, tableNameWithType, e);
if (_propertyStore.remove(segmentZKMetadataPath, AccessOption.PERSISTENT)) {
- LOGGER.info("Deleted segment ZK metadata for segment: {} of table: {}", segmentName, offlineTableName);
+ LOGGER.info("Deleted segment ZK metadata for segment: {} of table: {}", segmentName, tableNameWithType);
} else {
LOGGER
- .error("Failed to deleted segment ZK metadata for segment: {} of table: {}", segmentName, offlineTableName);
+ .error("Failed to deleted segment ZK metadata for segment: {} of table: {}", segmentName, tableNameWithType);
}
throw e;
}
}
+ public boolean isUpsertTable(String tableName) {
+ if (hasOfflineTable(tableName))
+ return false;
+ if (!hasRealtimeTable(tableName))
Review comment:
It is not redundant because a table which does not exist may reach here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] Jackie-Jiang commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r606584788
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -383,7 +383,12 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ // Means the segment is not completed and still consuming stream data
+ IN_PROGRESS,
+ // Means the segment is completed and sealed by some Pinot server and uploaded
+ DONE,
+ // Means the segment is uploaded to a Pinot controller by an external party
+ UPLOADED
Review comment:
I feel `CONSUMING` and `COMPLETED` is more clear, and consistent with the term we used in the segment assignment strategy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r598020542
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUntarSegmentToLocal(String uri, File tableDataDir, String segmentName)
+ throws Exception {
+ File tempDir = new File(tableDataDir, "tmp-" + segmentName + "-" + UUID.randomUUID());
+ FileUtils.forceMkdir(tempDir);
+ File tempTarFile = new File(tempDir, segmentName + TAR_GZ_SUFFIX);
+ File tempSegmentDir = new File(tempDir, segmentName);
+ try {
+ try {
+ SegmentFetcherFactory.fetchSegmentToLocal(uri, tempTarFile);
+ LOGGER.info("Downloaded tarred segment: {} from: {} to: {}, file length: {}", segmentName, uri, tempTarFile,
+ tempTarFile.length());
+ } catch (AttemptsExceededException e) {
+ LOGGER.error("Attempts exceeded when downloading segment: {} from: {} to: {}", segmentName, uri,
+ tempTarFile, e);
+ Utils.rethrowException(e);
+ }
+
+ try {
+ // If an exception is thrown when untarring, it means the tar file is broken OR not found after the retry.
+ // Thus, there's no need to retry again.
+ File tempIndexDir = TarGzCompressionUtils.untar(tempTarFile, tempSegmentDir).get(0);
+ File segmentDir = new File(tableDataDir, segmentName);
+ if (segmentDir.exists()) {
+ LOGGER.info("Deleting existing index directory for segment: {}", segmentName);
+ FileUtils.deleteDirectory(segmentDir);
+ }
+ FileUtils.moveDirectory(tempIndexDir, segmentDir);
+ LOGGER.info("Successfully downloaded segment: {} to: {}", segmentName, segmentDir);
+ } catch (Exception e) {
+ LOGGER.error("Exception when untarring segment: {} for from {} to {}", segmentName, tempTarFile, tempSegmentDir,
Review comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-805240112
> > A new integration test UpsertTableSegmentUploadIntegrationTest is added to verify the uploaded segment is added to a realtime upsert enabled table correctly. The segment name used in the test does not follow llc segment convention but it is loaded successfully. The query result is good.
> > I do quite understand why we should "Exclude uploaded segment from the processing". During segment load, the upsert table metadata should be processed based on the loaded segment, right?
> > > Please take a look at `PinotLLCRealtimeSegmentManager` and ensure it works on uploaded segments:
> > >
> > > * Segment name for uploaded segment does not follow the llc segment name convention
> > > * Exclude uploaded segment from the processing
>
> I took another look at `PinotLLCRealtimeSegmentManager` and seems non-LLC segments are already excluded (because it might co-exist with HLC segments before). We cannot include them during metadata processing because it relies on the segment name format.
>
> For the issue within the `processExistingSegment()`, for uploading the new segment it won't hit that code path, but for refreshing the existing segment it won't set the fields properly.
>
> If possible, I would recommend setting up an actual table, upload some segments and keep it running for some time to ensure all components can properly work on the uploaded segments
The integration test I added uploads a segment to a LLC table with all Pinot components up. The test shows the query can be run successfully on a Pinot upsert table with correct query results returned. It also verifies the idealstate. I can add more verification to the test such as more metadata check and more segments uploads. I prefer Pinot integration tests than manual tests.
for `processExistingSegment()`, I think we should leave it to another PR and let this PR focus on adding new segments. For now, this PR will reject upload segments of the same name. Notice that for our Upsert table use case, upload new segments would be good enough. I will have a follow up PR for refreshing existing segments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r599979721
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUntarSegmentToLocal(String uri, File tableDataDir, String segmentName)
Review comment:
Done. Good catch. Use the existing downloadSegmentFromDeepstore() method which does the same function.
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -403,10 +414,13 @@ private void decryptFile(String crypterClassName, File tempEncryptedFile, File t
// it keeps it at the downloadURI header that is set. We will not support this endpoint going forward.
public void uploadSegmentAsJson(String segmentJsonStr,
@ApiParam(value = "Name of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_NAME) String tableName,
+ @ApiParam(value = "Type of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_TYPE) @DefaultValue("OFFLINE") String tableType,
@ApiParam(value = "Whether to enable parallel push protection") @DefaultValue("false") @QueryParam(FileUploadDownloadClient.QueryParameters.ENABLE_PARALLEL_PUSH_PROTECTION) boolean enableParallelPushProtection,
@Context HttpHeaders headers, @Context Request request, @Suspended final AsyncResponse asyncResponse) {
try {
- asyncResponse.resume(uploadSegment(tableName, null, enableParallelPushProtection, headers, request, false));
+ asyncResponse.resume(
+ uploadSegment(tableName, "OFFLINE".equalsIgnoreCase(tableType) ? TableType.OFFLINE : TableType.REALTIME, null,
Review comment:
done.
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java
##########
@@ -19,6 +19,7 @@
package org.apache.pinot.controller.helix.core.assignment.segment;
import com.google.common.base.Preconditions;
+
Review comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r597438938
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -358,7 +358,7 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ IN_PROGRESS, DONE, UPLOAD
Review comment:
Docs added. I will update the design with this additional status too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r611862455
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
##########
@@ -57,41 +61,44 @@ public ZKOperator(PinotHelixResourceManager pinotHelixResourceManager, Controlle
_controllerMetrics = controllerMetrics;
}
- public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata,
+ public void completeSegmentOperations(String tableNameWithType, SegmentMetadata segmentMetadata,
URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection,
HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter)
throws Exception {
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
String segmentName = segmentMetadata.getName();
-
- // Brand new segment, not refresh, directly add the segment
- ZNRecord segmentMetadataZnRecord =
- _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName);
+ ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType, segmentName);
if (segmentMetadataZnRecord == null) {
- LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName);
+ LOGGER.info("Adding new segment {} from table {}", segmentName, tableNameWithType);
processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter,
- rawTableName, segmentName, moveSegmentToFinalLocation);
+ tableNameWithType, segmentName, moveSegmentToFinalLocation);
return;
}
- LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, rawTableName);
+ // TODO Allow segment refreshing for realtime tables.
+ if (TableNameBuilder.isRealtimeTableResource(tableNameWithType)) {
+ LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, tableNameWithType);
Review comment:
Log line is misleading right? We are actually throwing an exception below. I think this log line belongs after line 84. It may be useful to have another log line here that says we do not support refresing existing segment for realtime table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r611857537
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -383,7 +383,12 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ // Means the segment is not completed and still consuming stream data
+ IN_PROGRESS,
+ // Means the segment is completed and sealed by some Pinot server and uploaded
+ DONE,
+ // Means the segment is uploaded to a Pinot controller by an external party
+ UPLOADED
Review comment:
@Jackie-Jiang agreed, just that it will need a migration of the existing metadata and/or code to scan both. Best to keep it out of this PR. @chenboat will be great if you can file an issue for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-804394441
A new integration test UpsertTableSegmentUploadIntegrationTest is added to verify the uploaded segment is added to a realtime upsert enabled table correctly. The segment name used in the test does not follow llc segment convention but it is loaded successfully. The query result is good.
I do quite understand why we should "Exclude uploaded segment from the processing". During segment load, the upsert table metadata should be processed based on the loaded segment, right?
> Please take a look at `PinotLLCRealtimeSegmentManager` and ensure it works on uploaded segments:
>
> * Segment name for uploaded segment does not follow the llc segment name convention
> * Exclude uploaded segment from the processing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r603451458
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1642,70 +1644,106 @@ public PinotResourceManagerResponse toggleTableState(String tableNameWithType, S
return instanceSet;
}
- public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl) {
- addNewSegment(tableName, segmentMetadata, downloadUrl, null);
+ public void addNewSegment(String tableNameWithType, SegmentMetadata segmentMetadata, String downloadUrl) {
+ addNewSegment(tableNameWithType, segmentMetadata, downloadUrl, null);
}
- public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
+ public void addNewSegment(String tableNameWithType, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
- OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
- ZKMetadataUtils.updateSegmentMetadata(offlineSegmentZKMetadata, segmentMetadata);
- offlineSegmentZKMetadata.setDownloadUrl(downloadUrl);
- offlineSegmentZKMetadata.setCrypterName(crypter);
- offlineSegmentZKMetadata.setPushTime(System.currentTimeMillis());
+ ZNRecord znRecord;
+ if (TableNameBuilder.isRealtimeTableResource(tableNameWithType)) {
+ Preconditions.checkState(isUpsertTable(tableNameWithType),
+ "Upload segment " + segmentName + " for non upsert enabled realtime table " + tableNameWithType
+ + " is not supported");
+ instancePartitionsType = InstancePartitionsType.CONSUMING;
+ // Build the realtime segment zk metadata with necessary fields.
+ LLCRealtimeSegmentZKMetadata segmentZKMetadata = new LLCRealtimeSegmentZKMetadata();
+ ZKMetadataUtils
+ .updateSegmentMetadata(segmentZKMetadata, segmentMetadata, CommonConstants.Segment.SegmentType.REALTIME);
+ segmentZKMetadata.setDownloadUrl(downloadUrl);
+ segmentZKMetadata.setCrypterName(crypter);
+ segmentZKMetadata.setStatus(CommonConstants.Segment.Realtime.Status.UPLOADED);
+ znRecord = segmentZKMetadata.toZNRecord();
+ } else {
+ instancePartitionsType = InstancePartitionsType.OFFLINE;
+ // Build the offline segment zk metadata with necessary fields.
+ OfflineSegmentZKMetadata segmentZKMetadata = new OfflineSegmentZKMetadata();
+ ZKMetadataUtils
+ .updateSegmentMetadata(segmentZKMetadata, segmentMetadata, CommonConstants.Segment.SegmentType.OFFLINE);
+ segmentZKMetadata.setDownloadUrl(downloadUrl);
+ segmentZKMetadata.setCrypterName(crypter);
+ segmentZKMetadata.setPushTime(System.currentTimeMillis());
+ znRecord = segmentZKMetadata.toZNRecord();
+ }
String segmentZKMetadataPath =
- ZKMetadataProvider.constructPropertyStorePathForSegment(offlineTableName, segmentName);
+ ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType, segmentName);
Preconditions.checkState(
- _propertyStore.set(segmentZKMetadataPath, offlineSegmentZKMetadata.toZNRecord(), AccessOption.PERSISTENT),
- "Failed to set segment ZK metadata for table: " + offlineTableName + ", segment: " + segmentName);
- LOGGER.info("Added segment: {} of table: {} to property store", segmentName, offlineTableName);
+ _propertyStore.set(segmentZKMetadataPath, znRecord, AccessOption.PERSISTENT),
+ "Failed to set segment ZK metadata for table: " + tableNameWithType + ", segment: " + segmentName);
+ LOGGER.info("Added segment: {} of table: {} to property store", segmentName, tableNameWithType);
+ assignTableSegment(tableNameWithType, segmentName, segmentZKMetadataPath, instancePartitionsType);
+ }
+
+ private void assignTableSegment(String tableNameWithType, String segmentName, String segmentZKMetadataPath,
+ InstancePartitionsType instancePartitionsType) {
// Assign instances for the segment and add it into IdealState
try {
- TableConfig offlineTableConfig = getTableConfig(offlineTableName);
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
Preconditions
- .checkState(offlineTableConfig != null, "Failed to find table config for table: " + offlineTableName);
+ .checkState(tableConfig != null, "Failed to find table config for table: " + tableNameWithType);
SegmentAssignment segmentAssignment =
- SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, offlineTableConfig);
+ SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, tableConfig);
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap = Collections
- .singletonMap(InstancePartitionsType.OFFLINE, InstancePartitionsUtils
- .fetchOrComputeInstancePartitions(_helixZkManager, offlineTableConfig, InstancePartitionsType.OFFLINE));
- synchronized (getTableUpdaterLock(offlineTableName)) {
- HelixHelper.updateIdealState(_helixZkManager, offlineTableName, idealState -> {
+ .singletonMap(instancePartitionsType, InstancePartitionsUtils
+ .fetchOrComputeInstancePartitions(_helixZkManager, tableConfig, instancePartitionsType));
+ synchronized (getTableUpdaterLock(tableNameWithType)) {
+ HelixHelper.updateIdealState(_helixZkManager, tableNameWithType, idealState -> {
assert idealState != null;
Map<String, Map<String, String>> currentAssignment = idealState.getRecord().getMapFields();
if (currentAssignment.containsKey(segmentName)) {
LOGGER.warn("Segment: {} already exists in the IdealState for table: {}, do not update", segmentName,
- offlineTableName);
+ tableNameWithType);
} else {
List<String> assignedInstances = segmentAssignment.assignSegment(segmentName, currentAssignment, instancePartitionsMap);
LOGGER.info("Assigning segment: {} to instances: {} for table: {}", segmentName, assignedInstances,
- offlineTableName);
- currentAssignment.put(segmentName, SegmentAssignmentUtils.getInstanceStateMap(assignedInstances, SegmentStateModel.ONLINE));
+ tableNameWithType);
+ currentAssignment.put(segmentName, SegmentAssignmentUtils
+ .getInstanceStateMap(assignedInstances, SegmentStateModel.ONLINE));
}
return idealState;
});
- LOGGER.info("Added segment: {} to IdealState for table: {}", segmentName, offlineTableName);
+ LOGGER.info("Added segment: {} to IdealState for table: {}", segmentName, tableNameWithType);
}
} catch (Exception e) {
LOGGER
.error("Caught exception while adding segment: {} to IdealState for table: {}, deleting segment ZK metadata",
- segmentName, offlineTableName, e);
+ segmentName, tableNameWithType, e);
if (_propertyStore.remove(segmentZKMetadataPath, AccessOption.PERSISTENT)) {
- LOGGER.info("Deleted segment ZK metadata for segment: {} of table: {}", segmentName, offlineTableName);
+ LOGGER.info("Deleted segment ZK metadata for segment: {} of table: {}", segmentName, tableNameWithType);
} else {
LOGGER
- .error("Failed to deleted segment ZK metadata for segment: {} of table: {}", segmentName, offlineTableName);
+ .error("Failed to deleted segment ZK metadata for segment: {} of table: {}", segmentName, tableNameWithType);
}
throw e;
}
}
+ public boolean isUpsertTable(String tableName) {
+ if (hasOfflineTable(tableName))
+ return false;
+ if (!hasRealtimeTable(tableName))
+ return false;
+
+ UpsertConfig upsertConfig =
Review comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] Jackie-Jiang commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r573404392
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
Review comment:
The `tableName` passed in might be with type suffix. If it already has type embedded, directly use that type.
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
Review comment:
Since we are going to allow uploading segments to both offline and real-time tables, we should use `tableNameWithType` in methods such as `completeZkOperations` etc to avoid identifying the table type again
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1649,63 +1649,81 @@ public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, Str
public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ String tableNameWithType;
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
ZKMetadataUtils.updateSegmentMetadata(offlineSegmentZKMetadata, segmentMetadata);
offlineSegmentZKMetadata.setDownloadUrl(downloadUrl);
offlineSegmentZKMetadata.setCrypterName(crypter);
offlineSegmentZKMetadata.setPushTime(System.currentTimeMillis());
+
+ if (isRealtimeOnlyTable(tableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ instancePartitionsType = InstancePartitionsType.CONSUMING;
Review comment:
This should be COMPLETED
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] Jackie-Jiang commented on a change in pull request #6567: Enable adding new segments. to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r599978452
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java
##########
@@ -551,6 +554,30 @@ public SimpleHttpResponse uploadSegment(URI uri, String segmentName, File segmen
return uploadSegment(uri, segmentName, segmentFile, null, parameters, DEFAULT_SOCKET_TIMEOUT_MS);
}
+ /**
+ * Upload segment with segment file using default settings. Include table name and type as a request parameters.
+ *
+ * @param uri URI
+ * @param segmentName Segment name
+ * @param segmentFile Segment file
+ * @param tableName Table name with or without type suffix
+ * @param tableType Table type
+ * @return Response
+ * @throws IOException
+ * @throws HttpErrorStatusException
+ */
+ public SimpleHttpResponse uploadSegment(URI uri, String segmentName, File segmentFile, String tableName,
+ TableType tableType)
+ throws IOException, HttpErrorStatusException {
+ // Add table name and type request parameters
+ NameValuePair tableNameValuePair = new BasicNameValuePair(QueryParameters.TABLE_NAME, tableName);
+ NameValuePair tableTypeValuePair = new BasicNameValuePair(QueryParameters.TABLE_TYPE, tableType.name());
+ List<NameValuePair> parameters = new ArrayList<>();
Review comment:
(nit) Can be simplified to `Arrays.asList(tableNameValuePair, tableTypeValuePair)`
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -247,24 +248,35 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
+ if (tableType == TableType.OFFLINE) {
+ tableNameWithType = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ } else {
+ if (!_pinotHelixResourceManager.isUpsertTable(rawTableName)) {
+ throw new UnsupportedOperationException(
+ "Upload segment to non-upsert realtime table is not supported " + rawTableName);
+ }
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+ }
+
String clientAddress = InetAddress.getByName(request.getRemoteAddr()).getHostName();
LOGGER.info("Processing upload request for segment: {} of table: {} from client: {}, ingestion descriptor: {}",
- segmentName, offlineTableName, clientAddress, ingestionDescriptor);
+ segmentName, tableNameWithType, clientAddress, ingestionDescriptor);
- // Skip segment validation if upload only segment metadata
- if (uploadType != FileUploadDownloadClient.FileUploadType.METADATA) {
+ // Skip segment validation if upload only segment metadata or it is a realtime table segment.
+ if (uploadType != FileUploadDownloadClient.FileUploadType.METADATA && TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
Review comment:
(nit)
```suggestion
if (tableType == TableType.OFFLINE && uploadType != FileUploadDownloadClient.FileUploadType.METADATA) {
```
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -25,10 +25,16 @@
import java.util.List;
import java.util.Map;
+import java.util.UUID;
Review comment:
Revert the changes to this file
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/ZKMetadataUtils.java
##########
@@ -35,24 +33,24 @@
private ZKMetadataUtils() {
}
- public static void updateSegmentMetadata(OfflineSegmentZKMetadata offlineSegmentZKMetadata,
- SegmentMetadata segmentMetadata) {
- offlineSegmentZKMetadata.setSegmentName(segmentMetadata.getName());
- offlineSegmentZKMetadata.setTableName(segmentMetadata.getTableName());
- offlineSegmentZKMetadata.setIndexVersion(segmentMetadata.getVersion());
- offlineSegmentZKMetadata.setSegmentType(SegmentType.OFFLINE);
+ public static void updateSegmentMetadata(SegmentZKMetadata segmentZKMetadata,
+ SegmentMetadata segmentMetadata, SegmentType segmentType) {
Review comment:
Reformat
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1642,70 +1644,106 @@ public PinotResourceManagerResponse toggleTableState(String tableNameWithType, S
return instanceSet;
}
- public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl) {
- addNewSegment(tableName, segmentMetadata, downloadUrl, null);
+ public void addNewSegment(String tableNameWithType, SegmentMetadata segmentMetadata, String downloadUrl) {
+ addNewSegment(tableNameWithType, segmentMetadata, downloadUrl, null);
}
- public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
+ public void addNewSegment(String tableNameWithType, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
- OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
- ZKMetadataUtils.updateSegmentMetadata(offlineSegmentZKMetadata, segmentMetadata);
- offlineSegmentZKMetadata.setDownloadUrl(downloadUrl);
- offlineSegmentZKMetadata.setCrypterName(crypter);
- offlineSegmentZKMetadata.setPushTime(System.currentTimeMillis());
+ ZNRecord znRecord;
+ if (TableNameBuilder.isRealtimeTableResource(tableNameWithType)) {
+ Preconditions.checkState(isUpsertTable(tableNameWithType),
+ "Upload segment " + segmentName + " for non upsert enabled realtime table " + tableNameWithType
+ + " is not supported");
+ instancePartitionsType = InstancePartitionsType.CONSUMING;
+ // Build the realtime segment zk metadata with necessary fields.
+ LLCRealtimeSegmentZKMetadata segmentZKMetadata = new LLCRealtimeSegmentZKMetadata();
+ ZKMetadataUtils
+ .updateSegmentMetadata(segmentZKMetadata, segmentMetadata, CommonConstants.Segment.SegmentType.REALTIME);
+ segmentZKMetadata.setDownloadUrl(downloadUrl);
+ segmentZKMetadata.setCrypterName(crypter);
+ segmentZKMetadata.setStatus(CommonConstants.Segment.Realtime.Status.UPLOADED);
+ znRecord = segmentZKMetadata.toZNRecord();
+ } else {
+ instancePartitionsType = InstancePartitionsType.OFFLINE;
+ // Build the offline segment zk metadata with necessary fields.
+ OfflineSegmentZKMetadata segmentZKMetadata = new OfflineSegmentZKMetadata();
+ ZKMetadataUtils
+ .updateSegmentMetadata(segmentZKMetadata, segmentMetadata, CommonConstants.Segment.SegmentType.OFFLINE);
+ segmentZKMetadata.setDownloadUrl(downloadUrl);
+ segmentZKMetadata.setCrypterName(crypter);
+ segmentZKMetadata.setPushTime(System.currentTimeMillis());
+ znRecord = segmentZKMetadata.toZNRecord();
+ }
String segmentZKMetadataPath =
- ZKMetadataProvider.constructPropertyStorePathForSegment(offlineTableName, segmentName);
+ ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType, segmentName);
Preconditions.checkState(
- _propertyStore.set(segmentZKMetadataPath, offlineSegmentZKMetadata.toZNRecord(), AccessOption.PERSISTENT),
- "Failed to set segment ZK metadata for table: " + offlineTableName + ", segment: " + segmentName);
- LOGGER.info("Added segment: {} of table: {} to property store", segmentName, offlineTableName);
+ _propertyStore.set(segmentZKMetadataPath, znRecord, AccessOption.PERSISTENT),
+ "Failed to set segment ZK metadata for table: " + tableNameWithType + ", segment: " + segmentName);
+ LOGGER.info("Added segment: {} of table: {} to property store", segmentName, tableNameWithType);
+ assignTableSegment(tableNameWithType, segmentName, segmentZKMetadataPath, instancePartitionsType);
+ }
+
+ private void assignTableSegment(String tableNameWithType, String segmentName, String segmentZKMetadataPath,
+ InstancePartitionsType instancePartitionsType) {
// Assign instances for the segment and add it into IdealState
try {
- TableConfig offlineTableConfig = getTableConfig(offlineTableName);
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
Preconditions
- .checkState(offlineTableConfig != null, "Failed to find table config for table: " + offlineTableName);
+ .checkState(tableConfig != null, "Failed to find table config for table: " + tableNameWithType);
SegmentAssignment segmentAssignment =
- SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, offlineTableConfig);
+ SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, tableConfig);
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap = Collections
- .singletonMap(InstancePartitionsType.OFFLINE, InstancePartitionsUtils
- .fetchOrComputeInstancePartitions(_helixZkManager, offlineTableConfig, InstancePartitionsType.OFFLINE));
- synchronized (getTableUpdaterLock(offlineTableName)) {
- HelixHelper.updateIdealState(_helixZkManager, offlineTableName, idealState -> {
+ .singletonMap(instancePartitionsType, InstancePartitionsUtils
+ .fetchOrComputeInstancePartitions(_helixZkManager, tableConfig, instancePartitionsType));
+ synchronized (getTableUpdaterLock(tableNameWithType)) {
+ HelixHelper.updateIdealState(_helixZkManager, tableNameWithType, idealState -> {
assert idealState != null;
Map<String, Map<String, String>> currentAssignment = idealState.getRecord().getMapFields();
if (currentAssignment.containsKey(segmentName)) {
LOGGER.warn("Segment: {} already exists in the IdealState for table: {}, do not update", segmentName,
- offlineTableName);
+ tableNameWithType);
} else {
List<String> assignedInstances = segmentAssignment.assignSegment(segmentName, currentAssignment, instancePartitionsMap);
LOGGER.info("Assigning segment: {} to instances: {} for table: {}", segmentName, assignedInstances,
- offlineTableName);
- currentAssignment.put(segmentName, SegmentAssignmentUtils.getInstanceStateMap(assignedInstances, SegmentStateModel.ONLINE));
+ tableNameWithType);
+ currentAssignment.put(segmentName, SegmentAssignmentUtils
+ .getInstanceStateMap(assignedInstances, SegmentStateModel.ONLINE));
}
return idealState;
});
- LOGGER.info("Added segment: {} to IdealState for table: {}", segmentName, offlineTableName);
+ LOGGER.info("Added segment: {} to IdealState for table: {}", segmentName, tableNameWithType);
}
} catch (Exception e) {
LOGGER
.error("Caught exception while adding segment: {} to IdealState for table: {}, deleting segment ZK metadata",
- segmentName, offlineTableName, e);
+ segmentName, tableNameWithType, e);
if (_propertyStore.remove(segmentZKMetadataPath, AccessOption.PERSISTENT)) {
- LOGGER.info("Deleted segment ZK metadata for segment: {} of table: {}", segmentName, offlineTableName);
+ LOGGER.info("Deleted segment ZK metadata for segment: {} of table: {}", segmentName, tableNameWithType);
} else {
LOGGER
- .error("Failed to deleted segment ZK metadata for segment: {} of table: {}", segmentName, offlineTableName);
+ .error("Failed to deleted segment ZK metadata for segment: {} of table: {}", segmentName, tableNameWithType);
}
throw e;
}
}
+ public boolean isUpsertTable(String tableName) {
+ if (hasOfflineTable(tableName))
+ return false;
+ if (!hasRealtimeTable(tableName))
+ return false;
+
+ UpsertConfig upsertConfig =
Review comment:
Check if table config is `null` before calling `getUpsertConfig()`.
I think we can skip the table existence check. We should not return `false` when offline table exist because we will still enable upsert even though it might not work. The table config null check can check the existence of the realtime table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-776953396
> How do we manage the stream offset if we upload a segment to the realtime table? Also, how about the segment name format?
> We need to have an integration test to ensure the `PinotLLCRealtimeSegmentManager` can work properly on uploaded segments.
Please refer to the comments I added with respect to Subbu's questions. For upsert tables, there should be validation that the offset of uploaded segments will not affect the consuming segment's offset. This is inline with the goals of segment upload — which are mainly about bootstrapping historical data.
The segment name format I propose conforms to the LLC name format with the following parts. tableName(string)_partitionId_(int)_sequenceNumber(int)_creationTime(string). The main difference is that the creationTime string now does not need to an actual timestamp but rather any unique string (e.g., timestamp + an uuid).
Yes. I will add an integration test.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r595471260
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUtarSegmentToLocal(String uri, File tableDataDir, String segmentName)
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] snleee commented on pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
snleee commented on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-776427097
@chenboat Can you open up the access for the linked design doc?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-807806592
> For the new integration test, it would be good if we push some events into kafka as well and make the commit threshold low so that we have CONSUMING, DONE, UPLOADED segments at the same time. Also, we can run the `RealtimeSegmentValidationManager` and see if it handles UPLOADED segments properly
Added the Kafka data ingestion to the integration test. Also add more segments (now 2 realtime segments and 3 uploaded segments are involved in the tests). Discovered a bug in the previous version. I will check if there is a way to test the `RealtimeSegmentValidationManager`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r611872581
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
##########
@@ -57,41 +61,44 @@ public ZKOperator(PinotHelixResourceManager pinotHelixResourceManager, Controlle
_controllerMetrics = controllerMetrics;
}
- public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata,
+ public void completeSegmentOperations(String tableNameWithType, SegmentMetadata segmentMetadata,
URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection,
HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter)
throws Exception {
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
String segmentName = segmentMetadata.getName();
-
- // Brand new segment, not refresh, directly add the segment
- ZNRecord segmentMetadataZnRecord =
- _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName);
+ ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType, segmentName);
if (segmentMetadataZnRecord == null) {
- LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName);
+ LOGGER.info("Adding new segment {} from table {}", segmentName, tableNameWithType);
processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter,
- rawTableName, segmentName, moveSegmentToFinalLocation);
+ tableNameWithType, segmentName, moveSegmentToFinalLocation);
return;
}
- LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, rawTableName);
+ // TODO Allow segment refreshing for realtime tables.
+ if (TableNameBuilder.isRealtimeTableResource(tableNameWithType)) {
+ LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, tableNameWithType);
Review comment:
Moved the log line after the if block.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat merged pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat merged pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r595469435
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
##########
@@ -61,12 +61,15 @@ public void completeSegmentOperations(String rawTableName, SegmentMetadata segme
URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection,
HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter)
throws Exception {
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
String segmentName = segmentMetadata.getName();
- // Brand new segment, not refresh, directly add the segment
- ZNRecord segmentMetadataZnRecord =
- _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName);
+ if (_pinotHelixResourceManager.isRealtimeOnlyTable(rawTableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+ } else {
+ tableNameWithType= TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ }
+ ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType, segmentName);
if (segmentMetadataZnRecord == null) {
LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName);
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r595470662
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1649,63 +1649,81 @@ public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, Str
public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ String tableNameWithType;
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
Review comment:
InstancePartitionsType is shared between offline table and realtime table segment assignment. So it is put here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat edited a comment on pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat edited a comment on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-807806592
> For the new integration test, it would be good if we push some events into kafka as well and make the commit threshold low so that we have CONSUMING, DONE, UPLOADED segments at the same time. Also, we can run the `RealtimeSegmentValidationManager` and see if it handles UPLOADED segments properly
Added the Kafka data ingestion to the integration test. Also add more segments (now 2 realtime segments and 3 uploaded segments are involved in the tests). Discovered a bug in the previous version.
Revised the integration test so that it runs the `RealtimeSegmentValidationManager` every 1 second and verifies the query results and idealstate are still valid. @Jackie-Jiang
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r573984359
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
+ if (_pinotHelixResourceManager.isRealtimeOnlyTable(rawTableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+ } else {
+ tableNameWithType = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ }
String clientAddress = InetAddress.getByName(request.getRemoteAddr()).getHostName();
LOGGER.info("Processing upload request for segment: {} of table: {} from client: {}, ingestion descriptor: {}",
- segmentName, offlineTableName, clientAddress, ingestionDescriptor);
+ segmentName, tableNameWithType, clientAddress, ingestionDescriptor);
- // Skip segment validation if upload only segment metadata
- if (uploadType != FileUploadDownloadClient.FileUploadType.METADATA) {
+ // Skip segment validation if upload only segment metadata or it is a realtime table segment.
+ // TODO Perform a validation check for realtime segments too.
Review comment:
For normal Pinot realtime tables, yes, the validation check on the uploaded segment should be done for multiple aspects as you mentioned. Otherwise, problems like duplicated records will arise.
For upsert enabled Pinot realtime tables (which is the focus of this work), many issues (especially for duplication) have been resolved already by upsertManager bookeeping. Note that the upsertManager will perform conflict resolution of records of the same primary key. There is still some work to be done on the segment validation. The main one I think is about being careful about not letting the upload segment change the stream consumed offset.
So overall, I think the realtime table segment upload can be enabled for upsert tables first. The support for upload for normal realtime tables can be added after a more complete validation is ready.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r596469843
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
##########
@@ -528,4 +542,41 @@ private boolean isValid(Schema schema, IndexingConfig indexingConfig) {
}
return isValid;
}
+
+ private int getSegmentPartitionId(String segmentName, String tableName) {
+ // A fast path if the segmentName is a LLC segment name and we can get the partition id from the name directly.
+ if (LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
+ return new LLCSegmentName(segmentName).getPartitionId();
+ }
+ // Otherwise, retrieve the partition id from the segment zk metadata. Currently only realtime segments from upsert
+ // enabled tables have partition ids in their segment metadata.
+ RealtimeSegmentZKMetadata segmentZKMetadata =
+ ZKMetadataProvider.getRealtimeSegmentZKMetadata(_helixManager.getHelixPropertyStore(), tableName, segmentName);
+ Preconditions.checkState(isUpsertEnabled(),
+ "Only upsert enabled table has partition ids in its segment metadata: seg %s of table %s", segmentName,
+ tableName);
+ Preconditions
+ .checkState(segmentZKMetadata != null, "Failed to find segment ZK metadata for segment: %s of table: %s",
+ segmentName, tableName);
+ return getSegmentPartitionIdFromZkMetaData(segmentZKMetadata, tableName);
+ }
+
+ private int getSegmentPartitionIdFromZkMetaData(RealtimeSegmentZKMetadata segmentZKMetadata, String tableName) {
+ String segmentName = segmentZKMetadata.getSegmentName();
+ Preconditions.checkState(segmentZKMetadata.getPartitionMetadata() != null,
+ "Segment ZK metadata for segment: %s of table: %s does not contain partition metadata", segmentName,
+ tableName);
+
+ // Use any primary key column to fetch the partition metadata
+ ColumnPartitionMetadata partitionMetadata =
+ segmentZKMetadata.getPartitionMetadata().getColumnPartitionMap().get(_primaryKeyColumns.get(0));
+ Preconditions.checkState(partitionMetadata != null,
+ "Segment ZK metadata for segment: %s of table: %s does not contain partition metadata for column: %s",
+ segmentName, tableName, _primaryKeyColumns.get(0));
+ Set<Integer> partitions = partitionMetadata.getPartitions();
+ Preconditions.checkState(partitions.size() == 1,
+ "Segment ZK metadata for segment: %s of table: %s contains multiple partitions for column: %s", segmentName,
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r606531151
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -383,7 +383,12 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ // Means the segment is not completed and still consuming stream data
+ IN_PROGRESS,
+ // Means the segment is completed and sealed by some Pinot server and uploaded
+ DONE,
+ // Means the segment is uploaded to a Pinot controller by an external party
+ UPLOADED
Review comment:
You should only add enums at the end. But anyways, this is a deeper change. I was just asking of the other reviewers. Let us revert it and stick to your PR for now.
@npawar and @Jackie-Jiang may have an opinion. We can discuss separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] codecov-io edited a comment on pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-776350585
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=h1) Report
> Merging [#6567](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=desc) (c625489) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/1beaab59b73f26c4e35f3b9bc856b03806cddf5a?el=desc) (1beaab5) will **decrease** coverage by `22.68%`.
> The diff coverage is `46.55%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #6567 +/- ##
===========================================
- Coverage 66.44% 43.76% -22.69%
===========================================
Files 1075 1391 +316
Lines 54773 67691 +12918
Branches 8168 9814 +1646
===========================================
- Hits 36396 29622 -6774
- Misses 15700 35589 +19889
+ Partials 2677 2480 -197
```
| Flag | Coverage Δ | |
|---|---|---|
| integration | `43.76% <46.55%> (?)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...ot/broker/broker/AllowAllAccessControlFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL0FsbG93QWxsQWNjZXNzQ29udHJvbEZhY3RvcnkuamF2YQ==) | `100.00% <ø> (ø)` | |
| [...t/broker/broker/BasicAuthAccessControlFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL0Jhc2ljQXV0aEFjY2Vzc0NvbnRyb2xGYWN0b3J5LmphdmE=) | `0.00% <0.00%> (ø)` | |
| [.../helix/BrokerUserDefinedMessageHandlerFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL2hlbGl4L0Jyb2tlclVzZXJEZWZpbmVkTWVzc2FnZUhhbmRsZXJGYWN0b3J5LmphdmE=) | `52.83% <0.00%> (-13.84%)` | :arrow_down: |
| [...org/apache/pinot/broker/queryquota/HitCounter.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9IaXRDb3VudGVyLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...che/pinot/broker/queryquota/MaxHitRateTracker.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9NYXhIaXRSYXRlVHJhY2tlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...ache/pinot/broker/queryquota/QueryQuotaEntity.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9RdWVyeVF1b3RhRW50aXR5LmphdmE=) | `0.00% <0.00%> (-50.00%)` | :arrow_down: |
| [...ker/routing/instanceselector/InstanceSelector.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9pbnN0YW5jZXNlbGVjdG9yL0luc3RhbmNlU2VsZWN0b3IuamF2YQ==) | `100.00% <ø> (ø)` | |
| [...roker/routing/segmentpruner/TimeSegmentPruner.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL1RpbWVTZWdtZW50UHJ1bmVyLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...roker/routing/segmentpruner/interval/Interval.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL2ludGVydmFsL0ludGVydmFsLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...r/routing/segmentpruner/interval/IntervalTree.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL2ludGVydmFsL0ludGVydmFsVHJlZS5qYXZh) | `0.00% <0.00%> (ø)` | |
| ... and [1417 more](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=footer). Last update [53c2a3c...c625489](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] codecov-io edited a comment on pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-776350585
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=h1) Report
> Merging [#6567](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=desc) (a3ed69f) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/1beaab59b73f26c4e35f3b9bc856b03806cddf5a?el=desc) (1beaab5) will **increase** coverage by `7.29%`.
> The diff coverage is `80.62%`.
> :exclamation: Current head a3ed69f differs from pull request most recent head ab9c377. Consider uploading reports for the commit ab9c377 to get more accurate results
[](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #6567 +/- ##
==========================================
+ Coverage 66.44% 73.74% +7.29%
==========================================
Files 1075 1390 +315
Lines 54773 67633 +12860
Branches 8168 9809 +1641
==========================================
+ Hits 36396 49876 +13480
+ Misses 15700 14497 -1203
- Partials 2677 3260 +583
```
| Flag | Coverage Δ | |
|---|---|---|
| integration | `43.83% <46.46%> (?)` | |
| unittests | `65.86% <62.64%> (?)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...ot/broker/broker/AllowAllAccessControlFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL0FsbG93QWxsQWNjZXNzQ29udHJvbEZhY3RvcnkuamF2YQ==) | `100.00% <ø> (ø)` | |
| [.../helix/BrokerUserDefinedMessageHandlerFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL2hlbGl4L0Jyb2tlclVzZXJEZWZpbmVkTWVzc2FnZUhhbmRsZXJGYWN0b3J5LmphdmE=) | `52.83% <0.00%> (-13.84%)` | :arrow_down: |
| [...ker/routing/instanceselector/InstanceSelector.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9pbnN0YW5jZXNlbGVjdG9yL0luc3RhbmNlU2VsZWN0b3IuamF2YQ==) | `100.00% <ø> (ø)` | |
| [...ava/org/apache/pinot/client/AbstractResultSet.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0Fic3RyYWN0UmVzdWx0U2V0LmphdmE=) | `73.33% <ø> (+16.19%)` | :arrow_up: |
| [...n/java/org/apache/pinot/client/BrokerResponse.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0Jyb2tlclJlc3BvbnNlLmphdmE=) | `100.00% <ø> (ø)` | |
| [.../main/java/org/apache/pinot/client/Connection.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0Nvbm5lY3Rpb24uamF2YQ==) | `44.44% <ø> (-4.40%)` | :arrow_down: |
| [...org/apache/pinot/client/DynamicBrokerSelector.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0R5bmFtaWNCcm9rZXJTZWxlY3Rvci5qYXZh) | `88.57% <ø> (+15.84%)` | :arrow_up: |
| [...n/java/org/apache/pinot/client/ExecutionStats.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0V4ZWN1dGlvblN0YXRzLmphdmE=) | `68.88% <ø> (ø)` | |
| [...va/org/apache/pinot/client/ExternalViewReader.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0V4dGVybmFsVmlld1JlYWRlci5qYXZh) | `81.69% <ø> (+31.69%)` | :arrow_up: |
| [...inot/client/JsonAsyncHttpPinotClientTransport.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0pzb25Bc3luY0h0dHBQaW5vdENsaWVudFRyYW5zcG9ydC5qYXZh) | `56.36% <ø> (-5.64%)` | :arrow_down: |
| ... and [1235 more](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=footer). Last update [53c2a3c...ab9c377](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r606506748
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1649,63 +1649,81 @@ public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, Str
public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ String tableNameWithType;
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
ZKMetadataUtils.updateSegmentMetadata(offlineSegmentZKMetadata, segmentMetadata);
offlineSegmentZKMetadata.setDownloadUrl(downloadUrl);
offlineSegmentZKMetadata.setCrypterName(crypter);
offlineSegmentZKMetadata.setPushTime(System.currentTimeMillis());
+
+ if (isRealtimeOnlyTable(tableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ instancePartitionsType = InstancePartitionsType.CONSUMING;
Review comment:
added comments about why we chose CONSUMING today. Left TODO for further improvement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r603440716
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -25,10 +25,16 @@
import java.util.List;
import java.util.Map;
+import java.util.UUID;
Review comment:
reverted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] codecov-io commented on pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
codecov-io commented on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-776350585
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=h1) Report
> Merging [#6567](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=desc) (289f1cb) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/1beaab59b73f26c4e35f3b9bc856b03806cddf5a?el=desc) (1beaab5) will **decrease** coverage by `22.40%`.
> The diff coverage is `41.89%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #6567 +/- ##
===========================================
- Coverage 66.44% 44.04% -22.41%
===========================================
Files 1075 1341 +266
Lines 54773 65942 +11169
Branches 8168 9618 +1450
===========================================
- Hits 36396 29041 -7355
- Misses 15700 34463 +18763
+ Partials 2677 2438 -239
```
| Flag | Coverage Δ | |
|---|---|---|
| integration | `44.04% <41.89%> (?)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...ot/broker/broker/AllowAllAccessControlFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL0FsbG93QWxsQWNjZXNzQ29udHJvbEZhY3RvcnkuamF2YQ==) | `100.00% <ø> (ø)` | |
| [.../helix/BrokerUserDefinedMessageHandlerFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL2hlbGl4L0Jyb2tlclVzZXJEZWZpbmVkTWVzc2FnZUhhbmRsZXJGYWN0b3J5LmphdmE=) | `52.83% <0.00%> (-13.84%)` | :arrow_down: |
| [...org/apache/pinot/broker/queryquota/HitCounter.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9IaXRDb3VudGVyLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...che/pinot/broker/queryquota/MaxHitRateTracker.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9NYXhIaXRSYXRlVHJhY2tlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...ache/pinot/broker/queryquota/QueryQuotaEntity.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9RdWVyeVF1b3RhRW50aXR5LmphdmE=) | `0.00% <0.00%> (-50.00%)` | :arrow_down: |
| [...ker/routing/instanceselector/InstanceSelector.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9pbnN0YW5jZXNlbGVjdG9yL0luc3RhbmNlU2VsZWN0b3IuamF2YQ==) | `100.00% <ø> (ø)` | |
| [...ceselector/StrictReplicaGroupInstanceSelector.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9pbnN0YW5jZXNlbGVjdG9yL1N0cmljdFJlcGxpY2FHcm91cEluc3RhbmNlU2VsZWN0b3IuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...roker/routing/segmentpruner/TimeSegmentPruner.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL1RpbWVTZWdtZW50UHJ1bmVyLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...roker/routing/segmentpruner/interval/Interval.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL2ludGVydmFsL0ludGVydmFsLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...r/routing/segmentpruner/interval/IntervalTree.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL2ludGVydmFsL0ludGVydmFsVHJlZS5qYXZh) | `0.00% <0.00%> (ø)` | |
| ... and [1350 more](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=footer). Last update [12ee45c...289f1cb](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] yupeng9 commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
yupeng9 commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r574013285
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
##########
@@ -61,12 +61,15 @@ public void completeSegmentOperations(String rawTableName, SegmentMetadata segme
URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection,
HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter)
throws Exception {
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
String segmentName = segmentMetadata.getName();
- // Brand new segment, not refresh, directly add the segment
- ZNRecord segmentMetadataZnRecord =
- _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName);
+ if (_pinotHelixResourceManager.isRealtimeOnlyTable(rawTableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+ } else {
+ tableNameWithType= TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ }
+ ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType, segmentName);
if (segmentMetadataZnRecord == null) {
LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName);
Review comment:
log with `tableNameWithType `?
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1649,63 +1649,81 @@ public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, Str
public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ String tableNameWithType;
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
Review comment:
shall this be moved into the branch of `offlineTable` handling?
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
+ if (_pinotHelixResourceManager.isRealtimeOnlyTable(rawTableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+ } else {
+ tableNameWithType = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ }
String clientAddress = InetAddress.getByName(request.getRemoteAddr()).getHostName();
LOGGER.info("Processing upload request for segment: {} of table: {} from client: {}, ingestion descriptor: {}",
- segmentName, offlineTableName, clientAddress, ingestionDescriptor);
+ segmentName, tableNameWithType, clientAddress, ingestionDescriptor);
- // Skip segment validation if upload only segment metadata
- if (uploadType != FileUploadDownloadClient.FileUploadType.METADATA) {
+ // Skip segment validation if upload only segment metadata or it is a realtime table segment.
+ // TODO Perform a validation check for realtime segments too.
Review comment:
Shall we allow the realtime segment upload for the upsert table only? This can avoid the issues that @mcvsubbu pointed out.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] codecov-io edited a comment on pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-776350585
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=h1) Report
> Merging [#6567](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=desc) (773b45e) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/1beaab59b73f26c4e35f3b9bc856b03806cddf5a?el=desc) (1beaab5) will **decrease** coverage by `22.53%`.
> The diff coverage is `46.46%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #6567 +/- ##
===========================================
- Coverage 66.44% 43.91% -22.54%
===========================================
Files 1075 1392 +317
Lines 54773 67678 +12905
Branches 8168 9810 +1642
===========================================
- Hits 36396 29720 -6676
- Misses 15700 35472 +19772
+ Partials 2677 2486 -191
```
| Flag | Coverage Δ | |
|---|---|---|
| integration | `43.91% <46.46%> (?)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...ot/broker/broker/AllowAllAccessControlFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL0FsbG93QWxsQWNjZXNzQ29udHJvbEZhY3RvcnkuamF2YQ==) | `100.00% <ø> (ø)` | |
| [...t/broker/broker/BasicAuthAccessControlFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL0Jhc2ljQXV0aEFjY2Vzc0NvbnRyb2xGYWN0b3J5LmphdmE=) | `0.00% <0.00%> (ø)` | |
| [.../helix/BrokerUserDefinedMessageHandlerFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL2hlbGl4L0Jyb2tlclVzZXJEZWZpbmVkTWVzc2FnZUhhbmRsZXJGYWN0b3J5LmphdmE=) | `52.83% <0.00%> (-13.84%)` | :arrow_down: |
| [...org/apache/pinot/broker/queryquota/HitCounter.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9IaXRDb3VudGVyLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...che/pinot/broker/queryquota/MaxHitRateTracker.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9NYXhIaXRSYXRlVHJhY2tlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...ache/pinot/broker/queryquota/QueryQuotaEntity.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9RdWVyeVF1b3RhRW50aXR5LmphdmE=) | `0.00% <0.00%> (-50.00%)` | :arrow_down: |
| [...ker/routing/instanceselector/InstanceSelector.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9pbnN0YW5jZXNlbGVjdG9yL0luc3RhbmNlU2VsZWN0b3IuamF2YQ==) | `100.00% <ø> (ø)` | |
| [...roker/routing/segmentpruner/TimeSegmentPruner.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL1RpbWVTZWdtZW50UHJ1bmVyLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...roker/routing/segmentpruner/interval/Interval.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL2ludGVydmFsL0ludGVydmFsLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...r/routing/segmentpruner/interval/IntervalTree.java](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL2ludGVydmFsL0ludGVydmFsVHJlZS5qYXZh) | `0.00% <0.00%> (ø)` | |
| ... and [1414 more](https://codecov.io/gh/apache/incubator-pinot/pull/6567/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=footer). Last update [53c2a3c...773b45e](https://codecov.io/gh/apache/incubator-pinot/pull/6567?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r595468638
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
Review comment:
Done to make sure it is clear that which subtable the addSegment() call pushes to.
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1649,63 +1649,81 @@ public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, Str
public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ String tableNameWithType;
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
ZKMetadataUtils.updateSegmentMetadata(offlineSegmentZKMetadata, segmentMetadata);
offlineSegmentZKMetadata.setDownloadUrl(downloadUrl);
offlineSegmentZKMetadata.setCrypterName(crypter);
offlineSegmentZKMetadata.setPushTime(System.currentTimeMillis());
+
+ if (isRealtimeOnlyTable(tableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ instancePartitionsType = InstancePartitionsType.CONSUMING;
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r606511554
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/util/SegmentUtils.java
##########
@@ -0,0 +1,67 @@
+/**
+ * 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.util;
Review comment:
Moved. I think you meant the class should be moved to pinot/common/utils.
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/util/SegmentUtils.java
##########
@@ -0,0 +1,67 @@
+/**
+ * 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.util;
+
+import com.google.common.base.Preconditions;
+import java.util.Set;
+import org.apache.helix.HelixManager;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.ColumnPartitionMetadata;
+import org.apache.pinot.common.metadata.segment.RealtimeSegmentZKMetadata;
+import org.apache.pinot.common.utils.LLCSegmentName;
+
+
+// Util functions related to segments.
+public class SegmentUtils {
+ // Returns the partition id of a realtime segment based segment name and segment metadata info retrieved via Helix.
+ public static int getRealtimeSegmentPartitionId(String segmentName, String realtimeTableName,
Review comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r611879274
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -383,7 +383,12 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ // Means the segment is not completed and still consuming stream data
+ IN_PROGRESS,
+ // Means the segment is completed and sealed by some Pinot server and uploaded
+ DONE,
+ // Means the segment is uploaded to a Pinot controller by an external party
+ UPLOADED
Review comment:
https://github.com/apache/incubator-pinot/issues/6777 filed for this issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r599980587
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
##########
@@ -57,41 +57,37 @@ public ZKOperator(PinotHelixResourceManager pinotHelixResourceManager, Controlle
_controllerMetrics = controllerMetrics;
}
- public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata,
+ public void completeSegmentOperations(String tableNameWithType, SegmentMetadata segmentMetadata,
URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection,
HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter)
throws Exception {
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
String segmentName = segmentMetadata.getName();
-
- // Brand new segment, not refresh, directly add the segment
- ZNRecord segmentMetadataZnRecord =
- _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName);
+ ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType, segmentName);
if (segmentMetadataZnRecord == null) {
- LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName);
+ LOGGER.info("Adding new segment {} from table {}", segmentName, tableNameWithType);
processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter,
- rawTableName, segmentName, moveSegmentToFinalLocation);
+ tableNameWithType, segmentName, moveSegmentToFinalLocation);
return;
}
- LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, rawTableName);
+ LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, tableNameWithType);
processExistingSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation,
- enableParallelPushProtection, headers, zkDownloadURI, crypter, offlineTableName, segmentName,
+ enableParallelPushProtection, headers, zkDownloadURI, crypter, tableNameWithType, segmentName,
segmentMetadataZnRecord, moveSegmentToFinalLocation);
}
private void processExistingSegment(SegmentMetadata segmentMetadata, URI finalSegmentLocationURI,
File currentSegmentLocation, boolean enableParallelPushProtection, HttpHeaders headers, String zkDownloadURI,
- String crypter, String offlineTableName, String segmentName, ZNRecord znRecord,
+ String crypter, String tableNameWithType, String segmentName, ZNRecord znRecord,
boolean moveSegmentToFinalLocation)
throws Exception {
OfflineSegmentZKMetadata existingSegmentZKMetadata = new OfflineSegmentZKMetadata(znRecord);
Review comment:
for processExistingSegment(), I think we should leave it to another PR and let this PR focus on adding new segments. For now, this PR will reject upload segments of the same name. Notice that for our Upsert table use case, uploading new segments would be sufficient. I will have a follow up PR for refreshing existing segments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r595569771
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -358,7 +358,7 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ IN_PROGRESS, DONE, UPLOAD
Review comment:
```suggestion
IN_PROGRESS, DONE, UPLOADED
```
Can you please add some comments on what each of these mean? With the addition of the new value here, we need to do that.
IN_PROGRESS means the segment is being consumed
DONE means the segment is completed, and uploaded by pinot-server.
UPLOADED means the segment is uploaded by some external agent (?)
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
+ if (_pinotHelixResourceManager.isUpsertTable(rawTableName)) {
Review comment:
This logic seems a little strange. Does this mean that we cannot upload segment to the offline part of an upsert table ? Or that offline part does not exist for an upsert table? What if we introduce this part later?
Instead, I suggest that we include the type query-part as well in the url. If the type is not present, we assume it is OFFLINE. If it is present, we treat it accordingly, but throw an exception if it is realtime type but upsert is not turned on
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r603442187
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java
##########
@@ -551,6 +554,30 @@ public SimpleHttpResponse uploadSegment(URI uri, String segmentName, File segmen
return uploadSegment(uri, segmentName, segmentFile, null, parameters, DEFAULT_SOCKET_TIMEOUT_MS);
}
+ /**
+ * Upload segment with segment file using default settings. Include table name and type as a request parameters.
+ *
+ * @param uri URI
+ * @param segmentName Segment name
+ * @param segmentFile Segment file
+ * @param tableName Table name with or without type suffix
+ * @param tableType Table type
+ * @return Response
+ * @throws IOException
+ * @throws HttpErrorStatusException
+ */
+ public SimpleHttpResponse uploadSegment(URI uri, String segmentName, File segmentFile, String tableName,
+ TableType tableType)
+ throws IOException, HttpErrorStatusException {
+ // Add table name and type request parameters
+ NameValuePair tableNameValuePair = new BasicNameValuePair(QueryParameters.TABLE_NAME, tableName);
+ NameValuePair tableTypeValuePair = new BasicNameValuePair(QueryParameters.TABLE_TYPE, tableType.name());
+ List<NameValuePair> parameters = new ArrayList<>();
Review comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r606353425
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -383,7 +383,12 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ // Means the segment is not completed and still consuming stream data
Review comment:
```suggestion
// Means the segment is in CONSUMING state
```
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -383,7 +383,12 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ // Means the segment is not completed and still consuming stream data
+ IN_PROGRESS,
+ // Means the segment is completed and sealed by some Pinot server and uploaded
+ DONE,
+ // Means the segment is uploaded to a Pinot controller by an external party
+ UPLOADED
Review comment:
Should we take this opportunity to add two more enums: CONSUMING and COMPLETED (these are the terms used elsewhere), and slowly obsolete the IN_PROGRESS and DONE. For now, they will be treated equivalent to their counterpart.
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
##########
@@ -57,41 +61,44 @@ public ZKOperator(PinotHelixResourceManager pinotHelixResourceManager, Controlle
_controllerMetrics = controllerMetrics;
}
- public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata,
+ public void completeSegmentOperations(String tableNameWithType, SegmentMetadata segmentMetadata,
URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection,
HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter)
throws Exception {
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
String segmentName = segmentMetadata.getName();
-
- // Brand new segment, not refresh, directly add the segment
- ZNRecord segmentMetadataZnRecord =
- _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName);
+ ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType, segmentName);
if (segmentMetadataZnRecord == null) {
- LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName);
+ LOGGER.info("Adding new segment {} from table {}", segmentName, tableNameWithType);
processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter,
- rawTableName, segmentName, moveSegmentToFinalLocation);
+ tableNameWithType, segmentName, moveSegmentToFinalLocation);
return;
}
- LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, rawTableName);
+ LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, tableNameWithType);
Review comment:
Move this log message to below the if statement. Otherwise it is a misleading log.
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/util/SegmentUtils.java
##########
@@ -0,0 +1,67 @@
+/**
+ * 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.util;
Review comment:
Could this class not be in pinot/common/utils along with classes like `LLCSegmentName`
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -383,7 +383,12 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ // Means the segment is not completed and still consuming stream data
+ IN_PROGRESS,
+ // Means the segment is completed and sealed by some Pinot server and uploaded
Review comment:
```suggestion
// Means the segment is in ONLINE state (segment completed consuming and has been saved in segment store).
```
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1649,63 +1649,81 @@ public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, Str
public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ String tableNameWithType;
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
ZKMetadataUtils.updateSegmentMetadata(offlineSegmentZKMetadata, segmentMetadata);
offlineSegmentZKMetadata.setDownloadUrl(downloadUrl);
offlineSegmentZKMetadata.setCrypterName(crypter);
offlineSegmentZKMetadata.setPushTime(System.currentTimeMillis());
+
+ if (isRealtimeOnlyTable(tableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ instancePartitionsType = InstancePartitionsType.CONSUMING;
Review comment:
I agree with Jackie that it is counter-intuitive. Add a comment here reasoning why it should be set to CONSUMING. If we enable uploading realtime segments to non-upsert tables, at least we will know why/what we will need to fix
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -247,24 +248,35 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
+ if (tableType == TableType.OFFLINE) {
+ tableNameWithType = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ } else {
+ if (!_pinotHelixResourceManager.isUpsertTable(rawTableName)) {
+ throw new UnsupportedOperationException(
+ "Upload segment to non-upsert realtime table is not supported " + rawTableName);
+ }
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+ }
+
String clientAddress = InetAddress.getByName(request.getRemoteAddr()).getHostName();
LOGGER.info("Processing upload request for segment: {} of table: {} from client: {}, ingestion descriptor: {}",
- segmentName, offlineTableName, clientAddress, ingestionDescriptor);
+ segmentName, tableNameWithType, clientAddress, ingestionDescriptor);
- // Skip segment validation if upload only segment metadata
- if (uploadType != FileUploadDownloadClient.FileUploadType.METADATA) {
+ // Skip segment validation if upload is to an offline table and only segment metadata.
Review comment:
Add some comment on why we dont need validation for realtime segment
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/util/SegmentUtils.java
##########
@@ -0,0 +1,67 @@
+/**
+ * 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.util;
+
+import com.google.common.base.Preconditions;
+import java.util.Set;
+import org.apache.helix.HelixManager;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.ColumnPartitionMetadata;
+import org.apache.pinot.common.metadata.segment.RealtimeSegmentZKMetadata;
+import org.apache.pinot.common.utils.LLCSegmentName;
+
+
+// Util functions related to segments.
+public class SegmentUtils {
+ // Returns the partition id of a realtime segment based segment name and segment metadata info retrieved via Helix.
+ public static int getRealtimeSegmentPartitionId(String segmentName, String realtimeTableName,
Review comment:
Add in javadocs saying this method is NOT to be called during query processing. I am being a bit paranoid that someone may make a change in the broker at some point without realizing what is going on
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r595473794
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -358,7 +358,7 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ IN_PROGRESS, DONE, UPLOAD
Review comment:
Currently UPLOAD is only applicable to upsert enable tables. Is there an effort to expand the scope? for upsert table, UPLOAD_APPEND is not needed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r598027470
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java
##########
@@ -172,6 +183,39 @@ private void checkReplication(InstancePartitions instancePartitions) {
}
}
+ private int getSegmentPartitionId(String segmentName) {
+ // A fast path if the segmentName is a LLC segment name and we can get the partition id from the name directly.
+ if (LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
+ return new LLCSegmentName(segmentName).getPartitionGroupId();
+ }
+ // Otherwise, retrieve the partition id from the segment zk metadata. Currently only realtime segments from upsert
+ // enabled tables have partition ids in their segment metadata.
+ RealtimeSegmentZKMetadata segmentZKMetadata = ZKMetadataProvider
+ .getRealtimeSegmentZKMetadata(_helixManager.getHelixPropertyStore(), _realtimeTableName, segmentName);
+ Preconditions
+ .checkState(segmentZKMetadata != null, "Failed to find segment ZK metadata for segment: %s of table: %s",
+ segmentName, _realtimeTableName);
+ return getSegmentPartitionIdFromZkMetaData(segmentZKMetadata);
+ }
+
+ private int getSegmentPartitionIdFromZkMetaData(RealtimeSegmentZKMetadata segmentZKMetadata) {
+ String segmentName = segmentZKMetadata.getSegmentName();
+ Preconditions.checkState(segmentZKMetadata.getPartitionMetadata() != null,
+ "Segment ZK metadata for segment: %s of table: %s does not contain partition metadata", segmentName,
+ _realtimeTableName);
+
+ ColumnPartitionMetadata partitionMetadata =
+ segmentZKMetadata.getPartitionMetadata().getColumnPartitionMap().get(_partitionColumn);
+ Preconditions.checkState(partitionMetadata != null,
+ "Segment ZK metadata for segment: %s of table: %s does not contain partition metadata for column: %s",
Review comment:
good idea. done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] yupeng9 commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
yupeng9 commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r598156395
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUntarSegmentToLocal(String uri, File tableDataDir, String segmentName)
Review comment:
makes sense
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUntarSegmentToLocal(String uri, File tableDataDir, String segmentName)
+ throws Exception {
+ File tempDir = new File(tableDataDir, "tmp-" + segmentName + "-" + UUID.randomUUID());
+ FileUtils.forceMkdir(tempDir);
+ File tempTarFile = new File(tempDir, segmentName + TAR_GZ_SUFFIX);
+ File tempSegmentDir = new File(tempDir, segmentName);
+ try {
+ try {
+ SegmentFetcherFactory.fetchSegmentToLocal(uri, tempTarFile);
+ LOGGER.info("Downloaded tarred segment: {} from: {} to: {}, file length: {}", segmentName, uri, tempTarFile,
+ tempTarFile.length());
+ } catch (AttemptsExceededException e) {
+ LOGGER.error("Attempts exceeded when downloading segment: {} from: {} to: {}", segmentName, uri,
+ tempTarFile, e);
+ Utils.rethrowException(e);
+ }
+
+ try {
+ // If an exception is thrown when untarring, it means the tar file is broken OR not found after the retry.
+ // Thus, there's no need to retry again.
+ File tempIndexDir = TarGzCompressionUtils.untar(tempTarFile, tempSegmentDir).get(0);
+ File segmentDir = new File(tableDataDir, segmentName);
+ if (segmentDir.exists()) {
+ LOGGER.info("Deleting existing index directory for segment: {}", segmentName);
+ FileUtils.deleteDirectory(segmentDir);
+ }
+ FileUtils.moveDirectory(tempIndexDir, segmentDir);
+ LOGGER.info("Successfully downloaded segment: {} to: {}", segmentName, segmentDir);
+ } catch (Exception e) {
+ LOGGER.error("Exception when untarring segment: {} for from {} to {}", segmentName, tempTarFile, tempSegmentDir,
+ e);
+ Utils.rethrowException(e);
+ }
+ } finally {
+ FileUtils.deleteQuietly(tempDir);
Review comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r595427533
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
Review comment:
The method tableNameWithType() used later has this behavior already. Here I try to make it clear that the table name is typed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] Jackie-Jiang commented on pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-805169440
> A new integration test UpsertTableSegmentUploadIntegrationTest is added to verify the uploaded segment is added to a realtime upsert enabled table correctly. The segment name used in the test does not follow llc segment convention but it is loaded successfully. The query result is good.
>
> I do quite understand why we should "Exclude uploaded segment from the processing". During segment load, the upsert table metadata should be processed based on the loaded segment, right?
>
> > Please take a look at `PinotLLCRealtimeSegmentManager` and ensure it works on uploaded segments:
> >
> > * Segment name for uploaded segment does not follow the llc segment name convention
> > * Exclude uploaded segment from the processing
I took another look at `PinotLLCRealtimeSegmentManager` and seems non-LLC segments are already excluded (because it might co-exist with HLC segments before). We cannot include them during metadata processing because it relies on the segment name format.
For the issue within the `processExistingSegment()`, for uploading the new segment it won't hit that code path, but for refreshing the existing segment it won't set the fields properly.
If possible, I would recommend setting up an actual table, upload some segments and keep it running for some time to ensure all components can properly work on the uploaded segments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r595472428
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUtarSegmentToLocal(String uri, File tableDataDir, String segmentName)
+ throws Exception {
+ File tempDir = new File(tableDataDir, "tmp-" + segmentName + "-" + UUID.randomUUID());
+ FileUtils.forceMkdir(tempDir);
+ File tempTarFile = new File(tempDir, segmentName + TAR_GZ_SUFFIX);
+ File tempSegmentDir = new File(tempDir, segmentName);
+ try {
+ try {
+ SegmentFetcherFactory.fetchSegmentToLocal(uri, tempTarFile);
+ LOGGER.info("Downloaded tarred segment: {} from: {} to: {}, file length: {}", segmentName, uri, tempTarFile,
+ tempTarFile.length());
+ } catch (AttemptsExceededException e) {
+ LOGGER.error("Attempts exceeded when downloading segment: {} : {} from: {} to: {}", segmentName, uri,
+ tempTarFile);
+ Utils.rethrowException(e);
+ }
+
+ try {
+ // If an exception is thrown when untarring, it means the tar file is broken OR not found after the retry.
+ // Thus, there's no need to retry again.
+ File tempIndexDir = TarGzCompressionUtils.untar(tempTarFile, tempSegmentDir).get(0);
+ File segmentDir = new File(tableDataDir, segmentName);
+ if (segmentDir.exists()) {
+ LOGGER.info("Deleting existing index directory for segment: {}", segmentName);
+ FileUtils.deleteDirectory(segmentDir);
+ }
+ FileUtils.moveDirectory(tempIndexDir, segmentDir);
+ LOGGER.info("Successfully downloaded segment: {} to: {}", segmentName, segmentDir);
+ } catch (Exception e) {
+ LOGGER
+ .error("Exception when untarring segment: {} for from {} to {}", segmentName, tempTarFile, tempSegmentDir);
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r596470495
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java
##########
@@ -85,15 +92,19 @@
private HelixManager _helixManager;
private String _realtimeTableName;
private int _replication;
+ private String _partitionColumn;
@Override
public void init(HelixManager helixManager, TableConfig tableConfig) {
_helixManager = helixManager;
_realtimeTableName = tableConfig.getTableName();
_replication = tableConfig.getValidationConfig().getReplicasPerPartitionNumber();
+ ReplicaGroupStrategyConfig replicaGroupStrategyConfig =
+ tableConfig.getValidationConfig().getReplicaGroupStrategyConfig();
+ _partitionColumn = replicaGroupStrategyConfig != null ? replicaGroupStrategyConfig.getPartitionColumn() : null;
- LOGGER.info("Initialized RealtimeSegmentAssignment with replication: {} for table: {}", _replication,
- _realtimeTableName);
+ LOGGER.info("Initialized RealtimeSegmentAssignment with replication: {} partitionColumn: {} for table: {}",
Review comment:
Done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] Jackie-Jiang commented on pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-806028614
For the new integration test, it would be good if we push some events into kafka as well and make the commit threshold low so that we have CONSUMING, DONE, UPLOADED segments at the same time. Also, we can run the `RealtimeSegmentValidationManager` and see if it handles UPLOADED segments properly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r598027802
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUntarSegmentToLocal(String uri, File tableDataDir, String segmentName)
Review comment:
SegmentFetcher fetches segment while SegmentUploader is responsible for segment uploading. So fetch related functions are kept in this class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r597438708
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
+ if (_pinotHelixResourceManager.isUpsertTable(rawTableName)) {
Review comment:
yes it is a good idea to divide into realtime and offline cases rather than upsert vs non-upsert case. Added the table type parameters to the REST end point too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] jtao15 commented on pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
jtao15 commented on pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#issuecomment-861725297
@chenboat Hi, can you open the read permission of the design document?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r606494472
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
##########
@@ -57,41 +61,44 @@ public ZKOperator(PinotHelixResourceManager pinotHelixResourceManager, Controlle
_controllerMetrics = controllerMetrics;
}
- public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata,
+ public void completeSegmentOperations(String tableNameWithType, SegmentMetadata segmentMetadata,
URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection,
HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter)
throws Exception {
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
String segmentName = segmentMetadata.getName();
-
- // Brand new segment, not refresh, directly add the segment
- ZNRecord segmentMetadataZnRecord =
- _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName);
+ ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType, segmentName);
if (segmentMetadataZnRecord == null) {
- LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName);
+ LOGGER.info("Adding new segment {} from table {}", segmentName, tableNameWithType);
processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter,
- rawTableName, segmentName, moveSegmentToFinalLocation);
+ tableNameWithType, segmentName, moveSegmentToFinalLocation);
return;
}
- LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, rawTableName);
+ LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, tableNameWithType);
Review comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r573367216
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
##########
@@ -245,21 +245,28 @@ private SuccessResponse uploadSegment(@Nullable String tableName, FormDataMultiP
LOGGER.info("Uploading a segment {} to table: {}, push type {}, (Derived from segment metadata)", segmentName, tableName, uploadType);
}
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ String tableNameWithType;
+ if (_pinotHelixResourceManager.isRealtimeOnlyTable(rawTableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+ } else {
+ tableNameWithType = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
+ }
String clientAddress = InetAddress.getByName(request.getRemoteAddr()).getHostName();
LOGGER.info("Processing upload request for segment: {} of table: {} from client: {}, ingestion descriptor: {}",
- segmentName, offlineTableName, clientAddress, ingestionDescriptor);
+ segmentName, tableNameWithType, clientAddress, ingestionDescriptor);
- // Skip segment validation if upload only segment metadata
- if (uploadType != FileUploadDownloadClient.FileUploadType.METADATA) {
+ // Skip segment validation if upload only segment metadata or it is a realtime table segment.
+ // TODO Perform a validation check for realtime segments too.
Review comment:
For realtime segments, one of the validations that need to be performed is that the segment is generated off the same stream endpoint as the subscription of realtime table. So, the segment generation should include something in the metadata to indicate this. Otherwise, queries will return bad results.
For example, if the same topic was consumed from one data center, it is not possible to copy that segment over, and "upload" to the realtime table in another data center, even though the stream data, schema, etc. are same. We need some indication of the stream source (so that offsets are aligned).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r595468719
##########
File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
##########
@@ -1649,63 +1649,81 @@ public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, Str
public void addNewSegment(String tableName, SegmentMetadata segmentMetadata, String downloadUrl,
@Nullable String crypter) {
String segmentName = segmentMetadata.getName();
- String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName);
-
+ String tableNameWithType;
+ InstancePartitionsType instancePartitionsType;
// NOTE: must first set the segment ZK metadata before assigning segment to instances because segment assignment
// might need them to determine the partition of the segment, and server will need them to download the segment
OfflineSegmentZKMetadata offlineSegmentZKMetadata = new OfflineSegmentZKMetadata();
ZKMetadataUtils.updateSegmentMetadata(offlineSegmentZKMetadata, segmentMetadata);
offlineSegmentZKMetadata.setDownloadUrl(downloadUrl);
offlineSegmentZKMetadata.setCrypterName(crypter);
offlineSegmentZKMetadata.setPushTime(System.currentTimeMillis());
+
+ if (isRealtimeOnlyTable(tableName)) {
+ tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ instancePartitionsType = InstancePartitionsType.CONSUMING;
Review comment:
In an upsert enabled LLC realtime table, all segments of the same partition are colocated on the same server -- consuming or completed. So it is fine to use CONSUMING as the InstancePartitionsType.
In addition, the current RealtimeSegmentAssignment.assignSegment(..) method, it is hardcoded explicitly to assign new segments to CONSUMING instance partition.
https://github.com/apache/incubator-pinot/blob/3b7093039469b2a00269f747543c6c29c8f1801f/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java#L102
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r595595831
##########
File path: pinot-integration-tests/src/test/resources/upsert_table_test.schema
##########
@@ -0,0 +1,33 @@
+{
+ "dimensionFieldSpecs": [
+ {
+ "dataType": "INT",
+ "singleValueField": true,
+ "name": "clientId"
+ },
+ {
+ "dataType": "STRING",
+ "singleValueField": true,
+ "name": "city"
+ },
+ {
+ "dataType": "STRING",
+ "singleValueField": true,
+ "name": "description"
+ },
+ {
+ "dataType": "INT",
+ "singleValueField": true,
+ "name": "salary"
+ }
+ ],
+ "timeFieldSpec": {
+ "incomingGranularitySpec": {
+ "timeType": "DAYS",
+ "dataType": "INT",
+ "name": "DaysSinceEpoch"
+ }
+ },
+ "primaryKeyColumns": ["clientId"],
+ "schemaName": "upsertSchema"
+}
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r596468455
##########
File path: pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterIntegrationTestUtils.java
##########
@@ -285,7 +285,7 @@ public static void buildSegmentFromAvro(File avroFile, TableConfig tableConfig,
segmentGeneratorConfig.setOutDir(segmentDir.getPath());
segmentGeneratorConfig.setTableName(tableConfig.getTableName());
// Test segment with space and special character in the file name
- segmentGeneratorConfig.setSegmentNamePostfix(segmentIndex + " %");
+ segmentGeneratorConfig.setSegmentNamePostfix(Integer.toString(segmentIndex));
Review comment:
Reverted back.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r607259301
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -383,7 +383,12 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ // Means the segment is not completed and still consuming stream data
+ IN_PROGRESS,
+ // Means the segment is completed and sealed by some Pinot server and uploaded
+ DONE,
+ // Means the segment is uploaded to a Pinot controller by an external party
+ UPLOADED
Review comment:
Removed the two enums.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable adding new segments to a upsert-enabled realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r606491097
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
##########
@@ -383,7 +383,12 @@
public static class Segment {
public static class Realtime {
public enum Status {
- IN_PROGRESS, DONE
+ // Means the segment is not completed and still consuming stream data
+ IN_PROGRESS,
+ // Means the segment is completed and sealed by some Pinot server and uploaded
+ DONE,
+ // Means the segment is uploaded to a Pinot controller by an external party
+ UPLOADED
Review comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #6567: Enable segment upload to a realtime table
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #6567:
URL: https://github.com/apache/incubator-pinot/pull/6567#discussion_r595472001
##########
File path: pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
##########
@@ -118,6 +125,47 @@ public static void fetchSegmentToLocal(String uri, File dest)
fetchSegmentToLocal(new URI(uri), dest);
}
+ /**
+ * Fetches a segment from a given URI and untar the segment file to the dest dir (i.e., tableDataDir + segmentName).
+ */
+ public static void fetchAndUtarSegmentToLocal(String uri, File tableDataDir, String segmentName)
+ throws Exception {
+ File tempDir = new File(tableDataDir, "tmp-" + segmentName + "-" + UUID.randomUUID());
+ FileUtils.forceMkdir(tempDir);
+ File tempTarFile = new File(tempDir, segmentName + TAR_GZ_SUFFIX);
+ File tempSegmentDir = new File(tempDir, segmentName);
+ try {
+ try {
+ SegmentFetcherFactory.fetchSegmentToLocal(uri, tempTarFile);
+ LOGGER.info("Downloaded tarred segment: {} from: {} to: {}, file length: {}", segmentName, uri, tempTarFile,
+ tempTarFile.length());
+ } catch (AttemptsExceededException e) {
+ LOGGER.error("Attempts exceeded when downloading segment: {} : {} from: {} to: {}", segmentName, uri,
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org