Copilot commented on code in PR #2540:
URL: https://github.com/apache/phoenix/pull/2540#discussion_r3500644902
##########
phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java:
##########
@@ -551,6 +556,31 @@ public void append(String tableName, long commitId,
Mutation mutation) throws IO
if (LOG.isTraceEnabled()) {
LOG.trace("Append: table={}, commitId={}, mutation={}", tableName,
commitId, mutation);
}
+ List<Cell> cells = new ArrayList<>();
+ for (List<Cell> familyCells : mutation.getFamilyCellMap().values()) {
+ cells.addAll(familyCells);
+ }
+ publishDataEvent(new Record(tableName, commitId, cells));
+ }
Review Comment:
append(String,long,Mutation) now flattens the mutation into a cell list and
publishes it asynchronously, but it doesn't guard against empty mutations. If a
Put/Delete has no cells, the consumer will later fail when encoding
(LogFileCodec rejects empty records), potentially triggering fail-stop.
Validate and fail fast on empty input before publishing.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java:
##########
@@ -551,6 +556,31 @@ public void append(String tableName, long commitId,
Mutation mutation) throws IO
if (LOG.isTraceEnabled()) {
LOG.trace("Append: table={}, commitId={}, mutation={}", tableName,
commitId, mutation);
}
+ List<Cell> cells = new ArrayList<>();
+ for (List<Cell> familyCells : mutation.getFamilyCellMap().values()) {
+ cells.addAll(familyCells);
+ }
+ publishDataEvent(new Record(tableName, commitId, cells));
+ }
+
+ /**
+ * Append a coalesced batch of cells as a single record on the log. The
cells must already be
+ * grouped by row+type so consumers can reconstruct Put/Delete mutations on
the row+type boundary
+ * (see {@link MutationCellGrouper}). Behaves identically to
+ * {@link #append(String, long, Mutation)} with respect to backpressure and
fail-stop.
+ * @param tableName The HBase table name shared by every cell in {@code
cells}.
+ * @param commitId The commit identifier (e.g., SCN) for the batch.
+ * @param cells The flat ordered cell stream for one batch on one table.
+ * @throws IOException If the writer is closed or the ring buffer is full.
+ */
+ public void append(String tableName, long commitId, List<Cell> cells) throws
IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Append: table={}, commitId={}, cells={}", tableName,
commitId, cells.size());
+ }
+ publishDataEvent(new Record(tableName, commitId, cells));
+ }
Review Comment:
append(String,long,List<Cell>) will throw a NullPointerException when trace
logging is enabled if callers pass a null cell list (cells.size()). Since the
record body is also required to be non-empty (LogFileCodec rejects empty
records), validate inputs before publishing so the failure is synchronous and
doesn't poison the consumer thread later.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/replication/MutationCellGrouper.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.phoenix.replication;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Put;
+
+/**
+ * Groups a flat cell stream into Put/Delete mutations, mirroring the
algorithm HBase's
+ * ReplicationSink uses to reconstruct mutations from a WALEdit. A new
mutation is started whenever
+ * the row key or the put-vs-delete disposition differs from the immediately
preceding cell;
+ * consecutive cells sharing both are collected into one mutation. There is no
precondition on the
+ * input ordering: any cell stream produces valid mutations. Ordering only
affects how the cells are
Review Comment:
The class-level Javadoc says the boundary is "put-vs-delete disposition",
but the implementation actually starts a new mutation whenever the *cell type*
changes (previousCell.getType() != cell.getType()), which intentionally splits
distinct delete subtypes. Please align the documentation with the actual
row+cell-type boundary so readers don't misinterpret the contract.
--
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]
