PaimonTableSink.java
// 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.PaimonCppWriteSupport;
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.datasource.paimon.PaimonWriteTarget;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import org.apache.doris.nereids.trees.plans.commands.insert.InsertCommandContext;
import org.apache.doris.nereids.trees.plans.commands.insert.PaimonInsertCommandContext;
import org.apache.doris.qe.ConnectContext;
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.TPaimonStorageDescriptor;
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 the pinned table
* metadata, Hadoop authentication config, transaction identity, write mode,
* and sink column names.
*
* The upstream Exchange may establish concurrent HASH_FIXED writer ownership;
* partition and bucket handling inside each writer remains delegated to the SDK.
*/
public class PaimonTableSink extends BaseExternalTableDataSink {
public static final String ROW_KIND_COLUMN = "__DORIS_PAIMON_ROW_KIND__";
private final PaimonExternalTable targetTable;
private final PaimonWriteTarget writeTarget;
private final DMLCommandType dmlCommandType;
private final TPaimonWriteMode writeMode;
private List<Expr> outputExprs;
private List<Column> cols;
private TPaimonWriteBackendType backendType;
private TPaimonStorageDescriptor backendStorage;
private String backendSelectionReason;
private static final HashSet<TFileFormatType> supportedTypes = new HashSet<TFileFormatType>() {{
add(TFileFormatType.FORMAT_ORC);
add(TFileFormatType.FORMAT_PARQUET);
}};
public PaimonTableSink(PaimonWriteTarget writeTarget, DMLCommandType dmlCommandType,
TPaimonWriteMode writeMode) {
super();
this.writeTarget = writeTarget;
this.targetTable = writeTarget.getDorisTable();
this.dmlCommandType = dmlCommandType;
this.writeMode = writeMode;
}
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");
Preconditions.checkState(backendType != null,
"Paimon backend decision must be prepared during planning");
strBuilder.append(prefix).append(" backend: ").append(backendType)
.append(" (").append(backendSelectionReason).append(")\n");
if (explainLevel == TExplainLevel.BRIEF) {
return strBuilder.toString();
}
strBuilder.append(prefix).append(" table: ").append(targetTable.getName()).append("\n");
return strBuilder.toString();
}
/** Select the runtime backend during planning so EXPLAIN and execution share one decision. */
public void prepareBackendDecision() throws AnalysisException {
List<String> outputColumnNames = outputColumnNames();
backendType = TPaimonWriteBackendType.JNI;
backendStorage = null;
backendSelectionReason = "JNI selected";
String writeBackend = ConnectContext.get() == null
? "CPP" : ConnectContext.get().getSessionVariable().paimonWriteBackend;
if (!"CPP".equalsIgnoreCase(writeBackend)) {
return;
}
PaimonCppWriteSupport.Decision decision = PaimonCppWriteSupport.decide(
writeTarget.getTable(), outputColumnNames, writeMode,
targetTable.getCatalog().getCatalogProperty().getStoragePropertiesMap());
if (decision.isSupported()) {
backendType = TPaimonWriteBackendType.CPP;
backendStorage = decision.getStorage();
backendSelectionReason = "paimon-cpp supported";
} else {
backendSelectionReason = "JNI fallback: " + decision.getFallbackReason();
}
}
@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(writeTarget, ctx);
} catch (AnalysisException e) {
throw e;
} catch (UserException e) {
throw new AnalysisException("Failed to bind Paimon write transaction: "
+ e.getMessage(), e);
}
transaction.bind(binding);
tSink.setTransactionId(ctx.getTxnId());
tSink.setCommitUser(ctx.getCommitUser());
// Thrift column_names is the single column-order protocol shared by BE
// Arrow conversion and the Java writer schema.
List<String> outputColumnNames = outputColumnNames();
TPaimonWriteMode executionWriteMode = isChangelogWrite()
? TPaimonWriteMode.CHANGELOG
: ctx.isOverwrite() ? TPaimonWriteMode.OVERWRITE : TPaimonWriteMode.APPEND;
if (executionWriteMode != writeMode) {
throw new AnalysisException("Paimon write mode differs between planning and execution: planned="
+ writeMode + ", execution=" + executionWriteMode);
}
if (backendType == null) {
throw new AnalysisException("Paimon backend decision was not prepared during planning");
}
// Both SDKs construct their writer from the same FE-selected table definition.
tSink.setTableDescriptor(binding.getTableDescriptor());
tSink.setBackendType(backendType);
tSink.setWriteMode(writeMode);
tSink.setColumnNames(outputColumnNames);
if (backendType == TPaimonWriteBackendType.CPP) {
Preconditions.checkState(backendStorage != null,
"CPP Paimon backend requires a storage descriptor");
tSink.getTableDescriptor().setStorage(backendStorage);
}
tDataSink = new TDataSink(TDataSinkType.PAIMON_TABLE_SINK);
tDataSink.setPaimonTableSink(tSink);
}
private List<String> outputColumnNames() throws AnalysisException {
int columnOffset = isChangelogWrite() ? 1 : 0;
if (cols.size() + columnOffset != outputExprs.size()) {
throw new AnalysisException("Paimon sink output column size mismatch, columns="
+ cols.size() + ", exprs=" + outputExprs.size());
}
List<String> names = new ArrayList<>(outputExprs.size());
if (isChangelogWrite()) {
names.add(ROW_KIND_COLUMN);
}
for (Column col : cols) {
names.add(col.getName());
}
return names;
}
private boolean isChangelogWrite() {
return dmlCommandType == DMLCommandType.UPDATE
|| dmlCommandType == DMLCommandType.DELETE
|| dmlCommandType == DMLCommandType.MERGE;
}
}