You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "wirybeaver (via GitHub)" <gi...@apache.org> on 2025/11/19 07:29:56 UTC
[PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
wirybeaver opened a new pull request, #17235:
URL: https://github.com/apache/pinot/pull/17235
(no comment)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2666379437
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1789,10 +1807,16 @@ public void addTable(TableConfig tableConfig)
// Add ideal state
_helixAdmin.addResource(_helixClusterName, tableNameWithType, idealState);
LOGGER.info("Adding table {}: Added ideal state for offline table", tableNameWithType);
- } else {
+ } else if (consumeMeta == null || consumeMeta.isEmpty()) {
// Add ideal state with the first CONSUMING segment
_pinotLLCRealtimeSegmentManager.setUpNewTable(tableConfig, idealState);
LOGGER.info("Adding table {}: Added ideal state with first consuming segment", tableNameWithType);
+ } else {
+ // Add ideal state with the first CONSUMING segment with designated partition consuming metadata
+ // Add ideal state with the first CONSUMING segment
+ _pinotLLCRealtimeSegmentManager.setUpNewTable(tableConfig, idealState, consumeMeta);
+ LOGGER.info("Adding table {}: Added consuming segments ideal state given the designated consuming metadata",
Review Comment:
This routine can be shared for other potential purposes. for example, a use can create a table with kafka offset they wanted to ingestion from
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2663375714
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/WatermarkInductionResult.java:
##########
@@ -0,0 +1,110 @@
+/**
+ * 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.controller.helix.core;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+
+
+/**
+ * Represents the result of a watermark induction process, containing a list of watermarks.
+ */
+public class WatermarkInductionResult {
+
+ private List<Watermark> _watermarks;
+
+ /**
+ * The @JsonCreator annotation marks this constructor to be used for deserializing
+ * a JSON array back into a WaterMarks object.
+ *
+ * @param watermarks The list of watermarks.
+ */
+ @JsonCreator
+ public WatermarkInductionResult(@JsonProperty("watermarks") List<Watermark> watermarks) {
+ _watermarks = watermarks;
+ }
+
+ /**
+ * Gets the list of watermarks.
+ *
+ * @return The list of watermarks.
+ */
+ @JsonGetter("watermarks")
+ public List<Watermark> getWatermarks() {
+ return _watermarks;
+ }
+
+ /**
+ * Represents a single watermark with its partition, sequence, and offset.
Review Comment:
sure, will modify the javadoc
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2621391126
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
Review Comment:
it seems we only support realtime, do we have any validation to avoid this using on hybrid and offline?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2621392206
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
Review Comment:
same for Upsert.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2666889336
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ if (realtimeTableConfig.getUpsertConfig() != null) {
+ return new CopyTableResponse("fail", "upsert table copy not supported");
+ }
Review Comment:
yeah, you are right. the add schema is moved after validation 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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2630020019
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4737,6 +4761,37 @@ public QueryWorkloadManager getQueryWorkloadManager() {
return _queryWorkloadManager;
}
+ public WatermarkInductionResult inductConsumingWatermarks(String tableName) throws TableNotFoundException {
+ String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ if (!hasRealtimeTable(tableName)) {
+ throw new TableNotFoundException("Table " + tableNameWithType + " does not exist");
+ }
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Preconditions.checkNotNull(tableConfig, "Table " + tableNameWithType + "exists but null tableConfig");
+ List<StreamConfig> streamConfigs = IngestionConfigUtils.getStreamConfigs(tableConfig);
+ IdealState idealState = _helixAdmin
+ .getResourceIdealState(getHelixClusterName(), tableNameWithType);
+ List<PartitionGroupConsumptionStatus> lst = _pinotLLCRealtimeSegmentManager
+ .getPartitionGroupConsumptionStatusList(idealState, streamConfigs);
+ List<WatermarkInductionResult.Watermark> watermarks = lst.stream().map(status -> {
+ long seq = status.getSequenceNumber();
+ long startOffset;
+ try {
+ if (status.getStatus().equalsIgnoreCase("done")) {
Review Comment:
Yeah, good catch. 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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2621380681
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -449,6 +451,25 @@ public String getPauselessTableDebugInfo(
}
}
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/tables/{tableName}/consumerWatermarks")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.GET_IDEAL_STATE)
+ @ApiOperation(value = "Get table ideal state", notes = "Get table ideal state")
+ public WatermarkInductionResult inductConsumingWatermark(
+ @ApiParam(value = "Name of the realtime table", required = true) @PathParam("tableName") String tableName,
+ @Context HttpHeaders headers) {
+ try {
+ String table = DatabaseUtils.translateTableName(tableName, headers);
Review Comment:
Table copy won't change table name, is that correct?
That said, we cannot copy table in the same cluster (share controller) from one tenant to another, is that correct?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #17235:
URL: https://github.com/apache/pinot/pull/17235#issuecomment-3672837020
The unit test Set-1 fail in java 21 but succeed in java 11. The error is about "selectionCombineOperator Not found issue", which is nothing related to table copy. I just do a rebase without any conflict resolving. Previous tests were all successful.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2548138260
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -433,6 +435,25 @@ public String getPauselessTableDebugInfo(
}
}
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/tables/{tableName}/watermarks")
Review Comment:
the endpoints name is not intuitive. shall we rename it to "@Path("/tables/{tableName}/consumptionWatermark"
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2663313232
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4737,6 +4761,40 @@ public QueryWorkloadManager getQueryWorkloadManager() {
return _queryWorkloadManager;
}
+ public WatermarkInductionResult inductConsumingWatermarks(String tableName) throws TableNotFoundException {
Review Comment:
Please add javadoc and comments on this public method.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2621386460
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
Review Comment:
if the schema already exists, what will happen?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2663321756
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/WatermarkInductionResult.java:
##########
@@ -0,0 +1,110 @@
+/**
+ * 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.controller.helix.core;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+
+
+/**
+ * Represents the result of a watermark induction process, containing a list of watermarks.
+ */
+public class WatermarkInductionResult {
+
+ private List<Watermark> _watermarks;
+
+ /**
+ * The @JsonCreator annotation marks this constructor to be used for deserializing
+ * a JSON array back into a WaterMarks object.
+ *
+ * @param watermarks The list of watermarks.
+ */
+ @JsonCreator
+ public WatermarkInductionResult(@JsonProperty("watermarks") List<Watermark> watermarks) {
+ _watermarks = watermarks;
+ }
+
+ /**
+ * Gets the list of watermarks.
+ *
+ * @return The list of watermarks.
+ */
+ @JsonGetter("watermarks")
+ public List<Watermark> getWatermarks() {
+ return _watermarks;
+ }
+
+ /**
+ * Represents a single watermark with its partition, sequence, and offset.
+ */
+ public static class Watermark {
+ private long _partitionGroupId;
+ private long _sequenceNumber;
+ private long _offset;
+
+ /**
+ * The @JsonCreator annotation tells Jackson to use this constructor to create
+ * a WaterMark instance from a JSON object. The @JsonProperty annotations
+ * map the keys in the JSON object to the constructor parameters.
+ *
+ * @param partitionGroupId The ID of the partition group.
+ * @param sequenceNumber The sequence number of the watermark.
+ * @param offset The offset of the watermark.
+ */
+ @JsonCreator
+ public Watermark(@JsonProperty("partitionGroupId") long partitionGroupId,
+ @JsonProperty("sequenceNumber") long sequenceNumber, @JsonProperty("offset") long offset) {
+ _partitionGroupId = partitionGroupId;
+ _sequenceNumber = sequenceNumber;
+ _offset = offset;
+ }
+
+ /**
+ * Gets the partition group ID.
+ *
+ * @return The partition group ID.
+ */
+ @JsonGetter("partitionGroupId")
+ public long getPartitionGroupId() {
+ return _partitionGroupId;
+ }
+
+ /**
+ * Gets the sequence number.
+ *
+ * @return The sequence number.
+ */
+ @JsonGetter("sequenceNumber")
+ public long getSequenceNumber() {
+ return _sequenceNumber;
+ }
+
+ /**
+ * Gets the offset.
Review Comment:
Instead of just repeating the method name, can you be more specific? This is the offset of the stream partition?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2677047071
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PartitionGroupInfo.java:
##########
@@ -0,0 +1,78 @@
+/**
+ * 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.controller.helix.core.realtime;
+
+import org.apache.pinot.controller.helix.core.WatermarkInductionResult;
+import org.apache.pinot.spi.stream.LongMsgOffset;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+
+
+/**
+ * The {@code PartitionGroupInfo} class represents the metadata and sequence number for a partition group.
+ * It encapsulates the {@link PartitionGroupMetadata} and the sequence number associated with it.
+ */
+public class PartitionGroupInfo {
Review Comment:
Replace with Pair class
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1729,8 +1734,21 @@ public void updateDatabaseConfig(DatabaseConfig databaseConfig) {
*/
public void addTable(TableConfig tableConfig)
throws IOException {
+ addTable(tableConfig, Collections.emptyList());
+ }
+
+ /**
+ * Performs validations of table config and adds the table to zookeeper
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2673501184
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -449,6 +451,25 @@ public String getPauselessTableDebugInfo(
}
}
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/tables/{tableName}/consumerWatermarks")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.GET_IDEAL_STATE)
+ @ApiOperation(value = "Get table ideal state", notes = "Get table ideal state")
Review Comment:
good catch.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #17235:
URL: https://github.com/apache/pinot/pull/17235#issuecomment-3554733903
Unit Test Failure: seems that the error is not caused by my PR. Checking.
PinotHelixResourceManagerStatelessTest.testGetLiveBrokers:383 ยป InvalidTableConfig No schema defined for table: testTable_OFFLINE
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2630024665
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
Review Comment:
if the tableconfig already exists, an exception will throw. if the schema exist, it won't throw exception
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2621393154
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}, tableConfig: {}",
+ tableName, realtimeTableConfig.toString());
+
+ URI watermarkUri = new URI(sourceControllerUri + "/tables/" + tableName + "/consumerWatermarks");
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ List<PartitionGroupInfo> partitionGroupInfos = watermarkInductionResult.getWatermarks().stream()
+ .map(PartitionGroupInfo::from)
+ .collect(Collectors.toList());
+
+ _pinotHelixResourceManager.addTable(realtimeTableConfig, partitionGroupInfos);
+ }
+ LOGGER.info("[copyTable] Finished Table Config copy: {}", tableName);
+ return new CopyTableResponse("success");
+ } catch (Exception e) {
+ LOGGER.error("[copyTable] Error copying table: {}", tableName, e);
+ throw new ControllerApplicationException(LOGGER, "Error copying table: " + e.getMessage(),
+ Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
+
+ /**
+ * Tweaks the realtime table config with the given broker and server tenants.
+ *
+ * @param realtimeTableConfigNode The JSON object representing the realtime table config.
+ * @param brokerTenant The broker tenant to set in the config.
+ * @param serverTenant The server tenant to set in the config.
+ */
+ @VisibleForTesting
+ static void tweakRealtimeTableConfig(ObjectNode realtimeTableConfigNode, String brokerTenant, String serverTenant,
+ @Nullable Map<String, String> tagPoolReplacementMap) {
+ ObjectNode tenantConfig = (ObjectNode) realtimeTableConfigNode.get("tenants");
+ tenantConfig.put("broker", brokerTenant);
+ tenantConfig.put("server", serverTenant);
+ if (tagPoolReplacementMap == null || tagPoolReplacementMap.isEmpty()) {
+ return;
+ }
+ JsonNode instanceAssignmentConfigMap = realtimeTableConfigNode.get("instanceAssignmentConfigMap");
+ if (instanceAssignmentConfigMap == null) {
+ return;
+ }
+ java.util.Iterator<Map.Entry<String, JsonNode>> iterator = instanceAssignmentConfigMap.fields();
+ while (iterator.hasNext()) {
+ Map.Entry<String, JsonNode> entry = iterator.next();
+ JsonNode instanceAssignmentConfig = entry.getValue();
+ // tagPoolConfig is a required field
+ ObjectNode tagPoolConfig = (ObjectNode) instanceAssignmentConfig.get("tagPoolConfig");
+ // tag is a required json field
+ if (tagPoolConfig != null) {
+ String srcTag = tagPoolConfig.get("tag").asText();
Review Comment:
add a null check for srcTags.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2677047497
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1729,8 +1734,21 @@ public void updateDatabaseConfig(DatabaseConfig databaseConfig) {
*/
public void addTable(TableConfig tableConfig)
throws IOException {
+ addTable(tableConfig, Collections.emptyList());
+ }
+
+ /**
+ * Performs validations of table config and adds the table to zookeeper
Review Comment:
Oh, I just copy the javadoc of original addTable.
/**
* Performs validations of table config and adds the table to zookeeper
* @throws InvalidTableConfigException if validations fail
* @throws TableAlreadyExistsException for offline tables only if the table already exists
*/
public void addTable(TableConfig tableConfig)
throws IOException {
addTable(tableConfig, Collections.emptyList());
}
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2666991674
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -449,6 +451,25 @@ public String getPauselessTableDebugInfo(
}
}
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/tables/{tableName}/consumerWatermarks")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.GET_IDEAL_STATE)
+ @ApiOperation(value = "Get table ideal state", notes = "Get table ideal state")
+ public WatermarkInductionResult inductConsumingWatermark(
+ @ApiParam(value = "Name of the realtime table", required = true) @PathParam("tableName") String tableName,
+ @Context HttpHeaders headers) {
+ try {
+ String table = DatabaseUtils.translateTableName(tableName, headers);
Review Comment:
the segment itself has cdc, likely include the table name
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2663540038
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/WatermarkInductionResult.java:
##########
@@ -0,0 +1,110 @@
+/**
+ * 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.controller.helix.core;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+
+
+/**
+ * Represents the result of a watermark induction process, containing a list of watermarks.
+ */
+public class WatermarkInductionResult {
+
+ private List<Watermark> _watermarks;
+
+ /**
+ * The @JsonCreator annotation marks this constructor to be used for deserializing
+ * a JSON array back into a WaterMarks object.
+ *
+ * @param watermarks The list of watermarks.
+ */
+ @JsonCreator
+ public WatermarkInductionResult(@JsonProperty("watermarks") List<Watermark> watermarks) {
+ _watermarks = watermarks;
+ }
+
+ /**
+ * Gets the list of watermarks.
+ *
+ * @return The list of watermarks.
+ */
+ @JsonGetter("watermarks")
+ public List<Watermark> getWatermarks() {
+ return _watermarks;
+ }
+
+ /**
+ * Represents a single watermark with its partition, sequence, and offset.
+ */
+ public static class Watermark {
+ private long _partitionGroupId;
+ private long _sequenceNumber;
+ private long _offset;
+
+ /**
+ * The @JsonCreator annotation tells Jackson to use this constructor to create
+ * a WaterMark instance from a JSON object. The @JsonProperty annotations
+ * map the keys in the JSON object to the constructor parameters.
+ *
+ * @param partitionGroupId The ID of the partition group.
+ * @param sequenceNumber The sequence number of the watermark.
+ * @param offset The offset of the watermark.
+ */
+ @JsonCreator
+ public Watermark(@JsonProperty("partitionGroupId") long partitionGroupId,
+ @JsonProperty("sequenceNumber") long sequenceNumber, @JsonProperty("offset") long offset) {
+ _partitionGroupId = partitionGroupId;
+ _sequenceNumber = sequenceNumber;
+ _offset = offset;
+ }
+
+ /**
+ * Gets the partition group ID.
+ *
+ * @return The partition group ID.
+ */
+ @JsonGetter("partitionGroupId")
+ public long getPartitionGroupId() {
+ return _partitionGroupId;
+ }
+
+ /**
+ * Gets the sequence number.
+ *
+ * @return The sequence number.
+ */
+ @JsonGetter("sequenceNumber")
+ public long getSequenceNumber() {
+ return _sequenceNumber;
+ }
+
+ /**
+ * Gets the offset.
Review Comment:
Add a javadoc to explain which kafka offset it represents
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2673507545
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PartitionGroupInfo.java:
##########
@@ -0,0 +1,78 @@
+/**
+ * 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.controller.helix.core.realtime;
+
+import org.apache.pinot.controller.helix.core.WatermarkInductionResult;
+import org.apache.pinot.spi.stream.LongMsgOffset;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+
+
+/**
+ * The {@code PartitionGroupInfo} class represents the metadata and sequence number for a partition group.
+ * It encapsulates the {@link PartitionGroupMetadata} and the sequence number associated with it.
+ */
+public class PartitionGroupInfo {
Review Comment:
why do we need any wrapper class like this? Can we reuse PartitionGroupMetadata or add to it instead of creating a new 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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2629944309
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}, tableConfig: {}",
+ tableName, realtimeTableConfig.toString());
+
+ URI watermarkUri = new URI(sourceControllerUri + "/tables/" + tableName + "/consumerWatermarks");
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ List<PartitionGroupInfo> partitionGroupInfos = watermarkInductionResult.getWatermarks().stream()
+ .map(PartitionGroupInfo::from)
+ .collect(Collectors.toList());
+
+ _pinotHelixResourceManager.addTable(realtimeTableConfig, partitionGroupInfos);
+ }
+ LOGGER.info("[copyTable] Finished Table Config copy: {}", tableName);
+ return new CopyTableResponse("success");
+ } catch (Exception e) {
+ LOGGER.error("[copyTable] Error copying table: {}", tableName, e);
+ throw new ControllerApplicationException(LOGGER, "Error copying table: " + e.getMessage(),
+ Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
+
+ /**
+ * Tweaks the realtime table config with the given broker and server tenants.
+ *
+ * @param realtimeTableConfigNode The JSON object representing the realtime table config.
+ * @param brokerTenant The broker tenant to set in the config.
+ * @param serverTenant The server tenant to set in the config.
+ */
+ @VisibleForTesting
+ static void tweakRealtimeTableConfig(ObjectNode realtimeTableConfigNode, String brokerTenant, String serverTenant,
+ @Nullable Map<String, String> tagPoolReplacementMap) {
+ ObjectNode tenantConfig = (ObjectNode) realtimeTableConfigNode.get("tenants");
+ tenantConfig.put("broker", brokerTenant);
Review Comment:
TenantConfig is required for the TableConfig constructor. Should be safe to skip the null check?
https://sourcegraph.com/github.com/apache/pinot@2962dd445fbd25e70d627afca14c96e4b50df5fd/-/blob/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TableConfig.java?L121
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2630019232
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4737,6 +4761,37 @@ public QueryWorkloadManager getQueryWorkloadManager() {
return _queryWorkloadManager;
}
+ public WatermarkInductionResult inductConsumingWatermarks(String tableName) throws TableNotFoundException {
+ String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ if (!hasRealtimeTable(tableName)) {
+ throw new TableNotFoundException("Table " + tableNameWithType + " does not exist");
+ }
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Preconditions.checkNotNull(tableConfig, "Table " + tableNameWithType + "exists but null tableConfig");
+ List<StreamConfig> streamConfigs = IngestionConfigUtils.getStreamConfigs(tableConfig);
+ IdealState idealState = _helixAdmin
+ .getResourceIdealState(getHelixClusterName(), tableNameWithType);
Review Comment:
add null check. 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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2621389583
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}, tableConfig: {}",
+ tableName, realtimeTableConfig.toString());
+
+ URI watermarkUri = new URI(sourceControllerUri + "/tables/" + tableName + "/consumerWatermarks");
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ List<PartitionGroupInfo> partitionGroupInfos = watermarkInductionResult.getWatermarks().stream()
+ .map(PartitionGroupInfo::from)
+ .collect(Collectors.toList());
+
+ _pinotHelixResourceManager.addTable(realtimeTableConfig, partitionGroupInfos);
+ }
+ LOGGER.info("[copyTable] Finished Table Config copy: {}", tableName);
+ return new CopyTableResponse("success");
+ } catch (Exception e) {
+ LOGGER.error("[copyTable] Error copying table: {}", tableName, e);
+ throw new ControllerApplicationException(LOGGER, "Error copying table: " + e.getMessage(),
+ Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
+
+ /**
+ * Tweaks the realtime table config with the given broker and server tenants.
+ *
+ * @param realtimeTableConfigNode The JSON object representing the realtime table config.
+ * @param brokerTenant The broker tenant to set in the config.
+ * @param serverTenant The server tenant to set in the config.
+ */
+ @VisibleForTesting
+ static void tweakRealtimeTableConfig(ObjectNode realtimeTableConfigNode, String brokerTenant, String serverTenant,
+ @Nullable Map<String, String> tagPoolReplacementMap) {
+ ObjectNode tenantConfig = (ObjectNode) realtimeTableConfigNode.get("tenants");
+ tenantConfig.put("broker", brokerTenant);
Review Comment:
null check for tenantConfig
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2621395272
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
Review Comment:
for next PR, we might need to add a mode param and introduce a dryrun mode for feature safety.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2630022631
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}, tableConfig: {}",
+ tableName, realtimeTableConfig.toString());
+
+ URI watermarkUri = new URI(sourceControllerUri + "/tables/" + tableName + "/consumerWatermarks");
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ List<PartitionGroupInfo> partitionGroupInfos = watermarkInductionResult.getWatermarks().stream()
+ .map(PartitionGroupInfo::from)
+ .collect(Collectors.toList());
+
+ _pinotHelixResourceManager.addTable(realtimeTableConfig, partitionGroupInfos);
+ }
+ LOGGER.info("[copyTable] Finished Table Config copy: {}", tableName);
+ return new CopyTableResponse("success");
+ } catch (Exception e) {
+ LOGGER.error("[copyTable] Error copying table: {}", tableName, e);
+ throw new ControllerApplicationException(LOGGER, "Error copying table: " + e.getMessage(),
+ Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
+
+ /**
+ * Tweaks the realtime table config with the given broker and server tenants.
+ *
+ * @param realtimeTableConfigNode The JSON object representing the realtime table config.
+ * @param brokerTenant The broker tenant to set in the config.
+ * @param serverTenant The server tenant to set in the config.
+ */
+ @VisibleForTesting
+ static void tweakRealtimeTableConfig(ObjectNode realtimeTableConfigNode, String brokerTenant, String serverTenant,
+ @Nullable Map<String, String> tagPoolReplacementMap) {
+ ObjectNode tenantConfig = (ObjectNode) realtimeTableConfigNode.get("tenants");
+ tenantConfig.put("broker", brokerTenant);
Review Comment:
add comment to explain why skip null check because of mandatory field.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2621185919
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/CopyTablePayload.java:
##########
@@ -0,0 +1,84 @@
+/**
+ * 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.controller.api.resources;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CopyTablePayload {
+
+ private String _sourceClusterUri;
+ private Map<String, String> _headers;
+ /**
+ * Broker tenant for the new table.
+ * MUST NOT contain the tenant type suffix, i.e. _BROKER.
+ */
+ private String _brokerTenant;
+ /**
+ * Server tenant for the new table.
+ * MUST NOT contain the tenant type suffix, i.e. _REALTIME or _OFFLINE.
+ */
+ private String _serverTenant;
+
+ private Map<String, String> _tagPoolReplacementMap;
Review Comment:
what is the usage for this param?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2663376742
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/WatermarkInductionResult.java:
##########
@@ -0,0 +1,110 @@
+/**
+ * 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.controller.helix.core;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+
+
+/**
+ * Represents the result of a watermark induction process, containing a list of watermarks.
+ */
+public class WatermarkInductionResult {
+
+ private List<Watermark> _watermarks;
+
+ /**
+ * The @JsonCreator annotation marks this constructor to be used for deserializing
+ * a JSON array back into a WaterMarks object.
+ *
+ * @param watermarks The list of watermarks.
+ */
+ @JsonCreator
+ public WatermarkInductionResult(@JsonProperty("watermarks") List<Watermark> watermarks) {
+ _watermarks = watermarks;
+ }
+
+ /**
+ * Gets the list of watermarks.
+ *
+ * @return The list of watermarks.
+ */
+ @JsonGetter("watermarks")
+ public List<Watermark> getWatermarks() {
+ return _watermarks;
+ }
+
+ /**
+ * Represents a single watermark with its partition, sequence, and offset.
+ */
+ public static class Watermark {
+ private long _partitionGroupId;
+ private long _sequenceNumber;
+ private long _offset;
+
+ /**
+ * The @JsonCreator annotation tells Jackson to use this constructor to create
+ * a WaterMark instance from a JSON object. The @JsonProperty annotations
+ * map the keys in the JSON object to the constructor parameters.
+ *
+ * @param partitionGroupId The ID of the partition group.
+ * @param sequenceNumber The sequence number of the watermark.
+ * @param offset The offset of the watermark.
+ */
+ @JsonCreator
+ public Watermark(@JsonProperty("partitionGroupId") long partitionGroupId,
+ @JsonProperty("sequenceNumber") long sequenceNumber, @JsonProperty("offset") long offset) {
+ _partitionGroupId = partitionGroupId;
+ _sequenceNumber = sequenceNumber;
+ _offset = offset;
+ }
+
+ /**
+ * Gets the partition group ID.
+ *
+ * @return The partition group ID.
+ */
+ @JsonGetter("partitionGroupId")
+ public long getPartitionGroupId() {
+ return _partitionGroupId;
+ }
+
+ /**
+ * Gets the sequence number.
+ *
+ * @return The sequence number.
+ */
+ @JsonGetter("sequenceNumber")
+ public long getSequenceNumber() {
+ return _sequenceNumber;
+ }
+
+ /**
+ * Gets the offset.
Review Comment:
you are right
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2632926517
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
Review Comment:
for upsert and pure offline table, the status will be "fail" in reply. For hybrid table, the status will be "warn" with message specifying only realtime table config and segments is copied
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2629465487
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
Review Comment:
Yeah, for upsert and pauseless table, we should throw the exception.
For offline table or offline part of the hybrid table, we can log as a warning? The hunch is that, the use can write their own internal workflow to backfill the offline table. For hybrid table, the user need to copy realtime part.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #17235:
URL: https://github.com/apache/pinot/pull/17235#issuecomment-3648742146
@Jackie-Jiang @chenboat Could you take a review? The manual integration test gets done already. The purpose of this PR is to copy the schema and table config from the source cluster to the target cluster by replacing the server / broke tenant. And then copy the consuming segment to kick off the message consumption on the new table. The backfill of segments prior to consuming segments will be implemented in a follow up PR
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #17235:
URL: https://github.com/apache/pinot/pull/17235#issuecomment-3717126545
@abhishekbafna I add the dryRun as a boolean flag in the payload. The reply contains schema, modified table config and watermarkResult as well. The write operation (schema and table config addition) is moved to the end.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #17235:
URL: https://github.com/apache/pinot/pull/17235#issuecomment-3704082260
Walk through the design doc with @abhishekbafna on 12/23 and aligned on the direction of the 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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "abhishekbafna (via GitHub)" <gi...@apache.org>.
abhishekbafna commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2668973896
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/CopyTablePayload.java:
##########
@@ -0,0 +1,108 @@
+/**
+ * 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.controller.api.resources;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CopyTablePayload {
+
+ private String _sourceClusterUri;
+ private Map<String, String> _headers;
+ /**
+ * Broker tenant for the new table.
+ * MUST NOT contain the tenant type suffix, i.e. _BROKER.
+ */
+ private String _brokerTenant;
+ /**
+ * Server tenant for the new table.
+ * MUST NOT contain the tenant type suffix, i.e. _REALTIME or _OFFLINE.
+ */
+ private String _serverTenant;
+
+ /**
+ * The instanceAssignmentConfig's tagPoolConfig contains full tenant name. We will use this field to let user specify
+ * the replacement relation from source cluster's full tenant to target cluster's full tenant.
+ */
+ private Map<String, String> _tagPoolReplacementMap;
+
+ private boolean _verbose = false;
+ private boolean _dryRun = true;
Review Comment:
These should be API parameters and not copy payload. The copy payload should only contain table config specific key-value pairs only.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +290,136 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+
+ if (_pinotHelixResourceManager.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(tableName)) != null
+ || _pinotHelixResourceManager.getTableConfig(TableNameBuilder.OFFLINE.tableNameWithType(tableName)) != null) {
+ throw new TableAlreadyExistsException("Table config for " + tableName
+ + " already exists. If this is unexpected, try deleting the table to remove all metadata associated"
+ + " with it before attempting to recreate.");
+ }
+
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ ControllerRequestURLBuilder urlBuilder = ControllerRequestURLBuilder.baseUrl(sourceControllerUri);
+
+ URI schemaUri = new URI(urlBuilder.forTableSchemaGet(tableName));
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+
+ URI tableConfigUri = new URI(urlBuilder.forTableGet(tableName));
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}", tableName);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ URI watermarkUri = new URI(urlBuilder.forConsumerWatermarksGet(tableName));
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ boolean hasRealtime = tableConfigNode.has(TableType.REALTIME.name());
+ if (hasOffline && !hasRealtime) {
+ throw new IllegalArgumentException("pure offline table copy not supported yet");
Review Comment:
IllegalStateException
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +290,136 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+
+ if (_pinotHelixResourceManager.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(tableName)) != null
+ || _pinotHelixResourceManager.getTableConfig(TableNameBuilder.OFFLINE.tableNameWithType(tableName)) != null) {
+ throw new TableAlreadyExistsException("Table config for " + tableName
+ + " already exists. If this is unexpected, try deleting the table to remove all metadata associated"
+ + " with it before attempting to recreate.");
+ }
+
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ ControllerRequestURLBuilder urlBuilder = ControllerRequestURLBuilder.baseUrl(sourceControllerUri);
+
+ URI schemaUri = new URI(urlBuilder.forTableSchemaGet(tableName));
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+
+ URI tableConfigUri = new URI(urlBuilder.forTableGet(tableName));
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}", tableName);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ URI watermarkUri = new URI(urlBuilder.forConsumerWatermarksGet(tableName));
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ boolean hasRealtime = tableConfigNode.has(TableType.REALTIME.name());
+ if (hasOffline && !hasRealtime) {
+ throw new IllegalArgumentException("pure offline table copy not supported yet");
+ }
+
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, copyTablePayload);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ if (realtimeTableConfig.getUpsertConfig() != null) {
+ throw new IllegalArgumentException("upsert table copy not supported");
+ }
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}", tableName);
+
+ if (copyTablePayload.isDryRun()) {
+ return new CopyTableResponse("dryrun",
Review Comment:
status can not be dryrun. It should be success or failure.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +290,136 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+
+ if (_pinotHelixResourceManager.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(tableName)) != null
+ || _pinotHelixResourceManager.getTableConfig(TableNameBuilder.OFFLINE.tableNameWithType(tableName)) != null) {
+ throw new TableAlreadyExistsException("Table config for " + tableName
+ + " already exists. If this is unexpected, try deleting the table to remove all metadata associated"
+ + " with it before attempting to recreate.");
+ }
+
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ ControllerRequestURLBuilder urlBuilder = ControllerRequestURLBuilder.baseUrl(sourceControllerUri);
+
+ URI schemaUri = new URI(urlBuilder.forTableSchemaGet(tableName));
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+
+ URI tableConfigUri = new URI(urlBuilder.forTableGet(tableName));
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}", tableName);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ URI watermarkUri = new URI(urlBuilder.forConsumerWatermarksGet(tableName));
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ boolean hasRealtime = tableConfigNode.has(TableType.REALTIME.name());
+ if (hasOffline && !hasRealtime) {
+ throw new IllegalArgumentException("pure offline table copy not supported yet");
+ }
+
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, copyTablePayload);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ if (realtimeTableConfig.getUpsertConfig() != null) {
+ throw new IllegalArgumentException("upsert table copy not supported");
+ }
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}", tableName);
+
+ if (copyTablePayload.isDryRun()) {
+ return new CopyTableResponse("dryrun",
+ "Dry run successful. Fetched schema, table config and watermarks from source cluster.", schema,
+ realtimeTableConfig, watermarkInductionResult);
+ }
+
+ List<PartitionGroupInfo> partitionGroupInfos = watermarkInductionResult.getWatermarks().stream()
+ .map(PartitionGroupInfo::from)
+ .collect(Collectors.toList());
+
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}", tableName);
+ // Add the table with designated starting kafka offset and segment sequence number to create consuming segments
+ _pinotHelixResourceManager.addTable(realtimeTableConfig, partitionGroupInfos);
+ LOGGER.info("[copyTable] Successfully added table config: {} with designated high watermark", tableName);
+ // hybrid table
+ if (hasOffline) {
+ return copyTablePayload.isVerbose() ? new CopyTableResponse("warn",
+ "detect offline too; it will only copy real-time segments", schema,
+ realtimeTableConfig, watermarkInductionResult)
+ : new CopyTableResponse("warn", "detect offline too; it will only copy real-time segments", null, null,
+ null);
Review Comment:
Instead, create response object early on and set specific fields based on the logic and set the verbose fields at the end before returning the response. This would simplify all the verbose logic.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +290,136 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+
+ if (_pinotHelixResourceManager.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(tableName)) != null
+ || _pinotHelixResourceManager.getTableConfig(TableNameBuilder.OFFLINE.tableNameWithType(tableName)) != null) {
+ throw new TableAlreadyExistsException("Table config for " + tableName
+ + " already exists. If this is unexpected, try deleting the table to remove all metadata associated"
+ + " with it before attempting to recreate.");
+ }
+
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ ControllerRequestURLBuilder urlBuilder = ControllerRequestURLBuilder.baseUrl(sourceControllerUri);
+
+ URI schemaUri = new URI(urlBuilder.forTableSchemaGet(tableName));
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+
+ URI tableConfigUri = new URI(urlBuilder.forTableGet(tableName));
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}", tableName);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ URI watermarkUri = new URI(urlBuilder.forConsumerWatermarksGet(tableName));
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ boolean hasRealtime = tableConfigNode.has(TableType.REALTIME.name());
+ if (hasOffline && !hasRealtime) {
+ throw new IllegalArgumentException("pure offline table copy not supported yet");
+ }
+
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, copyTablePayload);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ if (realtimeTableConfig.getUpsertConfig() != null) {
+ throw new IllegalArgumentException("upsert table copy not supported");
+ }
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}", tableName);
+
+ if (copyTablePayload.isDryRun()) {
+ return new CopyTableResponse("dryrun",
+ "Dry run successful. Fetched schema, table config and watermarks from source cluster.", schema,
+ realtimeTableConfig, watermarkInductionResult);
+ }
+
+ List<PartitionGroupInfo> partitionGroupInfos = watermarkInductionResult.getWatermarks().stream()
+ .map(PartitionGroupInfo::from)
+ .collect(Collectors.toList());
+
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}", tableName);
+ // Add the table with designated starting kafka offset and segment sequence number to create consuming segments
+ _pinotHelixResourceManager.addTable(realtimeTableConfig, partitionGroupInfos);
+ LOGGER.info("[copyTable] Successfully added table config: {} with designated high watermark", tableName);
+ // hybrid table
+ if (hasOffline) {
+ return copyTablePayload.isVerbose() ? new CopyTableResponse("warn",
+ "detect offline too; it will only copy real-time segments", schema,
+ realtimeTableConfig, watermarkInductionResult)
+ : new CopyTableResponse("warn", "detect offline too; it will only copy real-time segments", null, null,
+ null);
+ }
+ return copyTablePayload.isVerbose() ? new CopyTableResponse("success", "", schema,
Review Comment:
`msg` parameter, instead of empty, can we put a simple success message.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1903,17 +1915,31 @@ private LLCSegmentName getNextLLCSegmentName(LLCSegmentName lastLLCSegmentName,
private String setupNewPartitionGroup(TableConfig tableConfig, StreamConfig streamConfig,
PartitionGroupMetadata partitionGroupMetadata, long creationTimeMs, InstancePartitions instancePartitions,
int numPartitions, int numReplicas) {
+ return setupNewPartitionGroup(tableConfig, streamConfig, new PartitionGroupInfo(partitionGroupMetadata,
+ STARTING_SEQUENCE_NUMBER), creationTimeMs, instancePartitions, numPartitions, numReplicas);
+ }
+
+ /**
+ * Sets up a new partition group.
+ * <p>Persists the ZK metadata for the first CONSUMING segment, and returns the segment name.
+ */
+ private String setupNewPartitionGroup(TableConfig tableConfig, StreamConfig streamConfig,
+ PartitionGroupInfo wrapper, long creationTimeMs, InstancePartitions instancePartitions,
+ int numPartitions, int numReplicas) {
String realtimeTableName = tableConfig.getTableName();
- int partitionGroupId = partitionGroupMetadata.getPartitionGroupId();
- String startOffset = partitionGroupMetadata.getStartOffset().toString();
+ int partitionGroupId = wrapper.getPartitionGroupId();
+ String startOffset = wrapper.getStartOffset().toString();
LOGGER.info("Setting up new partition group: {} for table: {}", partitionGroupId, realtimeTableName);
+ LOGGER.info("[copyTable] Setting up new partition group: {} for table: {} with sequence: {} and startOffset: {}",
Review Comment:
Remove this log.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +290,136 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+
+ if (_pinotHelixResourceManager.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(tableName)) != null
+ || _pinotHelixResourceManager.getTableConfig(TableNameBuilder.OFFLINE.tableNameWithType(tableName)) != null) {
+ throw new TableAlreadyExistsException("Table config for " + tableName
+ + " already exists. If this is unexpected, try deleting the table to remove all metadata associated"
+ + " with it before attempting to recreate.");
+ }
+
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ ControllerRequestURLBuilder urlBuilder = ControllerRequestURLBuilder.baseUrl(sourceControllerUri);
+
+ URI schemaUri = new URI(urlBuilder.forTableSchemaGet(tableName));
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+
+ URI tableConfigUri = new URI(urlBuilder.forTableGet(tableName));
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}", tableName);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ URI watermarkUri = new URI(urlBuilder.forConsumerWatermarksGet(tableName));
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ boolean hasRealtime = tableConfigNode.has(TableType.REALTIME.name());
+ if (hasOffline && !hasRealtime) {
+ throw new IllegalArgumentException("pure offline table copy not supported yet");
+ }
+
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, copyTablePayload);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ if (realtimeTableConfig.getUpsertConfig() != null) {
+ throw new IllegalArgumentException("upsert table copy not supported");
Review Comment:
IllegalStateException
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -376,17 +376,29 @@ public void stop() {
* <p>NOTE: the passed in IdealState may contain HLC segments if both HLC and LLC are configured.
*/
public void setUpNewTable(TableConfig tableConfig, IdealState idealState) {
+ List<StreamConfig> streamConfigs = IngestionConfigUtils.getStreamConfigs(tableConfig);
+ List<PartitionGroupInfo> newPartitionGroupMetadataList =
+ getNewPartitionGroupMetadataList(streamConfigs, Collections.emptyList(), idealState).stream().map(
+ x -> new PartitionGroupInfo(x, STARTING_SEQUENCE_NUMBER)
+ ).collect(Collectors.toList());
+ setUpNewTable(tableConfig, idealState, newPartitionGroupMetadataList);
+ }
+
+ /**
+ * Sets up the initial segments for a new LLC real-time table.
+ * <p>NOTE: the passed in IdealState may contain HLC segments if both HLC and LLC are configured.
+ */
+ public void setUpNewTable(TableConfig tableConfig, IdealState idealState,
+ List<PartitionGroupInfo> consumeMeta) {
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
- LOGGER.info("Setting up new LLC table: {}", realtimeTableName);
+ LOGGER.info("[copyTable] Setting up new LLC table: {}", realtimeTableName);
Review Comment:
We should revert this change. This is optimizing for the table copy while most operations would be regular table creations.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4715,6 +4739,51 @@ public QueryWorkloadManager getQueryWorkloadManager() {
return _queryWorkloadManager;
}
+ /**
+ * Retrieves the consumer watermark for a given real-time table.
+ * <p>The watermark represents the next offset to be consumed for each partition group.
+ * If the latest segment of a partition is in a DONE state, the watermark is the end offset of the completed segment.
+ * Otherwise, it is the start offset of the current consuming segment.
+ *
+ * @param tableName The name of the real-time table (without type suffix).
+ * @return A {@link WatermarkInductionResult} containing a list of watermarks for each partition group.
+ * @throws TableNotFoundException if the specified real-time table does not exist.
+ * @throws IllegalStateException if the IdealState for the table is not found.
+ */
+ public WatermarkInductionResult getConsumerWatermarks(String tableName) throws TableNotFoundException {
+ String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ if (!hasRealtimeTable(tableName)) {
+ throw new TableNotFoundException("Table " + tableNameWithType + " does not exist");
+ }
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Preconditions.checkNotNull(tableConfig, "Table " + tableNameWithType + "exists but null tableConfig");
+ List<StreamConfig> streamConfigs = IngestionConfigUtils.getStreamConfigs(tableConfig);
+ IdealState idealState = _helixAdmin
+ .getResourceIdealState(getHelixClusterName(), tableNameWithType);
+ if (idealState == null) {
+ throw new IllegalStateException("Null IdealState of the table " + tableNameWithType);
+ }
+ List<PartitionGroupConsumptionStatus> lst = _pinotLLCRealtimeSegmentManager
+ .getPartitionGroupConsumptionStatusList(idealState, streamConfigs);
+ List<WatermarkInductionResult.Watermark> watermarks = lst.stream().map(status -> {
+ int seq = status.getSequenceNumber();
+ long startOffset;
+ try {
+ if ("DONE".equalsIgnoreCase(status.getStatus())) {
+ Preconditions.checkNotNull(status.getEndOffset());
+ startOffset = NumberUtils.parseLong(status.getEndOffset().toString());
+ seq++;
+ } else {
Review Comment:
Add a test case for the else path 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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2670884310
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4715,6 +4739,51 @@ public QueryWorkloadManager getQueryWorkloadManager() {
return _queryWorkloadManager;
}
+ /**
+ * Retrieves the consumer watermark for a given real-time table.
+ * <p>The watermark represents the next offset to be consumed for each partition group.
+ * If the latest segment of a partition is in a DONE state, the watermark is the end offset of the completed segment.
+ * Otherwise, it is the start offset of the current consuming segment.
+ *
+ * @param tableName The name of the real-time table (without type suffix).
+ * @return A {@link WatermarkInductionResult} containing a list of watermarks for each partition group.
+ * @throws TableNotFoundException if the specified real-time table does not exist.
+ * @throws IllegalStateException if the IdealState for the table is not found.
+ */
+ public WatermarkInductionResult getConsumerWatermarks(String tableName) throws TableNotFoundException {
+ String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ if (!hasRealtimeTable(tableName)) {
+ throw new TableNotFoundException("Table " + tableNameWithType + " does not exist");
+ }
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Preconditions.checkNotNull(tableConfig, "Table " + tableNameWithType + "exists but null tableConfig");
+ List<StreamConfig> streamConfigs = IngestionConfigUtils.getStreamConfigs(tableConfig);
+ IdealState idealState = _helixAdmin
+ .getResourceIdealState(getHelixClusterName(), tableNameWithType);
+ if (idealState == null) {
+ throw new IllegalStateException("Null IdealState of the table " + tableNameWithType);
+ }
+ List<PartitionGroupConsumptionStatus> lst = _pinotLLCRealtimeSegmentManager
+ .getPartitionGroupConsumptionStatusList(idealState, streamConfigs);
+ List<WatermarkInductionResult.Watermark> watermarks = lst.stream().map(status -> {
+ int seq = status.getSequenceNumber();
+ long startOffset;
+ try {
+ if ("DONE".equalsIgnoreCase(status.getStatus())) {
+ Preconditions.checkNotNull(status.getEndOffset());
+ startOffset = NumberUtils.parseLong(status.getEndOffset().toString());
+ seq++;
+ } else {
Review Comment:
The testGetConsumerWatermarks already contains both cases.
```
PartitionGroupConsumptionStatus doneStatus = new PartitionGroupConsumptionStatus(0, 100,
new LongMsgOffset(123), new LongMsgOffset(456), "done");
PartitionGroupConsumptionStatus inProgressStatus =
new PartitionGroupConsumptionStatus(1, 200, new LongMsgOffset(789), null, "IN_PROGRESS");
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2673535750
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PartitionGroupInfo.java:
##########
@@ -0,0 +1,78 @@
+/**
+ * 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.controller.helix.core.realtime;
+
+import org.apache.pinot.controller.helix.core.WatermarkInductionResult;
+import org.apache.pinot.spi.stream.LongMsgOffset;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+
+
+/**
+ * The {@code PartitionGroupInfo} class represents the metadata and sequence number for a partition group.
+ * It encapsulates the {@link PartitionGroupMetadata} and the sequence number associated with it.
+ */
+public class PartitionGroupInfo {
Review Comment:
From a pure Kafka consumer point of view, the segment sequence number doesn't seem to belong to a PartitionGroupMeta? I
I am also concerned that the change of the constructor and adding the sequence number might cause review rejections. Adding a wrapper class would bring benefit of less existing code changes.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2673535750
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PartitionGroupInfo.java:
##########
@@ -0,0 +1,78 @@
+/**
+ * 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.controller.helix.core.realtime;
+
+import org.apache.pinot.controller.helix.core.WatermarkInductionResult;
+import org.apache.pinot.spi.stream.LongMsgOffset;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+
+
+/**
+ * The {@code PartitionGroupInfo} class represents the metadata and sequence number for a partition group.
+ * It encapsulates the {@link PartitionGroupMetadata} and the sequence number associated with it.
+ */
+public class PartitionGroupInfo {
Review Comment:
From a pure Kafka consumer point of view, the segment sequence number doesn't seem to belong to a PartitionGroupMeta? I
I am also concerned that the change of the constructor and adding the sequence number might cause review rejections. Many places already use the class PartitonGroupMeta. Adding a wrapper class would bring benefit of less existing code changes.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #17235:
URL: https://github.com/apache/pinot/pull/17235#issuecomment-3551422628
### :x: 3 Tests Failed:
| Tests completed | Failed | Passed | Skipped |
|---|---|---|---|
| 7180 | 3 | 7177 | 0 |
<details><summary>View the top 3 failed test(s) by shortest run time</summary>
> <pre><code class="language-python">org.apache.pinot.controller.helix.core.PinotHelixResourceManagerStatelessTest::testGetLiveBrokers</code></pre>
> <details><summary>Stack Traces | 0.006s run time</summary>
>
> > <pre><code class="language-python">No schema defined for table: testTable_OFFLINE</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.helix.core.PinotHelixResourceManagerStatelessTest::testGetLiveBrokers</code></pre>
> <details><summary>Stack Traces | 0.006s run time</summary>
>
> > <pre><code class="language-python">No schema defined for table: testTable_OFFLINE</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.controller.helix.core.PinotHelixResourceManagerStatelessTest::testGetLiveBrokers</code></pre>
> <details><summary>Stack Traces | 0.009s run time</summary>
>
> > <pre><code class="language-python">No schema defined for table: testTable_OFFLINE</code></pre>
>
> </details>
</details>
To view more test analytics, go to the [Test Analytics Dashboard](https://app.codecov.io/gh/apache/pinot/tests/wirybeaver%3Aoss%2FtableCopyConsuming)
<sub>๐ Got 3 mins? [Take this short survey](https://forms.gle/22i53Qa1CySZjA6c7) to help us improve Test Analytics.</sub>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on PR #17235:
URL: https://github.com/apache/pinot/pull/17235#issuecomment-3750860005
LGTM
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2548127822
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/CopyTablePayload.java:
##########
@@ -0,0 +1,84 @@
+/**
+ * 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.controller.api.resources;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CopyTablePayload {
+
+ String _sourceClusterUri;
+ Map<String, String> _headers;
+ /**
+ * Broker tenant for the new table.
+ * MUST NOT contain the tenant type suffix, i.e. _BROKER.
+ */
+ String _brokerTenant;
+ /**
+ * Server tenant for the new table.
+ * MUST NOT contain the tenant type suffix, i.e. _REALTIME or _OFFLINE.
+ */
+ String _serverTenant;
+
+ Map<String, String> _tagPoolReplacementMap;
+
+ @JsonCreator
+ public CopyTablePayload(@JsonProperty(value = "tableName", required = true) String tableName,
+ @JsonProperty(value = "sourceClusterUri", required = true) String sourceClusterUri,
+ @JsonProperty("sourceClusterHeaders") Map<String, String> headers,
+ @JsonProperty(value = "brokerTenant", required = true) String brokerTenant,
+ @JsonProperty(value = "serverTenant", required = true) String serverTenant,
+ @JsonProperty("tagPoolReplacementMap") @Nullable Map<String, String> tagPoolReplacementMap) {
+ _sourceClusterUri = sourceClusterUri;
+ _headers = headers;
Review Comment:
tableName is unused?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2629468263
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
Review Comment:
I don't think we need it because it essentially do table copy into a different cluster. User can safely delete the table copy.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2629468263
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
Review Comment:
Use can call the source clusters GET /tables/<table>/consumerWatermarks to get the dry run result, which will show watermarks as well as segments will be backfilled.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2621398775
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4737,6 +4761,37 @@ public QueryWorkloadManager getQueryWorkloadManager() {
return _queryWorkloadManager;
}
+ public WatermarkInductionResult inductConsumingWatermarks(String tableName) throws TableNotFoundException {
+ String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ if (!hasRealtimeTable(tableName)) {
+ throw new TableNotFoundException("Table " + tableNameWithType + " does not exist");
+ }
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Preconditions.checkNotNull(tableConfig, "Table " + tableNameWithType + "exists but null tableConfig");
+ List<StreamConfig> streamConfigs = IngestionConfigUtils.getStreamConfigs(tableConfig);
+ IdealState idealState = _helixAdmin
+ .getResourceIdealState(getHelixClusterName(), tableNameWithType);
+ List<PartitionGroupConsumptionStatus> lst = _pinotLLCRealtimeSegmentManager
+ .getPartitionGroupConsumptionStatusList(idealState, streamConfigs);
+ List<WatermarkInductionResult.Watermark> watermarks = lst.stream().map(status -> {
+ long seq = status.getSequenceNumber();
+ long startOffset;
+ try {
+ if (status.getStatus().equalsIgnoreCase("done")) {
Review Comment:
"DONE".equalsIgnoreCase(status.getStatus()) for null handling?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #17235:
URL: https://github.com/apache/pinot/pull/17235#issuecomment-3648733875
This Integration Test was not stable.
TableRebalancePauselessIntegrationTest.testForceCommit:143->BaseClusterIntegrationTestSet.waitForRebalanceToComplete:850 Failed to meet condition in 600000ms, error message: Failed to complete rebalance
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2663327546
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -449,6 +451,25 @@ public String getPauselessTableDebugInfo(
}
}
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/tables/{tableName}/consumerWatermarks")
Review Comment:
better to be consistent with method name. change to addConsumingWatermarks?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #17235:
URL: https://github.com/apache/pinot/pull/17235#issuecomment-3807332593
Can you please create a parent issue along with the design doc, and then link all the PRs to the parent issue? It is easier to track all the PRs under this topic
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2666889807
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ if (realtimeTableConfig.getUpsertConfig() != null) {
+ return new CopyTableResponse("fail", "upsert table copy not supported");
+ }
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}, tableConfig: {}",
+ tableName, realtimeTableConfig.toString());
+
+ URI watermarkUri = new URI(sourceControllerUri + "/tables/" + tableName + "/consumerWatermarks");
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ List<PartitionGroupInfo> partitionGroupInfos = watermarkInductionResult.getWatermarks().stream()
+ .map(PartitionGroupInfo::from)
+ .collect(Collectors.toList());
+
+ // Add the table with designated starting kafka offset and segment sequence number to create consuming segments
+ _pinotHelixResourceManager.addTable(realtimeTableConfig, partitionGroupInfos);
+ if (hasOffline) {
+ return new CopyTableResponse("warn", "detect offline; copy real-time segments only");
+ }
+ return new CopyTableResponse("success", "");
+ }
+ return new CopyTableResponse("fail", "copying offline table's segments is not supported yet");
Review Comment:
yeah, throw exception
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2673428063
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1729,8 +1734,21 @@ public void updateDatabaseConfig(DatabaseConfig databaseConfig) {
*/
public void addTable(TableConfig tableConfig)
throws IOException {
+ addTable(tableConfig, Collections.emptyList());
+ }
+
+ /**
+ * Performs validations of table config and adds the table to zookeeper
Review Comment:
Is there any restriction on this API? Realtime table only? Can it add upsert tables? If so, please clarify in the javadoc.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2629468263
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
Review Comment:
User can call the source clusters GET /tables/<table>/consumerWatermarks to get the dry run result, which will show watermarks as well as segments will be backfilled.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2608338598
##########
pinot-controller/src/test/resources/table/table_config_with_instance_assignment.json:
##########
@@ -0,0 +1,15 @@
+{
+ "tenants": {
+ "broker": "broker1",
+ "server": "rtaShared2"
Review Comment:
nit: rtaShared2 --> server1
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2621397588
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4737,6 +4761,37 @@ public QueryWorkloadManager getQueryWorkloadManager() {
return _queryWorkloadManager;
}
+ public WatermarkInductionResult inductConsumingWatermarks(String tableName) throws TableNotFoundException {
+ String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ if (!hasRealtimeTable(tableName)) {
+ throw new TableNotFoundException("Table " + tableNameWithType + " does not exist");
+ }
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Preconditions.checkNotNull(tableConfig, "Table " + tableNameWithType + "exists but null tableConfig");
+ List<StreamConfig> streamConfigs = IngestionConfigUtils.getStreamConfigs(tableConfig);
+ IdealState idealState = _helixAdmin
+ .getResourceIdealState(getHelixClusterName(), tableNameWithType);
Review Comment:
Can idealState be null?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2629951634
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}, tableConfig: {}",
+ tableName, realtimeTableConfig.toString());
+
+ URI watermarkUri = new URI(sourceControllerUri + "/tables/" + tableName + "/consumerWatermarks");
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ List<PartitionGroupInfo> partitionGroupInfos = watermarkInductionResult.getWatermarks().stream()
+ .map(PartitionGroupInfo::from)
+ .collect(Collectors.toList());
+
+ _pinotHelixResourceManager.addTable(realtimeTableConfig, partitionGroupInfos);
+ }
+ LOGGER.info("[copyTable] Finished Table Config copy: {}", tableName);
+ return new CopyTableResponse("success");
+ } catch (Exception e) {
+ LOGGER.error("[copyTable] Error copying table: {}", tableName, e);
+ throw new ControllerApplicationException(LOGGER, "Error copying table: " + e.getMessage(),
+ Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
+
+ /**
+ * Tweaks the realtime table config with the given broker and server tenants.
+ *
+ * @param realtimeTableConfigNode The JSON object representing the realtime table config.
+ * @param brokerTenant The broker tenant to set in the config.
+ * @param serverTenant The server tenant to set in the config.
+ */
+ @VisibleForTesting
+ static void tweakRealtimeTableConfig(ObjectNode realtimeTableConfigNode, String brokerTenant, String serverTenant,
+ @Nullable Map<String, String> tagPoolReplacementMap) {
+ ObjectNode tenantConfig = (ObjectNode) realtimeTableConfigNode.get("tenants");
+ tenantConfig.put("broker", brokerTenant);
+ tenantConfig.put("server", serverTenant);
+ if (tagPoolReplacementMap == null || tagPoolReplacementMap.isEmpty()) {
+ return;
+ }
+ JsonNode instanceAssignmentConfigMap = realtimeTableConfigNode.get("instanceAssignmentConfigMap");
+ if (instanceAssignmentConfigMap == null) {
+ return;
+ }
+ java.util.Iterator<Map.Entry<String, JsonNode>> iterator = instanceAssignmentConfigMap.fields();
+ while (iterator.hasNext()) {
+ Map.Entry<String, JsonNode> entry = iterator.next();
+ JsonNode instanceAssignmentConfig = entry.getValue();
+ // tagPoolConfig is a required field
+ ObjectNode tagPoolConfig = (ObjectNode) instanceAssignmentConfig.get("tagPoolConfig");
+ // tag is a required json field
+ if (tagPoolConfig != null) {
+ String srcTag = tagPoolConfig.get("tag").asText();
Review Comment:
https://sourcegraph.com/github.com/apache/pinot@2962dd445fbd25e70d627afca14c96e4b50df5fd/-/blob/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/InstanceTagPoolConfig.java?L49
tag is required in the TagPoolConfig constructor. Should be safe to skip the null 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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2630025570
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -449,6 +451,25 @@ public String getPauselessTableDebugInfo(
}
}
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/tables/{tableName}/consumerWatermarks")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.GET_IDEAL_STATE)
+ @ApiOperation(value = "Get table ideal state", notes = "Get table ideal state")
+ public WatermarkInductionResult inductConsumingWatermark(
+ @ApiParam(value = "Name of the realtime table", required = true) @PathParam("tableName") String tableName,
+ @Context HttpHeaders headers) {
+ try {
+ String table = DatabaseUtils.translateTableName(tableName, headers);
Review Comment:
you are right.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #17235:
URL: https://github.com/apache/pinot/pull/17235#issuecomment-3807462325
> Can you please create a parent issue along with the design doc, and then link all the PRs to the parent issue? It is easier to track all the PRs under this topic
Yeah, the parent issue was created before: https://github.com/apache/pinot/issues/16563
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2673411779
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -449,6 +451,25 @@ public String getPauselessTableDebugInfo(
}
}
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/tables/{tableName}/consumerWatermarks")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.GET_IDEAL_STATE)
+ @ApiOperation(value = "Get table ideal state", notes = "Get table ideal state")
Review Comment:
Why is it about "table ideal state"? Is it typo?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2663324436
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/WatermarkInductionResult.java:
##########
@@ -0,0 +1,110 @@
+/**
+ * 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.controller.helix.core;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+
+
+/**
+ * Represents the result of a watermark induction process, containing a list of watermarks.
+ */
+public class WatermarkInductionResult {
+
+ private List<Watermark> _watermarks;
+
+ /**
+ * The @JsonCreator annotation marks this constructor to be used for deserializing
+ * a JSON array back into a WaterMarks object.
+ *
+ * @param watermarks The list of watermarks.
+ */
+ @JsonCreator
+ public WatermarkInductionResult(@JsonProperty("watermarks") List<Watermark> watermarks) {
+ _watermarks = watermarks;
+ }
+
+ /**
+ * Gets the list of watermarks.
+ *
+ * @return The list of watermarks.
+ */
+ @JsonGetter("watermarks")
+ public List<Watermark> getWatermarks() {
+ return _watermarks;
+ }
+
+ /**
+ * Represents a single watermark with its partition, sequence, and offset.
Review Comment:
Here you use partition but the code variable is partitionGroupId? It needs to be consistent.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2663374750
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -449,6 +451,25 @@ public String getPauselessTableDebugInfo(
}
}
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/tables/{tableName}/consumerWatermarks")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.GET_IDEAL_STATE)
+ @ApiOperation(value = "Get table ideal state", notes = "Get table ideal state")
+ public WatermarkInductionResult inductConsumingWatermark(
Review Comment:
I can use get.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2663374219
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -449,6 +451,25 @@ public String getPauselessTableDebugInfo(
}
}
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/tables/{tableName}/consumerWatermarks")
Review Comment:
It's get instead add. The target cluster will invoke this api against the source cluster, the source cluster will return the current consumerWatermarks.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat merged PR #17235:
URL: https://github.com/apache/pinot/pull/17235
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2579487604
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/CopyTablePayload.java:
##########
@@ -0,0 +1,84 @@
+/**
+ * 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.controller.api.resources;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CopyTablePayload {
+
+ String _sourceClusterUri;
+ Map<String, String> _headers;
+ /**
+ * Broker tenant for the new table.
+ * MUST NOT contain the tenant type suffix, i.e. _BROKER.
+ */
+ String _brokerTenant;
+ /**
+ * Server tenant for the new table.
+ * MUST NOT contain the tenant type suffix, i.e. _REALTIME or _OFFLINE.
+ */
+ String _serverTenant;
+
+ Map<String, String> _tagPoolReplacementMap;
+
+ @JsonCreator
+ public CopyTablePayload(@JsonProperty(value = "tableName", required = true) String tableName,
+ @JsonProperty(value = "sourceClusterUri", required = true) String sourceClusterUri,
+ @JsonProperty("sourceClusterHeaders") Map<String, String> headers,
+ @JsonProperty(value = "brokerTenant", required = true) String brokerTenant,
+ @JsonProperty(value = "serverTenant", required = true) String serverTenant,
+ @JsonProperty("tagPoolReplacementMap") @Nullable Map<String, String> tagPoolReplacementMap) {
+ _sourceClusterUri = sourceClusterUri;
+ _headers = headers;
Review Comment:
yep
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2579486964
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -433,6 +435,25 @@ public String getPauselessTableDebugInfo(
}
}
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/tables/{tableName}/watermarks")
Review Comment:
/consumerWatermarks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2630031852
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/CopyTablePayload.java:
##########
@@ -0,0 +1,84 @@
+/**
+ * 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.controller.api.resources;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CopyTablePayload {
+
+ private String _sourceClusterUri;
+ private Map<String, String> _headers;
+ /**
+ * Broker tenant for the new table.
+ * MUST NOT contain the tenant type suffix, i.e. _BROKER.
+ */
+ private String _brokerTenant;
+ /**
+ * Server tenant for the new table.
+ * MUST NOT contain the tenant type suffix, i.e. _REALTIME or _OFFLINE.
+ */
+ private String _serverTenant;
+
+ private Map<String, String> _tagPoolReplacementMap;
Review Comment:
the instanceAssignmentConfig's tagPoolConfig contains full tenant name. We will use this field to let user specify the replacement relation from source cluster's full tenant to target cluster's full tenant.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2630021797
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}, tableConfig: {}",
+ tableName, realtimeTableConfig.toString());
+
+ URI watermarkUri = new URI(sourceControllerUri + "/tables/" + tableName + "/consumerWatermarks");
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ List<PartitionGroupInfo> partitionGroupInfos = watermarkInductionResult.getWatermarks().stream()
+ .map(PartitionGroupInfo::from)
+ .collect(Collectors.toList());
+
+ _pinotHelixResourceManager.addTable(realtimeTableConfig, partitionGroupInfos);
+ }
+ LOGGER.info("[copyTable] Finished Table Config copy: {}", tableName);
+ return new CopyTableResponse("success");
+ } catch (Exception e) {
+ LOGGER.error("[copyTable] Error copying table: {}", tableName, e);
+ throw new ControllerApplicationException(LOGGER, "Error copying table: " + e.getMessage(),
+ Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
+
+ /**
+ * Tweaks the realtime table config with the given broker and server tenants.
+ *
+ * @param realtimeTableConfigNode The JSON object representing the realtime table config.
+ * @param brokerTenant The broker tenant to set in the config.
+ * @param serverTenant The server tenant to set in the config.
+ */
+ @VisibleForTesting
+ static void tweakRealtimeTableConfig(ObjectNode realtimeTableConfigNode, String brokerTenant, String serverTenant,
+ @Nullable Map<String, String> tagPoolReplacementMap) {
+ ObjectNode tenantConfig = (ObjectNode) realtimeTableConfigNode.get("tenants");
+ tenantConfig.put("broker", brokerTenant);
+ tenantConfig.put("server", serverTenant);
+ if (tagPoolReplacementMap == null || tagPoolReplacementMap.isEmpty()) {
+ return;
+ }
+ JsonNode instanceAssignmentConfigMap = realtimeTableConfigNode.get("instanceAssignmentConfigMap");
+ if (instanceAssignmentConfigMap == null) {
+ return;
+ }
+ java.util.Iterator<Map.Entry<String, JsonNode>> iterator = instanceAssignmentConfigMap.fields();
+ while (iterator.hasNext()) {
+ Map.Entry<String, JsonNode> entry = iterator.next();
+ JsonNode instanceAssignmentConfig = entry.getValue();
+ // tagPoolConfig is a required field
+ ObjectNode tagPoolConfig = (ObjectNode) instanceAssignmentConfig.get("tagPoolConfig");
+ // tag is a required json field
+ if (tagPoolConfig != null) {
+ String srcTag = tagPoolConfig.get("tag").asText();
Review Comment:
add comment to explain why skip null check because of mandatory field.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2663328916
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -449,6 +451,25 @@ public String getPauselessTableDebugInfo(
}
}
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/tables/{tableName}/consumerWatermarks")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.GET_IDEAL_STATE)
+ @ApiOperation(value = "Get table ideal state", notes = "Get table ideal state")
+ public WatermarkInductionResult inductConsumingWatermark(
Review Comment:
Consider to use simple and straightforward verb like `add` rather than `induct` to easier understanding.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "abhishekbafna (via GitHub)" <gi...@apache.org>.
abhishekbafna commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2664718282
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
Review Comment:
Use `ControllerRequestURLBuilder` for building URLs.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ if (realtimeTableConfig.getUpsertConfig() != null) {
+ return new CopyTableResponse("fail", "upsert table copy not supported");
+ }
Review Comment:
Lets get the table config at the beginning and do all the validation. We have already added the schema for the upsert table, which is not desirable.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
Review Comment:
Same, use `ControllerRequestURLBuilder`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
Review Comment:
Instead pass the `CopyTablePayload` so that any new parameter in the future would not require method signature change.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
Review Comment:
At high level, the API should be like:
* validate input parameters
* get data from the source cluster (schema and table config)
* modify the table config and schema if needed
* perform any validation needed before making any write operation on the cluster
* perform write operations
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
Review Comment:
It would make to sense to support the dryrun mode directly on the API so that users can get and see the final result for the API.
Making two or more API requests and then manually verifying them is not same as what the code is doing. This should be supported programmatically.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
Review Comment:
Avoid logging table config, as ingestion config contains passwords and secrets, which will be logged in the logs.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ if (realtimeTableConfig.getUpsertConfig() != null) {
+ return new CopyTableResponse("fail", "upsert table copy not supported");
+ }
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}, tableConfig: {}",
+ tableName, realtimeTableConfig.toString());
+
+ URI watermarkUri = new URI(sourceControllerUri + "/tables/" + tableName + "/consumerWatermarks");
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ List<PartitionGroupInfo> partitionGroupInfos = watermarkInductionResult.getWatermarks().stream()
+ .map(PartitionGroupInfo::from)
+ .collect(Collectors.toList());
+
+ // Add the table with designated starting kafka offset and segment sequence number to create consuming segments
+ _pinotHelixResourceManager.addTable(realtimeTableConfig, partitionGroupInfos);
+ if (hasOffline) {
+ return new CopyTableResponse("warn", "detect offline; copy real-time segments only");
+ }
+ return new CopyTableResponse("success", "");
+ }
+ return new CopyTableResponse("fail", "copying offline table's segments is not supported yet");
Review Comment:
This would be a 200 OK response. Instead should send a failure message with 501 code or something.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/WatermarkInductionResult.java:
##########
@@ -0,0 +1,111 @@
+/**
+ * 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.controller.helix.core;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+
+
+/**
+ * Represents the result of a watermark induction process, containing a list of watermarks.
+ */
+public class WatermarkInductionResult {
+
+ private List<Watermark> _watermarks;
+
+ /**
+ * The @JsonCreator annotation marks this constructor to be used for deserializing
+ * a JSON array back into a WaterMarks object.
+ *
+ * @param watermarks The list of watermarks.
+ */
+ @JsonCreator
+ public WatermarkInductionResult(@JsonProperty("watermarks") List<Watermark> watermarks) {
+ _watermarks = watermarks;
+ }
+
+ /**
+ * Gets the list of watermarks.
+ *
+ * @return The list of watermarks.
+ */
+ @JsonGetter("watermarks")
+ public List<Watermark> getWatermarks() {
+ return _watermarks;
+ }
+
+ /**
+ * Represents a single watermark with its partitionGroupId, sequence, and offset.
+ */
+ public static class Watermark {
+ private long _partitionGroupId;
+ private long _sequenceNumber;
+ private long _offset;
Review Comment:
Why this can not be `int`?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1789,10 +1807,16 @@ public void addTable(TableConfig tableConfig)
// Add ideal state
_helixAdmin.addResource(_helixClusterName, tableNameWithType, idealState);
LOGGER.info("Adding table {}: Added ideal state for offline table", tableNameWithType);
- } else {
+ } else if (consumeMeta == null || consumeMeta.isEmpty()) {
// Add ideal state with the first CONSUMING segment
_pinotLLCRealtimeSegmentManager.setUpNewTable(tableConfig, idealState);
LOGGER.info("Adding table {}: Added ideal state with first consuming segment", tableNameWithType);
+ } else {
+ // Add ideal state with the first CONSUMING segment with designated partition consuming metadata
+ // Add ideal state with the first CONSUMING segment
+ _pinotLLCRealtimeSegmentManager.setUpNewTable(tableConfig, idealState, consumeMeta);
+ LOGGER.info("Adding table {}: Added consuming segments ideal state given the designated consuming metadata",
Review Comment:
Add `[copyTable]`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ if (realtimeTableConfig.getUpsertConfig() != null) {
+ return new CopyTableResponse("fail", "upsert table copy not supported");
+ }
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}, tableConfig: {}",
+ tableName, realtimeTableConfig.toString());
+
+ URI watermarkUri = new URI(sourceControllerUri + "/tables/" + tableName + "/consumerWatermarks");
+ SimpleHttpResponse watermarkResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(watermarkUri, requestHeaders));
+ String watermarkJson = watermarkResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched watermarks for table: {}. Result: {}", tableName, watermarkJson);
+ WatermarkInductionResult watermarkInductionResult =
+ JsonUtils.stringToObject(watermarkJson, WatermarkInductionResult.class);
+
+ List<PartitionGroupInfo> partitionGroupInfos = watermarkInductionResult.getWatermarks().stream()
+ .map(PartitionGroupInfo::from)
+ .collect(Collectors.toList());
+
+ // Add the table with designated starting kafka offset and segment sequence number to create consuming segments
+ _pinotHelixResourceManager.addTable(realtimeTableConfig, partitionGroupInfos);
+ if (hasOffline) {
+ return new CopyTableResponse("warn", "detect offline; copy real-time segments only");
Review Comment:
improve the message.
`detected offline table too; it will only copy realtime table.`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4715,6 +4739,51 @@ public QueryWorkloadManager getQueryWorkloadManager() {
return _queryWorkloadManager;
}
+ /**
+ * Retrieves the consumer watermark for a given real-time table.
+ * <p>The watermark represents the next offset to be consumed for each partition group.
+ * If the latest segment of a partition is in a DONE state, the watermark is the end offset of the completed segment.
+ * Otherwise, it is the start offset of the current consuming segment.
+ *
+ * @param tableName The name of the real-time table (without type suffix).
+ * @return A {@link WatermarkInductionResult} containing a list of watermarks for each partition group.
+ * @throws TableNotFoundException if the specified real-time table does not exist.
+ * @throws IllegalStateException if the IdealState for the table is not found.
+ */
+ public WatermarkInductionResult getConsumerWatermarks(String tableName) throws TableNotFoundException {
+ String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
Review Comment:
Move it after the validation check
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -376,17 +376,30 @@ public void stop() {
* <p>NOTE: the passed in IdealState may contain HLC segments if both HLC and LLC are configured.
*/
public void setUpNewTable(TableConfig tableConfig, IdealState idealState) {
+ List<StreamConfig> streamConfigs = IngestionConfigUtils.getStreamConfigs(tableConfig);
+ List<PartitionGroupInfo> newPartitionGroupMetadataList =
+ getNewPartitionGroupMetadataList(streamConfigs, Collections.emptyList(), idealState).stream().map(
+ x -> new PartitionGroupInfo(x, STARTING_SEQUENCE_NUMBER)
+ ).collect(Collectors.toList());
+ setUpNewTable(tableConfig, idealState, newPartitionGroupMetadataList);
+ }
+
+ /**
+ * Sets up the initial segments for a new LLC real-time table.
+ * <p>NOTE: the passed in IdealState may contain HLC segments if both HLC and LLC are configured.
+ */
+ public void setUpNewTable(TableConfig tableConfig, IdealState idealState,
+ List<PartitionGroupInfo> consumeMeta) {
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
LOGGER.info("Setting up new LLC table: {}", realtimeTableName);
+ LOGGER.info("[copyTable] Setting up new LLC table: {}", realtimeTableName);
Review Comment:
Duplicate log statement.
I see them other places in the code, these would created unnecessary log statements for the normal table creation operations.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/CopyTableResponse.java:
##########
@@ -0,0 +1,51 @@
+/**
+ * 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.controller.api.resources;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class CopyTableResponse {
Review Comment:
Consider adding schema and table config in the response for the users to verify in case needed. They can be hidden under a boolean verbose flag is needed. This will further enable to implement dryrun method for the API in future.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
+ if (tableConfigNode.has(TableType.REALTIME.name())) {
+ ObjectNode realtimeTableConfigNode = (ObjectNode) tableConfigNode.get(TableType.REALTIME.name());
+ tweakRealtimeTableConfig(realtimeTableConfigNode, brokerTenant, serverTenant, tagReplacementMap);
+ TableConfig realtimeTableConfig = JsonUtils.jsonNodeToObject(realtimeTableConfigNode, TableConfig.class);
+ if (realtimeTableConfig.getUpsertConfig() != null) {
+ return new CopyTableResponse("fail", "upsert table copy not supported");
+ }
+ LOGGER.info("[copyTable] Successfully fetched and tweaked table config for table: {}, tableConfig: {}",
+ tableName, realtimeTableConfig.toString());
+
+ URI watermarkUri = new URI(sourceControllerUri + "/tables/" + tableName + "/consumerWatermarks");
Review Comment:
Same, use `ControllerRequestURLBuilder`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,118 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName, String payload,
+ @Context HttpHeaders headers) {
+ try {
+ LOGGER.info("[copyTable] received request for table: {}, payload: {}", tableName, payload);
+ tableName = DatabaseUtils.translateTableName(tableName, headers);
+ CopyTablePayload copyTablePayload = JsonUtils.stringToObject(payload, CopyTablePayload.class);
+ String sourceControllerUri = copyTablePayload.getSourceClusterUri();
+ Map<String, String> requestHeaders = copyTablePayload.getHeaders();
+ String brokerTenant = copyTablePayload.getBrokerTenant();
+ String serverTenant = copyTablePayload.getServerTenant();
+ Map<String, String> tagReplacementMap = copyTablePayload.getTagPoolReplacementMap();
+
+ LOGGER.info("[copyTable] Start copying table: {} from source: {}", tableName, sourceControllerUri);
+
+ // Fetch and add schema
+ URI schemaUri = new URI(sourceControllerUri + "/tables/" + tableName + "/schema");
+ SimpleHttpResponse schemaResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(schemaUri, requestHeaders));
+ String schemaJson = schemaResponse.getResponse();
+ Schema schema = Schema.fromString(schemaJson);
+ _pinotHelixResourceManager.addSchema(schema, true, false);
+ LOGGER.info("[copyTable] Successfully added schema for table: {}, schema: {}", tableName, schema);
+
+ // Fetch and add table configs
+ URI tableConfigUri = new URI(sourceControllerUri + "/tables/" + tableName);
+ SimpleHttpResponse tableConfigResponse = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendGetRequest(tableConfigUri, requestHeaders));
+ String tableConfigJson = tableConfigResponse.getResponse();
+ LOGGER.info("[copyTable] Fetched table config for table: {}, tableConfig: {}", tableName, tableConfigJson);
+ JsonNode tableConfigNode = JsonUtils.stringToJsonNode(tableConfigJson);
+
+ boolean hasOffline = tableConfigNode.has(TableType.OFFLINE.name());
Review Comment:
Move it inside the if condition.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2548132800
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/CopyTablePayload.java:
##########
@@ -0,0 +1,84 @@
+/**
+ * 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.controller.api.resources;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CopyTablePayload {
+
+ String _sourceClusterUri;
Review Comment:
private?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #17235:
URL: https://github.com/apache/pinot/pull/17235#issuecomment-3717213852
MSE against 1.3 failed but succeed against 1.4 and master
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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
Re: [PR] [Real-time Table Replication X clusters][1/n] Creating new tables with designated consuming segments [pinot]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #17235:
URL: https://github.com/apache/pinot/pull/17235#discussion_r2666991156
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -283,6 +289,109 @@ public ConfigSuccessResponse addTable(String tableConfigStr,
}
}
+ @POST
+ @Path("/tables/{tableName}/copy")
+ @Authorize(targetType = TargetType.TABLE, action = Actions.Table.CREATE_TABLE)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Copy a table's schema and config from another cluster", notes = "Non upsert table only")
+ public CopyTableResponse copyTable(
Review Comment:
Yeah, already support the dryrun mode with a boolean flag in the input payload
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
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