github-actions[bot] commented on code in PR #65868: URL: https://github.com/apache/doris/pull/65868#discussion_r3671197884
########## be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp: ########## @@ -0,0 +1,430 @@ +// 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. + +#include "exec/sink/writer/paimon/jni_paimon_write_backend.h" + +#include <arrow/buffer.h> +#include <arrow/io/memory.h> +#include <arrow/ipc/writer.h> +#include <arrow/record_batch.h> +#include <arrow/type.h> + +#include <algorithm> +#include <map> +#include <string_view> +#include <vector> + +#include "common/check.h" +#include "common/logging.h" +#include "exec/sink/writer/paimon/paimon_jni_memory_manager.h" +#include "format/arrow/arrow_block_convertor.h" +#include "format/arrow/arrow_row_batch.h" +#include "runtime/exec_env.h" +#include "runtime/runtime_state.h" +#include "util/jni-util.h" +#include "util/pretty_printer.h" +#include "util/string_util.h" + +namespace doris { + +namespace { +constexpr std::string_view PAIMON_JNI_WRITER_IO_TMP_DIR = "paimon_jni_writer_io_tmp"; +} // namespace + +// ──────────────────────────────────────────────────────────── +// JNI helpers — JVM attachment and class loading +// ──────────────────────────────────────────────────────────── + +static constexpr const char* PAIMON_JNI_WRITER_CLASS = "org/apache/doris/paimon/PaimonJniWriter"; +static constexpr const char* SCANNER_LOADER_CLASS = + "org/apache/doris/common/classloader/ScannerLoader"; + +/// Attach the current native thread to the JVM if not already attached, +/// and return a valid JNIEnv pointer. +static Status _get_jni_env(JNIEnv** env) { + JavaVM* jvm = nullptr; + jsize n_vms = 0; + jint result = JNI_GetCreatedJavaVMs(&jvm, 1, &n_vms); + if (result != JNI_OK || n_vms == 0) { + return Status::InternalError("Failed to get created JavaVM"); + } + result = jvm->GetEnv(reinterpret_cast<void**>(env), JNI_VERSION_1_8); + if (result == JNI_EDETACHED) { + result = jvm->AttachCurrentThread(reinterpret_cast<void**>(env), nullptr); + if (result != JNI_OK) { + return Status::InternalError("Failed to attach current thread to JVM"); + } + } else if (result != JNI_OK) { + return Status::InternalError("Failed to get JNIEnv"); + } + return Status::OK(); +} + +JniPaimonWriteBackend::~JniPaimonWriteBackend() { + JNIEnv* env = nullptr; + bool java_users_stopped = _jni_writer_obj == nullptr; + Status env_status = _get_jni_env(&env); + if (env_status.ok()) { + if (_jni_writer_obj != nullptr) { + _refresh_memory_profile(); + DCHECK(_close_id != nullptr); + env->CallVoidMethod(_jni_writer_obj, _close_id); + Status close_status = _check_jni_exception(env, "close PaimonJniWriter"); Review Comment: [P1] Propagate Java close failures before reporting success `PaimonTableWriter::close()` has already added the prepared payloads to `RuntimeState` when destroying this backend. If Java `close()` throws, this destructor only logs it and deliberately releases the manager pointer, leaking all of that writer's Doris-allocated pages until process exit; the C++ close still returns success, so FE can commit and admit more writers. Repeated SDK/IO cleanup failures can therefore exhaust the BE without failing the responsible INSERT. Please make backend shutdown an explicit status-bearing step before publishing success, propagate the error so the prepared transaction is aborted, and keep any unavoidable quarantine reachable and subject to bounded admission instead of abandoning it. ########## fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java: ########## @@ -0,0 +1,152 @@ +// 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.doris.planner; + +import org.apache.doris.analysis.Expr; +import org.apache.doris.catalog.Column; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.UserException; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.datasource.paimon.PaimonTransaction; +import org.apache.doris.datasource.paimon.PaimonWriteBinding; +import org.apache.doris.nereids.trees.plans.commands.insert.InsertCommandContext; +import org.apache.doris.nereids.trees.plans.commands.insert.PaimonInsertCommandContext; +import org.apache.doris.thrift.TDataSink; +import org.apache.doris.thrift.TDataSinkType; +import org.apache.doris.thrift.TExplainLevel; +import org.apache.doris.thrift.TFileFormatType; +import org.apache.doris.thrift.TPaimonTableSink; +import org.apache.doris.thrift.TPaimonWriteBackendType; +import org.apache.doris.thrift.TPaimonWriteMode; + +import com.google.common.base.Preconditions; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Optional; +import java.util.Set; + +/** + * Paimon table sink. + * + * Generates TPaimonTableSink payload consumed by BE, including serialized table + * metadata, Hadoop authentication config, transaction identity, write mode, + * and sink column names. + * + * v1: single-writer architecture; partition/bucket routing delegated to SDK. + */ +public class PaimonTableSink extends BaseExternalTableDataSink { + private final PaimonExternalTable targetTable; + private List<Expr> outputExprs; + private List<Column> cols; + + private static final HashSet<TFileFormatType> supportedTypes = new HashSet<TFileFormatType>() {{ + add(TFileFormatType.FORMAT_ORC); + add(TFileFormatType.FORMAT_PARQUET); + }}; + + public PaimonTableSink(PaimonExternalTable targetTable) { + super(); + this.targetTable = targetTable; + } + + public void setCols(List<Column> cols) { + this.cols = cols; + } + + public void setOutputExprs(List<Expr> outputExprs) { + this.outputExprs = outputExprs; + } + + @Override + protected Set<TFileFormatType> supportedFileFormatTypes() { + return supportedTypes; + } + + @Override + public String getExplainString(String prefix, TExplainLevel explainLevel) { + StringBuilder strBuilder = new StringBuilder(); + strBuilder.append(prefix).append("PAIMON TABLE SINK\n"); + if (explainLevel == TExplainLevel.BRIEF) { + return strBuilder.toString(); + } + strBuilder.append(prefix).append(" table: ").append(targetTable.getName()).append("\n"); + return strBuilder.toString(); + } + + @Override + public void bindDataSink(Optional<InsertCommandContext> insertCtx) throws AnalysisException { + TPaimonTableSink tSink = new TPaimonTableSink(); + PaimonInsertCommandContext ctx = (PaimonInsertCommandContext) insertCtx.get(); + Preconditions.checkState(ctx.getTxnId() > 0, + "Paimon transaction must begin before sink binding"); + + PaimonTransaction transaction; + PaimonWriteBinding binding; + try { + transaction = (PaimonTransaction) targetTable.getCatalog() + .getTransactionManager().getTransaction(ctx.getTxnId()); + binding = PaimonWriteBinding.create(targetTable, ctx); Review Comment: [P1] Bind the same table generation used for writer distribution Physical planning has already chosen gather/hash/random execution from an earlier `getPaimonTableForWrite()` result, but this call reloads and serializes another latest table without validating the bucket/compaction options that determined that choice. For example, if a fixed-bucket append table changes from `write-only=true` to `false` after planning, the schema/ID retry check still passes, the fragment remains parallel, and the newly bound writers can compact the same bucket inputs. Please pin one `FileStoreTable` generation through physical planning and binding, or compare its generation and all write-safety options here and replan before opening writers. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonArrowConverter.java: ########## @@ -0,0 +1,373 @@ +// 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.doris.paimon; + +import org.apache.arrow.vector.BigIntVector; +import org.apache.arrow.vector.BitVector; +import org.apache.arrow.vector.DateDayVector; +import org.apache.arrow.vector.DecimalVector; +import org.apache.arrow.vector.FieldVector; +import org.apache.arrow.vector.Float4Vector; +import org.apache.arrow.vector.Float8Vector; +import org.apache.arrow.vector.IntVector; +import org.apache.arrow.vector.SmallIntVector; +import org.apache.arrow.vector.TimeStampVector; +import org.apache.arrow.vector.TinyIntVector; +import org.apache.arrow.vector.VarBinaryVector; +import org.apache.arrow.vector.VarCharVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.complex.ListVector; +import org.apache.arrow.vector.complex.MapVector; +import org.apache.arrow.vector.complex.StructVector; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.Decimal; +import org.apache.paimon.data.GenericArray; +import org.apache.paimon.data.GenericMap; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.Timestamp; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.BinaryType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.LocalZonedTimestampType; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.RowType; +import org.apache.paimon.types.TimestampType; +import org.apache.paimon.types.VarBinaryType; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.nio.ByteOrder; +import java.nio.charset.StandardCharsets; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Converts Arrow columns into Paimon internal values without owning writer state. */ +final class PaimonArrowConverter { + + private final ZoneId sessionTimeZone; + + PaimonArrowConverter(ZoneId sessionTimeZone) { + this.sessionTimeZone = sessionTimeZone; + } + + RowReader rows(VectorSchemaRoot root, DataType[] targetTypes) { + List<Field> fields = root.getSchema().getFields(); + List<FieldVector> vectors = root.getFieldVectors(); + if (fields.size() != targetTypes.length) { + throw new IllegalArgumentException("Arrow column count does not match Paimon write type"); + } + return new RowReader(fields, vectors, targetTypes); + } + + /** Bound view of one Arrow batch which converts only the requested row. */ + final class RowReader { + private final List<Field> fields; + private final List<FieldVector> vectors; + private final DataType[] targetTypes; + + private RowReader( + List<Field> fields, List<FieldVector> vectors, DataType[] targetTypes) { + this.fields = fields; + this.vectors = vectors; + this.targetTypes = targetTypes; + } + + Object[] values(int rowIndex) { + Object[] values = new Object[vectors.size()]; + for (int column = 0; column < vectors.size(); column++) { + values[column] = convertVectorValue( + vectors.get(column), rowIndex, fields.get(column), targetTypes[column]); + } + return values; + } + } + + private Object convertVectorValue( + FieldVector vector, int index, Field arrowField, DataType targetType) { + if (vector.isNull(index)) { + return null; + } + if (vector instanceof StructVector && targetType instanceof RowType) { + return convertStructVector((StructVector) vector, index, (RowType) targetType); + } + if (vector instanceof MapVector && targetType instanceof MapType) { + return convertMapVector((MapVector) vector, index, (MapType) targetType); + } + if (vector instanceof ListVector && targetType instanceof ArrayType) { + return convertArrayVector((ListVector) vector, index, (ArrayType) targetType); + } + if (vector instanceof IntVector) { + return ((IntVector) vector).get(index); + } + if (vector instanceof BigIntVector) { + return ((BigIntVector) vector).get(index); + } + if (vector instanceof SmallIntVector) { + return ((SmallIntVector) vector).get(index); + } + if (vector instanceof TinyIntVector) { + return ((TinyIntVector) vector).get(index); + } + if (vector instanceof Float4Vector) { + return ((Float4Vector) vector).get(index); + } + if (vector instanceof Float8Vector) { + return ((Float8Vector) vector).get(index); + } + if (vector instanceof BitVector) { + return ((BitVector) vector).get(index) == 1; + } + if (vector instanceof DateDayVector) { + return ((DateDayVector) vector).get(index); + } + if (vector instanceof VarCharVector) { Review Comment: [P1] Convert JSON text into a Paimon Variant Doris exports `TYPE_VARIANT` to Arrow as UTF-8 JSON, so this branch returns a `BinaryString` even when `targetType` is Paimon's `VariantType`. Paimon's `InternalRow` contract expects an `org.apache.paimon.data.variant.Variant` for that field (for example from `GenericVariant.fromJson`); when the writer later reads/serializes this `GenericRow`, the value has the wrong internal class and VARIANT inserts fail. Since the schema conversion in this PR explicitly maps Paimon VARIANT to Doris VARIANT and back, please add a `VariantType` case that parses the JSON into Paimon's variant representation and cover scalar, array, and object values. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java: ########## @@ -859,6 +873,89 @@ private void validateStaticPartition(UnboundIcebergTableSink<?> sink, IcebergExt } } + private Plan bindPaimonTableSink(MatchingContext<UnboundPaimonTableSink<Plan>> ctx) { + UnboundPaimonTableSink<?> sink = ctx.root; + Pair<PaimonExternalDatabase, PaimonExternalTable> pair = bind(ctx.cascadesContext, sink); + PaimonExternalDatabase database = pair.first; + PaimonExternalTable table = pair.second; + LogicalPlan child = ((LogicalPlan) sink.child()); + + Map<String, Expression> staticPartitions = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER); + staticPartitions.putAll(sink.getStaticPartitionKeyValues()); + Set<String> staticPartitionColNames = staticPartitions.keySet(); + if (!staticPartitionColNames.isEmpty()) { + Set<String> partitionColumnNames = Sets.newTreeSet(String.CASE_INSENSITIVE_ORDER); + partitionColumnNames.addAll(table.getWritePartitionColumnNames()); + for (String columnName : staticPartitionColNames) { + if (!partitionColumnNames.contains(columnName)) { + throw new AnalysisException(String.format( + "Column '%s' is not a partition column of Paimon table '%s'", + columnName, table.getName())); + } + Expression partitionValue = staticPartitions.get(columnName); + if (!(partitionValue instanceof Literal)) { + throw new AnalysisException(String.format( + "Partition value for column '%s' must be a literal, but got: %s", Review Comment: [P1] Resolve target columns from the latest write schema These target-column lookups still go through `getBaseSchema()`/`getColumn()`, whose `getFullSchema()` uses `MvccUtil.getSnapshotFromContext(this)`. A self-insert source pinned with `FOR VERSION AS OF` registers that historical snapshot under the same catalog/database/table identity, so after schema evolution an explicit latest column is rejected here (and the implicit list can still include a column that was later dropped) even though `PaimonWriteBinding` now serializes the latest table. Please resolve the target column list and names from the same latest write schema used by the binding, and cover a schema-evolved time-travel self-insert. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java: ########## @@ -0,0 +1,448 @@ +// 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.doris.datasource.paimon; + +import org.apache.doris.common.UserException; +import org.apache.doris.thrift.TPaimonCommitMessage; +import org.apache.doris.transaction.Transaction; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Lists; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.io.DataInputDeserializer; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageSerializer; +import org.apache.paimon.table.sink.InnerTableCommit; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Paimon transaction. + * + * Lifecycle: + * 1. bind() — pin the concrete table and writer configuration + * 2. updateCommitMessages() — called multiple times as BE reports commit data + * 3. commit() — deserialize all CommitMessages, call StreamTableCommit.filterAndCommit() + * 4. rollback() — abort uncommitted data files + * + * CommitMessage wire format: + * BE ← JNI ← Java: byte[] with DPCM header (magic + version + length) + Paimon + * CommitMessageSerializer payload + */ +public class PaimonTransaction implements Transaction { + private static final Logger LOG = LogManager.getLogger(PaimonTransaction.class); + + enum CommitState { + PREPARED, + COMMITTING, + COMMITTED, + OUTCOME_UNKNOWN + } + + private static final int COMMIT_HEADER_SIZE = 12; + private static final byte[] COMMIT_MAGIC = new byte[] {'D', 'P', 'C', 'M'}; + + private final PaimonMetadataOps ops; + private final long transactionId; + private final String commitUser; + private PaimonWriteBinding binding; + private CommitState state = CommitState.PREPARED; + + private final List<byte[]> commitPayloads = Lists.newArrayList(); + private final Set<CommitPayloadKey> commitPayloadSet = new HashSet<>(); + + public PaimonTransaction(PaimonMetadataOps ops, long transactionId) { + this.ops = Preconditions.checkNotNull(ops, "Paimon metadata ops must not be null"); + Preconditions.checkArgument(transactionId > 0, "Paimon transaction id must be positive"); + this.transactionId = transactionId; + this.commitUser = commitUser(transactionId); + } + + // ──────────────────────────────────────────────────────────── + // Transaction lifecycle + // ──────────────────────────────────────────────────────────── + + public synchronized void bind(PaimonWriteBinding binding) { + Preconditions.checkNotNull(binding, "Paimon write binding must not be null"); + Preconditions.checkState(this.binding == null, "Paimon transaction is already bound"); + Preconditions.checkState(state == CommitState.PREPARED, + "Paimon transaction can only be bound while prepared"); + this.binding = binding; + } + + @Override + public void commit() throws UserException { + PaimonWriteBinding writeBinding = requireBinding(); + List<byte[]> rawPayloads = snapshotPayloads(); + if (rawPayloads.isEmpty() && !writeBinding.isOverwrite()) { + LOG.info("Skip empty PaimonTransaction commit, txnId={}, table={}", + transactionId, tableName()); + markPreparedTransactionCommitted(); + return; + } + try { + List<CommitMessage> allMessages = deserializePayloads(rawPayloads); + LOG.info("Commit PaimonTransaction, txnId={}, table={}, payloads={}, messages={}, overwrite={}", + transactionId, tableName(), rawPayloads.size(), allMessages.size(), + writeBinding.isOverwrite()); + if (allMessages.isEmpty() && !writeBinding.isOverwrite()) { + throw new RuntimeException( + "Paimon commit messages are empty, payloads=" + rawPayloads.size()); + } + doCommitWithReconciliation(writeBinding, allMessages); + } catch (Exception e) { + throw new UserException("Failed to commit paimon transaction on FE", e); + } + } + + @Override + public void rollback() { + CommitState currentState = getState(); + if (currentState == CommitState.COMMITTED) { + LOG.info("Skip rollback for committed PaimonTransaction, txnId={}, table={}", + transactionId, tableName()); + return; + } + if (currentState == CommitState.COMMITTING + || currentState == CommitState.OUTCOME_UNKNOWN) { + LOG.warn("Skip rollback for PaimonTransaction in state {}, txnId={}, table={}. " + + "Preserving data files for snapshot safety", + currentState, transactionId, tableName()); + return; + } + List<byte[]> rawPayloads = snapshotPayloads(); + if (rawPayloads.isEmpty()) { + LOG.info("Skip empty PaimonTransaction rollback, txnId={}, table={}", + transactionId, tableName()); + return; + } + try { + PaimonWriteBinding writeBinding = requireBinding(); + List<CommitMessage> allMessages = deserializePayloads(rawPayloads); + if (allMessages.isEmpty()) { + LOG.info("Skip PaimonTransaction rollback with empty decoded messages, " + + "txnId={}, table={}", transactionId, tableName()); + return; + } + LOG.info("Rollback PaimonTransaction, txnId={}, table={}, payloads={}, messages={}", + transactionId, tableName(), rawPayloads.size(), allMessages.size()); + doAbort(writeBinding, allMessages); + } catch (Exception e) { + LOG.warn("Failed to rollback PaimonTransaction, txnId={}, table={}", + transactionId, tableName(), e); + } + } + + // ──────────────────────────────────────────────────────────── + // CommitMessage collection (called from Coordinator) + // ──────────────────────────────────────────────────────────── + + public void updateCommitMessages(List<TPaimonCommitMessage> messages) { + if (messages == null || messages.isEmpty()) { + return; + } + synchronized (this) { + for (TPaimonCommitMessage msg : messages) { + addPayload(msg); + } + } + } + + private void addPayload(TPaimonCommitMessage message) { + if (message == null || !message.isSetPayload()) { + return; + } + byte[] payload = message.getPayload(); + if (payload == null || payload.length == 0) { + return; + } + // Treat the Thrift payload as immutable after report handling and adopt it directly. The + // report is not reused, so copying it would only create a second full representation. + if (commitPayloadSet.add(new CommitPayloadKey(payload))) { + commitPayloads.add(payload); + } + } + + // ──────────────────────────────────────────────────────────── + // Transaction identity + // ──────────────────────────────────────────────────────────── + + public static String commitUser(long transactionId) { Review Comment: [P1] Namespace the commit identity by Doris cluster This commit user is only the cluster-local `Env.getNextId()` value, and the same number is also used as the `filterAndCommit` identifier. Independent Doris clusters writing the same Paimon table can reuse that pair; once cluster A commits it, Paimon treats cluster B's different payload as an already-committed retry, while this code marks B successful and loses its rows. Paimon's stream-write contract requires different applications to use different commit users. Please include a durable cluster namespace (for example the Doris cluster ID/UUID) while keeping the pair stable across retries, and cover two writers reusing the same local transaction number. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
