HTHou commented on code in PR #18132:
URL: https://github.com/apache/iotdb/pull/18132#discussion_r3557799300
##########
iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java:
##########
@@ -2857,6 +2867,491 @@ public void insertRelationalTablet(Tablet tablet)
}
}
+ /**
+ * insert relational Tablets.
+ *
+ * @param tablets data batches
+ */
+ public void insertRelationalTablets(List<Tablet> tablets)
+ throws IoTDBConnectionException, StatementExecutionException {
+ final boolean recordMergeTabletsCost = enableMergeTablets;
+ long mergeTabletsCost = 0;
+ if (enableMergeTablets) {
+ final long mergeTabletsStartTime = System.nanoTime();
+ tablets = mergeRelationalTablets(tablets);
+ mergeTabletsCost = System.nanoTime() - mergeTabletsStartTime;
+ }
+ if (tablets.isEmpty()) {
+ throw new BatchExecutionException(SessionMessages.NO_TABLET_INSERTING);
+ }
+ final long insertTabletsStartTime = System.nanoTime();
+ if (enableRedirection) {
+ insertRelationalTabletsWithLeaderCache(tablets);
+ } else {
+ TSInsertTabletsReq request = genTSInsertTabletsReq(tablets, false,
false);
+ request.setWriteToTable(true);
+ for (Tablet tablet : tablets) {
+
request.addToColumnCategoriesList(toEnumOrdinalsAsBytes(tablet.getColumnTypes()));
+ }
+ try {
+ getDefaultSessionConnection().insertTablets(request);
+ } catch (RedirectException ignored) {
+ }
+ }
+ if (recordMergeTabletsCost) {
+ recordMergeTabletsCost(mergeTabletsCost, System.nanoTime() -
insertTabletsStartTime);
+ }
+ }
+
+ private void insertRelationalTabletsWithLeaderCache(final List<Tablet>
tablets)
+ throws IoTDBConnectionException, StatementExecutionException {
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup = new
HashMap<>();
+ for (final Tablet tablet : tablets) {
+ addRelationalTabletToGroup(tabletGroup, tablet);
+ }
+ if (tabletGroup.size() == 1) {
+ insertRelationalTabletsOnce(tabletGroup);
+ } else {
+ insertRelationalTabletsByGroup(tabletGroup);
+ }
+ }
+
+ private void addRelationalTabletToGroup(
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup, final
Tablet tablet)
+ throws IoTDBConnectionException {
+ if (SessionUtils.isTabletContainsSingleDevice(tablet)) {
+ final SessionConnection connection =
getTableModelSessionConnection(tablet.getDeviceID(0));
+ updateRelationalTabletsReq(tabletGroup, connection, tablet);
+ return;
+ }
+ final Map<SessionConnection, Map<IDeviceID, Tablet>>
subTabletsByConnection =
+ new LinkedHashMap<>();
+ for (int row = 0; row < tablet.getRowSize(); row++) {
+ final IDeviceID deviceID = tablet.getDeviceID(row);
+ final SessionConnection connection =
getTableModelSessionConnection(deviceID);
+ final Tablet subTablet =
+ subTabletsByConnection
+ .computeIfAbsent(connection, ignored -> new LinkedHashMap<>())
+ .computeIfAbsent(deviceID, ignored ->
createEmptyRelationalTabletLike(tablet));
Review Comment:
**High:** This creates one sub-Tablet per distinct device, while
createEmptyRelationalTabletLike allocates every sub-tablet with the original
tablet full rowSize at line 2974. An R-row tablet with D devices therefore
allocates O(D * R * columns) storage, reaching O(R^2) when each row belongs to
a different device. This also happens with an empty leader cache: all devices
map to the default connection, but the inner map still creates D full-capacity
tablets. This can exhaust the client heap before any RPC. Could we group rows
once per endpoint, or pre-count rows per device and allocate exact capacities?
A many-device stress test would also help.
##########
iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java:
##########
@@ -2857,6 +2867,491 @@ public void insertRelationalTablet(Tablet tablet)
}
}
+ /**
+ * insert relational Tablets.
+ *
+ * @param tablets data batches
+ */
+ public void insertRelationalTablets(List<Tablet> tablets)
+ throws IoTDBConnectionException, StatementExecutionException {
+ final boolean recordMergeTabletsCost = enableMergeTablets;
+ long mergeTabletsCost = 0;
+ if (enableMergeTablets) {
+ final long mergeTabletsStartTime = System.nanoTime();
+ tablets = mergeRelationalTablets(tablets);
+ mergeTabletsCost = System.nanoTime() - mergeTabletsStartTime;
+ }
+ if (tablets.isEmpty()) {
+ throw new BatchExecutionException(SessionMessages.NO_TABLET_INSERTING);
+ }
+ final long insertTabletsStartTime = System.nanoTime();
+ if (enableRedirection) {
+ insertRelationalTabletsWithLeaderCache(tablets);
+ } else {
+ TSInsertTabletsReq request = genTSInsertTabletsReq(tablets, false,
false);
+ request.setWriteToTable(true);
+ for (Tablet tablet : tablets) {
+
request.addToColumnCategoriesList(toEnumOrdinalsAsBytes(tablet.getColumnTypes()));
+ }
+ try {
+ getDefaultSessionConnection().insertTablets(request);
+ } catch (RedirectException ignored) {
+ }
+ }
+ if (recordMergeTabletsCost) {
+ recordMergeTabletsCost(mergeTabletsCost, System.nanoTime() -
insertTabletsStartTime);
+ }
+ }
+
+ private void insertRelationalTabletsWithLeaderCache(final List<Tablet>
tablets)
+ throws IoTDBConnectionException, StatementExecutionException {
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup = new
HashMap<>();
+ for (final Tablet tablet : tablets) {
+ addRelationalTabletToGroup(tabletGroup, tablet);
+ }
+ if (tabletGroup.size() == 1) {
+ insertRelationalTabletsOnce(tabletGroup);
+ } else {
+ insertRelationalTabletsByGroup(tabletGroup);
+ }
+ }
+
+ private void addRelationalTabletToGroup(
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup, final
Tablet tablet)
+ throws IoTDBConnectionException {
+ if (SessionUtils.isTabletContainsSingleDevice(tablet)) {
+ final SessionConnection connection =
getTableModelSessionConnection(tablet.getDeviceID(0));
+ updateRelationalTabletsReq(tabletGroup, connection, tablet);
+ return;
+ }
+ final Map<SessionConnection, Map<IDeviceID, Tablet>>
subTabletsByConnection =
+ new LinkedHashMap<>();
+ for (int row = 0; row < tablet.getRowSize(); row++) {
+ final IDeviceID deviceID = tablet.getDeviceID(row);
+ final SessionConnection connection =
getTableModelSessionConnection(deviceID);
+ final Tablet subTablet =
+ subTabletsByConnection
+ .computeIfAbsent(connection, ignored -> new LinkedHashMap<>())
+ .computeIfAbsent(deviceID, ignored ->
createEmptyRelationalTabletLike(tablet));
+ for (int column = 0; column < subTablet.getSchemas().size(); column++) {
+ subTablet.addValue(
+ subTablet.getSchemas().get(column).getMeasurementName(),
+ subTablet.getRowSize(),
+ tablet.getValue(row, column));
+ }
+ subTablet.addTimestamp(subTablet.getRowSize(), tablet.getTimestamp(row));
+ }
+ subTabletsByConnection.forEach(
+ (connection, subTablets) ->
+ subTablets
+ .values()
+ .forEach(
+ subTablet -> updateRelationalTabletsReq(tabletGroup,
connection, subTablet)));
+ }
+
+ private SessionConnection getTableModelSessionConnection(final IDeviceID
deviceID)
+ throws IoTDBConnectionException {
+ return tableModelDeviceIdToEndpoint == null ||
tableModelDeviceIdToEndpoint.isEmpty()
+ ? getDefaultSessionConnection()
+ : getSessionConnection(deviceID);
+ }
+
+ private Tablet createEmptyRelationalTabletLike(final Tablet tablet) {
+ final List<String> measurements = new
ArrayList<>(tablet.getSchemas().size());
+ final List<TSDataType> dataTypes = new
ArrayList<>(tablet.getSchemas().size());
+ tablet
+ .getSchemas()
+ .forEach(
+ schema -> {
+ measurements.add(schema.getMeasurementName());
+ dataTypes.add(schema.getType());
+ });
+ return new Tablet(
+ tablet.getTableName(),
+ measurements,
+ dataTypes,
+ tablet.getColumnTypes(),
+ tablet.getRowSize());
+ }
+
+ private void updateRelationalTabletsReq(
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup,
+ final SessionConnection connection,
+ final Tablet tablet) {
+ tabletGroup
+ .computeIfAbsent(connection, ignored -> new RelationalTabletsReq())
+ .addTablet(tablet);
+ }
+
+ private void buildRelationalTabletsRequests(
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup) {
+ for (final RelationalTabletsReq relationalTabletsReq :
tabletGroup.values()) {
+ relationalTabletsReq.buildRequest();
+ }
+ }
+
+ private void insertRelationalTabletsOnce(
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup)
+ throws IoTDBConnectionException, StatementExecutionException {
+ buildRelationalTabletsRequests(tabletGroup);
+ final Map.Entry<SessionConnection, RelationalTabletsReq> entry =
+ tabletGroup.entrySet().iterator().next();
+ final SessionConnection connection = entry.getKey();
+ final RelationalTabletsReq relationalTabletsReq = entry.getValue();
+ try {
+ connection.insertTablets(relationalTabletsReq.request);
+ } catch (RedirectException e) {
+ handleRelationalTabletsRedirection(e, relationalTabletsReq.deviceIDs);
+ } catch (IoTDBConnectionException e) {
+ if (endPointToSessionConnection != null &&
endPointToSessionConnection.size() > 1) {
+ removeBrokenSessionConnection(connection);
+ try {
+
getDefaultSessionConnection().insertTablets(relationalTabletsReq.request);
+ } catch (RedirectException ignored) {
+ }
+ } else {
+ throw e;
+ }
+ }
+ }
+
+ @SuppressWarnings({
+ "squid:S3776"
+ }) // ignore Cognitive Complexity of methods should not be too high
+ private void insertRelationalTabletsByGroup(
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup)
+ throws IoTDBConnectionException, StatementExecutionException {
+ buildRelationalTabletsRequests(tabletGroup);
+ final List<CompletableFuture<Void>> completableFutures =
+ tabletGroup.entrySet().stream()
+ .map(
+ entry -> {
+ final SessionConnection connection = entry.getKey();
+ final RelationalTabletsReq relationalTabletsReq =
entry.getValue();
+ return CompletableFuture.runAsync(
+ () -> {
+ try {
+
connection.insertTablets(relationalTabletsReq.request);
+ } catch (RedirectException e) {
+ handleRelationalTabletsRedirection(e,
relationalTabletsReq.deviceIDs);
+ } catch (StatementExecutionException e) {
+ throw new CompletionException(e);
+ } catch (IoTDBConnectionException e) {
+ removeBrokenSessionConnection(connection);
+ try {
+ getDefaultSessionConnection()
+ .insertTablets(relationalTabletsReq.request);
+ } catch (IoTDBConnectionException |
StatementExecutionException ex) {
+ throw new CompletionException(ex);
+ } catch (RedirectException ignored) {
+ }
+ }
+ },
+ OPERATION_EXECUTOR);
+ })
+ .collect(Collectors.toList());
+
+ final StringBuilder errMsgBuilder = new StringBuilder();
+ for (final CompletableFuture<Void> completableFuture : completableFutures)
{
+ try {
+ completableFuture.join();
+ } catch (CompletionException completionException) {
+ final Throwable cause = completionException.getCause();
+ logger.error(SessionMessages.MEET_ERROR_WHEN_ASYNC_INSERT, cause);
+ if (cause instanceof IoTDBConnectionException) {
+ throw (IoTDBConnectionException) cause;
+ }
+ if (errMsgBuilder.length() > 0) {
+ errMsgBuilder.append(";");
+ }
+ errMsgBuilder.append(cause.getMessage());
+ }
+ }
+ if (errMsgBuilder.length() > 0) {
+ throw new StatementExecutionException(errMsgBuilder.toString());
+ }
+ }
+
+ private void handleRelationalTabletsRedirection(
+ final RedirectException redirectException, final List<IDeviceID>
deviceIDs) {
+ final List<TEndPoint> endPointList = redirectException.getEndPointList();
+ if (endPointList == null) {
+ return;
+ }
+ for (int i = 0; i < endPointList.size() && i < deviceIDs.size(); i++) {
+ if (endPointList.get(i) != null) {
+ handleRedirection(deviceIDs.get(i), endPointList.get(i));
+ }
+ }
+ }
+
+ private List<Tablet> mergeRelationalTablets(final List<Tablet> tablets) {
+ if (consecutiveRelationalTabletMergeMissCount
+ >= MAX_CONSECUTIVE_RELATIONAL_TABLET_MERGE_MISS_COUNT) {
+ return tablets;
+ }
+ final List<Tablet> sortedTablets = new ArrayList<>(tablets);
+ sortedTablets.sort(
+ Comparator.comparing(
+ Tablet::getTableName,
Comparator.nullsFirst(Comparator.naturalOrder())));
+ final List<RelationalTabletWithColumnIndexMap> mergedTablets = new
ArrayList<>(tablets.size());
+ boolean anyMerged = false;
+ for (final Tablet tablet : sortedTablets) {
+ final RelationalTabletWithColumnIndexMap relationalTablet =
+ new RelationalTabletWithColumnIndexMap(tablet,
getColumnIndexMap(tablet));
+ if (!mergedTablets.isEmpty()) {
+ final int lastIndex = mergedTablets.size() - 1;
+ final RelationalTabletWithColumnIndexMap candidate =
mergedTablets.get(lastIndex);
+ if (canMergeRelationalTablets(candidate, relationalTablet)) {
+ mergedTablets.set(lastIndex, mergeRelationalTablets(candidate,
relationalTablet.tablet));
Review Comment:
**Medium:** Every successful merge replaces the accumulated candidate with a
newly allocated tablet and recopies all previously accumulated rows at lines
3213-3224; constructing the wrapper then rescans those timestamps. Merging N
compatible tablets is therefore quadratic in copying and allocation, which
affects the exact workload this optimization targets. The adaptive cost check
runs only after the expensive call and after ten samples, so it cannot protect
the first large batch. Could we collect a compatible run and allocate and copy
once, or impose an explicit merge-work cap? Please add a scaling test with a
large mergeable list.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/SchemaValidator.java:
##########
@@ -85,6 +87,30 @@ public static void validate(
}
}
+ public static void validate(
+ final Metadata metadata,
+ final InsertTablets insertStatement,
+ final MPPQueryContext context,
+ final AccessControl accessControl) {
+ try {
+ for (InsertTabletStatement tabletStatement :
+
insertStatement.getInnerTreeStatement().getInsertTabletStatementList()) {
+ accessControl.checkCanInsertIntoTable(
+ context.getSession().getUserName(),
+ new QualifiedObjectName(
+ unQualifyDatabaseName(insertStatement.getDatabase()),
Review Comment:
**Medium:** Although insertStatement is the InsertTablets AST wrapper, its
inherited getDatabase method dynamically dispatches getInnerTreeStatement to
the wrapped InsertMultiTabletsStatement. That statement getDatabaseName method
scans every child, and AnalyzeUtils.getDatabaseName may invoke it twice.
Calling it inside this N-tablet permission loop makes validation O(N^2) before
schema work begins. Please resolve and unqualify the database once before the
loop; repeated checks for the same table can also be deduplicated.
##########
iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java:
##########
@@ -2857,6 +2867,491 @@ public void insertRelationalTablet(Tablet tablet)
}
}
+ /**
+ * insert relational Tablets.
+ *
+ * @param tablets data batches
+ */
+ public void insertRelationalTablets(List<Tablet> tablets)
+ throws IoTDBConnectionException, StatementExecutionException {
+ final boolean recordMergeTabletsCost = enableMergeTablets;
+ long mergeTabletsCost = 0;
+ if (enableMergeTablets) {
+ final long mergeTabletsStartTime = System.nanoTime();
+ tablets = mergeRelationalTablets(tablets);
+ mergeTabletsCost = System.nanoTime() - mergeTabletsStartTime;
+ }
+ if (tablets.isEmpty()) {
+ throw new BatchExecutionException(SessionMessages.NO_TABLET_INSERTING);
+ }
+ final long insertTabletsStartTime = System.nanoTime();
+ if (enableRedirection) {
+ insertRelationalTabletsWithLeaderCache(tablets);
+ } else {
+ TSInsertTabletsReq request = genTSInsertTabletsReq(tablets, false,
false);
+ request.setWriteToTable(true);
+ for (Tablet tablet : tablets) {
+
request.addToColumnCategoriesList(toEnumOrdinalsAsBytes(tablet.getColumnTypes()));
+ }
+ try {
+ getDefaultSessionConnection().insertTablets(request);
+ } catch (RedirectException ignored) {
+ }
+ }
+ if (recordMergeTabletsCost) {
+ recordMergeTabletsCost(mergeTabletsCost, System.nanoTime() -
insertTabletsStartTime);
+ }
+ }
+
+ private void insertRelationalTabletsWithLeaderCache(final List<Tablet>
tablets)
+ throws IoTDBConnectionException, StatementExecutionException {
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup = new
HashMap<>();
+ for (final Tablet tablet : tablets) {
+ addRelationalTabletToGroup(tabletGroup, tablet);
+ }
+ if (tabletGroup.size() == 1) {
+ insertRelationalTabletsOnce(tabletGroup);
+ } else {
+ insertRelationalTabletsByGroup(tabletGroup);
+ }
+ }
+
+ private void addRelationalTabletToGroup(
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup, final
Tablet tablet)
+ throws IoTDBConnectionException {
+ if (SessionUtils.isTabletContainsSingleDevice(tablet)) {
+ final SessionConnection connection =
getTableModelSessionConnection(tablet.getDeviceID(0));
+ updateRelationalTabletsReq(tabletGroup, connection, tablet);
+ return;
+ }
+ final Map<SessionConnection, Map<IDeviceID, Tablet>>
subTabletsByConnection =
+ new LinkedHashMap<>();
+ for (int row = 0; row < tablet.getRowSize(); row++) {
+ final IDeviceID deviceID = tablet.getDeviceID(row);
+ final SessionConnection connection =
getTableModelSessionConnection(deviceID);
+ final Tablet subTablet =
+ subTabletsByConnection
+ .computeIfAbsent(connection, ignored -> new LinkedHashMap<>())
+ .computeIfAbsent(deviceID, ignored ->
createEmptyRelationalTabletLike(tablet));
+ for (int column = 0; column < subTablet.getSchemas().size(); column++) {
+ subTablet.addValue(
+ subTablet.getSchemas().get(column).getMeasurementName(),
+ subTablet.getRowSize(),
+ tablet.getValue(row, column));
+ }
+ subTablet.addTimestamp(subTablet.getRowSize(), tablet.getTimestamp(row));
+ }
+ subTabletsByConnection.forEach(
+ (connection, subTablets) ->
+ subTablets
+ .values()
+ .forEach(
+ subTablet -> updateRelationalTabletsReq(tabletGroup,
connection, subTablet)));
+ }
+
+ private SessionConnection getTableModelSessionConnection(final IDeviceID
deviceID)
+ throws IoTDBConnectionException {
+ return tableModelDeviceIdToEndpoint == null ||
tableModelDeviceIdToEndpoint.isEmpty()
+ ? getDefaultSessionConnection()
+ : getSessionConnection(deviceID);
+ }
+
+ private Tablet createEmptyRelationalTabletLike(final Tablet tablet) {
+ final List<String> measurements = new
ArrayList<>(tablet.getSchemas().size());
+ final List<TSDataType> dataTypes = new
ArrayList<>(tablet.getSchemas().size());
+ tablet
+ .getSchemas()
+ .forEach(
+ schema -> {
+ measurements.add(schema.getMeasurementName());
+ dataTypes.add(schema.getType());
+ });
+ return new Tablet(
+ tablet.getTableName(),
+ measurements,
+ dataTypes,
+ tablet.getColumnTypes(),
+ tablet.getRowSize());
+ }
+
+ private void updateRelationalTabletsReq(
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup,
+ final SessionConnection connection,
+ final Tablet tablet) {
+ tabletGroup
+ .computeIfAbsent(connection, ignored -> new RelationalTabletsReq())
+ .addTablet(tablet);
+ }
+
+ private void buildRelationalTabletsRequests(
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup) {
+ for (final RelationalTabletsReq relationalTabletsReq :
tabletGroup.values()) {
+ relationalTabletsReq.buildRequest();
+ }
+ }
+
+ private void insertRelationalTabletsOnce(
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup)
+ throws IoTDBConnectionException, StatementExecutionException {
+ buildRelationalTabletsRequests(tabletGroup);
+ final Map.Entry<SessionConnection, RelationalTabletsReq> entry =
+ tabletGroup.entrySet().iterator().next();
+ final SessionConnection connection = entry.getKey();
+ final RelationalTabletsReq relationalTabletsReq = entry.getValue();
+ try {
+ connection.insertTablets(relationalTabletsReq.request);
+ } catch (RedirectException e) {
+ handleRelationalTabletsRedirection(e, relationalTabletsReq.deviceIDs);
+ } catch (IoTDBConnectionException e) {
+ if (endPointToSessionConnection != null &&
endPointToSessionConnection.size() > 1) {
+ removeBrokenSessionConnection(connection);
+ try {
+
getDefaultSessionConnection().insertTablets(relationalTabletsReq.request);
+ } catch (RedirectException ignored) {
+ }
+ } else {
+ throw e;
+ }
+ }
+ }
+
+ @SuppressWarnings({
+ "squid:S3776"
+ }) // ignore Cognitive Complexity of methods should not be too high
+ private void insertRelationalTabletsByGroup(
+ final Map<SessionConnection, RelationalTabletsReq> tabletGroup)
+ throws IoTDBConnectionException, StatementExecutionException {
+ buildRelationalTabletsRequests(tabletGroup);
+ final List<CompletableFuture<Void>> completableFutures =
+ tabletGroup.entrySet().stream()
+ .map(
+ entry -> {
+ final SessionConnection connection = entry.getKey();
+ final RelationalTabletsReq relationalTabletsReq =
entry.getValue();
+ return CompletableFuture.runAsync(
+ () -> {
+ try {
+
connection.insertTablets(relationalTabletsReq.request);
+ } catch (RedirectException e) {
+ handleRelationalTabletsRedirection(e,
relationalTabletsReq.deviceIDs);
+ } catch (StatementExecutionException e) {
+ throw new CompletionException(e);
+ } catch (IoTDBConnectionException e) {
+ removeBrokenSessionConnection(connection);
+ try {
+ getDefaultSessionConnection()
+ .insertTablets(relationalTabletsReq.request);
+ } catch (IoTDBConnectionException |
StatementExecutionException ex) {
+ throw new CompletionException(ex);
+ } catch (RedirectException ignored) {
+ }
+ }
+ },
+ OPERATION_EXECUTOR);
+ })
+ .collect(Collectors.toList());
+
+ final StringBuilder errMsgBuilder = new StringBuilder();
+ for (final CompletableFuture<Void> completableFuture : completableFutures)
{
+ try {
+ completableFuture.join();
+ } catch (CompletionException completionException) {
+ final Throwable cause = completionException.getCause();
+ logger.error(SessionMessages.MEET_ERROR_WHEN_ASYNC_INSERT, cause);
+ if (cause instanceof IoTDBConnectionException) {
+ throw (IoTDBConnectionException) cause;
+ }
+ if (errMsgBuilder.length() > 0) {
+ errMsgBuilder.append(";");
+ }
+ errMsgBuilder.append(cause.getMessage());
+ }
+ }
+ if (errMsgBuilder.length() > 0) {
+ throw new StatementExecutionException(errMsgBuilder.toString());
+ }
+ }
+
+ private void handleRelationalTabletsRedirection(
+ final RedirectException redirectException, final List<IDeviceID>
deviceIDs) {
+ final List<TEndPoint> endPointList = redirectException.getEndPointList();
Review Comment:
**Medium:** The batch redirection path cannot update the table leader cache
as currently wired. TableModelPlanner.setRedirectInfo emits redirect metadata
only for InsertTabletStatement, not the new InsertMultiTabletsStatement. In
addition, SessionConnection.insertTablets uses
verifySuccessWithRedirectionForMultiDevices, which creates a RedirectException
with deviceEndPointMap keyed from request prefix paths (table names), while
this handler reads only endPointList. Batch-only workloads therefore never
learn or repair per-device leader routes and keep going through the default or
stale DataNode. Please define redirect metadata aligned with the batch devices
and add a test that verifies cache updates; successful REDIRECTION_RECOMMEND
writes should not be retried.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
