github-actions[bot] commented on code in PR #67395:
URL: https://github.com/apache/doris/pull/67395#discussion_r4055878062
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java:
##########
@@ -222,11 +222,11 @@ private static FileStoreTable
copyWithPinnedFallback(FileStoreTable table, Map<S
Map<String, String> fallbackOptions = new
HashMap<>(dynamicOptions);
// Keep branch policy, but never retranslate the main fence
against a later fallback history.
fallbackOptions.remove(CoreOptions.BUCKET.key());
- fallbackOptions.put(CoreOptions.BRANCH.key(),
pair.fallback().coreOptions().branch());
+ fallbackOptions.put(CoreOptions.BRANCH.key(),
pair.other().coreOptions().branch());
fallbackOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(),
snapshotId);
return new FallbackReadFileStoreTable(
copyWithPinnedFallback(pair.wrapped(), dynamicOptions,
coordinates, path),
- copyWithPinnedFallback(pair.fallback(),
fallbackOptions, coordinates, fallbackPath));
+ copyWithPinnedFallback(pair.other(), fallbackOptions,
coordinates, fallbackPath), true);
Review Comment:
[P1] Preserve primary-branch precedence when rebuilding the fallback pair —
for a table configured with `scan.primary-branch`,
`PaimonReaderOptions.isWrappedFirst(pair)` is false, but this constructor
hard-codes true; `restoreBoundSchema()` does the same at line 249. The
statement-pin path reaches these rebuilds, so it can read the current/wrapped
branch before the configured primary branch and return the wrong rows. Preserve
`PaimonReaderOptions.isWrappedFirst(pair)` at both sites and cover the
false-order case through option application/schema restoration.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java:
##########
@@ -0,0 +1,129 @@
+// 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.connector.paimon;
+
+import org.apache.doris.connector.spi.DorisConnectorException;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InstantiationUtil;
+
+import java.io.IOException;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.TreeMap;
+
+/** Statement-scoped Paimon write target shared by sink planning and
transaction commit. */
+final class PaimonWriteBinding {
+
+ private final String tableName;
+ private final FileStoreTable table;
+ private final String serializedTable;
+ private final Map<String, String> hadoopConfig;
+ private final boolean overwrite;
+ private final Map<String, String> staticPartition;
+
+ private PaimonWriteBinding(String tableName, FileStoreTable table,
+ Map<String, String> hadoopConfig, boolean overwrite,
+ Map<String, String> staticPartition) {
+ this.tableName = tableName;
+ this.table = table;
+ this.serializedTable = serialize(table);
Review Comment:
[P1] Strip the metastore loader from the table sent to the BE — HMS/DLF
`FileStoreTable`s carry a `HiveCatalogLoader`, and the existing scan handoff
deliberately calls `dropCatalogLoader()` because the BE plugin excludes the
Hive runtime classes; its POM explicitly notes that serialized tables reaching
this loader fail. This binding serializes the write table unchanged, and
`PaimonJniWriter.open()` deserializes it before `newWrite()`, so
metastore-backed writes can fail on the BE. Keep the catalog-bearing table for
the FE committer, but serialize a loader-free backend copy and add an HMS
writer-handoff test.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java:
##########
@@ -0,0 +1,129 @@
+// 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.connector.paimon;
+
+import org.apache.doris.connector.spi.DorisConnectorException;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InstantiationUtil;
+
+import java.io.IOException;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.TreeMap;
+
+/** Statement-scoped Paimon write target shared by sink planning and
transaction commit. */
+final class PaimonWriteBinding {
+
+ private final String tableName;
+ private final FileStoreTable table;
+ private final String serializedTable;
+ private final Map<String, String> hadoopConfig;
+ private final boolean overwrite;
+ private final Map<String, String> staticPartition;
+
+ private PaimonWriteBinding(String tableName, FileStoreTable table,
+ Map<String, String> hadoopConfig, boolean overwrite,
+ Map<String, String> staticPartition) {
+ this.tableName = tableName;
+ this.table = table;
+ this.serializedTable = serialize(table);
+ this.hadoopConfig = Collections.unmodifiableMap(new
LinkedHashMap<>(hadoopConfig));
+ this.overwrite = overwrite;
+ this.staticPartition = Collections.unmodifiableMap(new
LinkedHashMap<>(staticPartition));
+ }
+
+ static PaimonWriteBinding create(PaimonTableHandle handle, FileStoreTable
table,
+ Map<String, String> hadoopConfig, boolean overwrite,
+ Map<String, String> requestedStaticPartition) {
+ Map<String, String> staticPartition = resolveStaticPartition(table,
requestedStaticPartition);
+ FileStoreTable writeTable = configureTableForWrite(table, overwrite,
staticPartition);
+ return new PaimonWriteBinding(handle.getDatabaseName() + "." +
handle.getTableName(),
+ writeTable, hadoopConfig, overwrite, staticPartition);
+ }
+
+ static FileStoreTable configureTableForWrite(FileStoreTable table, boolean
overwrite,
+ Map<String, String> staticPartition) {
+ if (!overwrite) {
+ return table;
+ }
+ String dynamicOverwriteKey =
CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key();
+ boolean explicitlyDynamic = staticPartition.isEmpty()
+ &&
Boolean.parseBoolean(table.options().get(dynamicOverwriteKey));
+ if (explicitlyDynamic) {
+ return table;
+ }
+ return table.copy(Collections.singletonMap(dynamicOverwriteKey,
Boolean.FALSE.toString()));
+ }
+
+ private static Map<String, String> resolveStaticPartition(FileStoreTable
table,
+ Map<String, String> requested) {
+ Map<String, String> canonicalNames = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ for (String partitionKey : table.partitionKeys()) {
+ canonicalNames.put(partitionKey, partitionKey);
+ }
+ String defaultPartitionName =
CoreOptions.fromMap(table.options()).partitionDefaultName();
+ Map<String, String> result = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : requested.entrySet()) {
+ String canonicalName = canonicalNames.get(entry.getKey());
+ if (canonicalName == null) {
+ throw new DorisConnectorException("Column '" + entry.getKey()
+ + "' is not a partition column of Paimon table");
+ }
+ String value = entry.getValue();
+ result.put(canonicalName,
+ value == null || "NULL".equalsIgnoreCase(value) ?
defaultPartitionName : value);
Review Comment:
[P1] Preserve typed NULL when building the overwrite partition spec — both a
genuine SQL NULL and the literal strings `'null'`/`'NULL'` arrive here as
strings because the insert commands call `Literal.getStringValue()`, and this
comparison maps all of them to Paimon's null-partition sentinel. The data row
still contains the typed literal, so `INSERT OVERWRITE ... PARTITION (region =
'null')` writes new files into the string partition while `withOverwrite()`
removes the genuine-null partition and leaves the old string files in place.
Carry an explicit typed/null representation into this binding, and add
overwrite coverage for literal `'null'` and `'NULL'` alongside SQL NULL.
##########
be/src/exec/sink/writer/paimon/paimon_table_writer.cpp:
##########
@@ -0,0 +1,161 @@
+// 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/paimon_table_writer.h"
+
+#include "common/check.h"
+#include "common/logging.h"
+#include "core/block/block.h"
+#include "core/block/materialize_block.h"
+#include "exprs/vexpr_context.h"
+#include "runtime/runtime_state.h"
+
+namespace doris {
+
+PaimonTableWriter::PaimonTableWriter(TDataSink t_sink, const
VExprContextSPtrs& output_exprs)
+ : _t_sink(std::move(t_sink)), _output_expr_ctxs(output_exprs) {
+ DCHECK(_t_sink.__isset.paimon_table_sink);
+}
+
+Status PaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
+ _state = state;
+
+ // Register profile counters
+ _written_rows_counter = ADD_COUNTER(profile, "WrittenRows", TUnit::UNIT);
+ _written_bytes_counter = ADD_COUNTER(profile, "WrittenBytes",
TUnit::BYTES);
+ _send_data_timer = ADD_TIMER(profile, "SendDataTime");
+ _project_timer = ADD_CHILD_TIMER(profile, "ProjectTime", "SendDataTime");
+ _file_store_write_timer = ADD_CHILD_TIMER(profile, "FileStoreWriteTime",
"SendDataTime");
+ _open_timer = ADD_TIMER(profile, "OpenTime");
+ _close_timer = ADD_TIMER(profile, "CloseTime");
+ _prepare_commit_timer = ADD_TIMER(profile, "PrepareCommitTime");
+ _commit_payload_count = ADD_COUNTER(profile, "CommitPayloadCount",
TUnit::UNIT);
+ _commit_payload_bytes_counter = ADD_COUNTER(profile, "CommitPayloadBytes",
TUnit::BYTES);
+
+ SCOPED_TIMER(_open_timer);
+
+ // Step 1: Create the backend (JNI or FFI) based on the sink configuration.
+
RETURN_IF_ERROR(PaimonWriteBackendFactory::create(_t_sink.paimon_table_sink,
&_backend));
+ DCHECK(_backend);
+ // Step 2: Open the backend — for JNI this loads the Java class and calls
PaimonJniWriter.open().
+ RETURN_IF_ERROR(_backend->open(_t_sink.paimon_table_sink, state, profile));
+ // Step 3: Create a lightweight writer adapter that delegates to the
opened backend.
+ RETURN_IF_ERROR(_backend->create_writer(&_writer));
+ DCHECK(_writer);
+
+ LOG(INFO) << "PaimonTableWriter opened: backend=" <<
static_cast<int>(_backend->type())
+ << ", writer_scope=local_state";
+ return Status::OK();
+}
+
+Status PaimonTableWriter::write(RuntimeState* state, Block& block) {
+ if (block.rows() == 0) {
+ return Status::OK();
+ }
+
+ SCOPED_TIMER(_send_data_timer);
+
+ // Step 1: Apply output expressions to produce the columns selected by FE.
+ Block output_block;
+ {
+ SCOPED_TIMER(_project_timer);
+
RETURN_IF_ERROR(VExprContext::get_output_block_after_execute_exprs(_output_expr_ctxs,
block,
+
&output_block));
+ materialize_block_inplace(output_block);
+ }
+
+ COUNTER_UPDATE(_written_rows_counter, block.rows());
+ COUNTER_UPDATE(_written_bytes_counter, block.bytes());
+ state->update_num_rows_load_total(block.rows());
+ state->update_num_bytes_load_total(block.bytes());
+
+ // Step 2: Delegate to the backend writer (JNI or FFI). For the JNI path
+ // this converts Block → Arrow RecordBatch → Arrow C Data → Java
PaimonJniWriter.
+ DCHECK(_writer);
+ {
+ SCOPED_TIMER(_file_store_write_timer);
+ RETURN_IF_ERROR(_writer->write(state, output_block));
+ }
+ _written_rows += block.rows();
+ return Status::OK();
+}
+
+Status PaimonTableWriter::close(Status status) {
+ SCOPED_TIMER(_close_timer);
+
+ // Prepare messages first, but do not publish them until the backend
confirms
+ // that every SDK user has stopped and its native backing memory is safe
to release.
+ std::vector<TPaimonCommitMessage> messages;
+ if (status.ok()) {
+ DCHECK(_writer);
+ {
+ SCOPED_TIMER(_prepare_commit_timer);
+ Status prep_st = _writer->prepare_commit(messages);
+ if (!prep_st.ok()) {
+ status = prep_st;
+ }
+ }
+ }
+
+ // If prepare_commit failed or the incoming status was already an error,
+ // abort the writer to clean up uncommitted data files.
+ if (!status.ok()) {
+ LOG(WARNING) << "Paimon writer closing with error: " <<
status.to_string();
+ if (_writer) {
+ Status abort_st = _writer->abort();
+ if (!abort_st.ok()) {
+ LOG(WARNING) << "Paimon writer abort failed: " <<
abort_st.to_string();
+ }
+ }
+ }
+
+ // The adapter only owns Arrow conversion resources. Release it before
closing
+ // the backend, whose Java close is the authoritative SDK shutdown
boundary.
+ _writer.reset();
+
+ if (_backend) {
+ Status close_st = _backend->close();
+ if (!close_st.ok()) {
Review Comment:
[P2] Abort prepared files when backend close fails before publication —
`prepare_commit()` has already returned the staged-file messages, but a
subsequent backend-close error changes `status` and skips the only block that
publishes them to `RuntimeState`. The earlier abort block is not re-entered,
`_writer` is already destroyed, and the local messages then vanish, so FE
rollback sees no payload to abort. Keep prepared payloads under an
abort-capable owner through close; abort them on a confirmed completed close
failure and retain/reconcile them only for a genuinely ambiguous live-task
close.
##########
be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp:
##########
@@ -0,0 +1,543 @@
+// 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/c/bridge.h>
+#include <arrow/io/memory.h>
+#include <arrow/ipc/reader.h>
+#include <arrow/record_batch.h>
+#include <fmt/format.h>
+
+#include <algorithm>
+#include <atomic>
+#include <map>
+#include <mutex>
+#include <string_view>
+#include <vector>
+
+#include "common/check.h"
+#include "common/logging.h"
+#include "exec/sink/writer/paimon/paimon_jni_memory_manager.h"
+#include "exec/spill/spill_file_manager.h"
+#include "format/arrow/arrow_block_convertor.h"
+#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "runtime/runtime_state.h"
+#include "util/defer_op.h"
+#include "util/jni-util.h"
+#include "util/jni_plugin_registry.h"
+#include "util/pretty_printer.h"
+
+namespace doris {
+
+namespace {
+constexpr std::string_view PAIMON_JNI_WRITER_IO_TMP_DIR =
"paimon_jni_writer_io_tmp";
+
+void throw_java_io_exception(JNIEnv* env, const std::string& message) {
+ jclass exception_class = env->FindClass("java/io/IOException");
+ env->ThrowNew(exception_class, message.c_str());
+ env->DeleteLocalRef(exception_class);
+}
+
+jobjectArray get_paimon_spill_directories(JNIEnv* env, jclass, jlong
spill_session_handle) {
+ auto* spill_session =
reinterpret_cast<ExternalSpillSession*>(spill_session_handle);
+ if (spill_session == nullptr) {
+ throw_java_io_exception(env, "Paimon external spill session is null");
+ return nullptr;
+ }
+
+ std::vector<std::string> paths;
+ Status st = spill_session->get_paths(&paths);
+ if (!st.ok()) {
+ throw_java_io_exception(env, st.to_string());
+ return nullptr;
+ }
+ jclass string_class = env->FindClass("java/lang/String");
+ if (string_class == nullptr) {
+ return nullptr;
+ }
+ jobjectArray result =
+ env->NewObjectArray(static_cast<jsize>(paths.size()),
string_class, nullptr);
+ env->DeleteLocalRef(string_class);
+ if (result == nullptr) {
+ return nullptr;
+ }
+ for (jsize i = 0; i < static_cast<jsize>(paths.size()); ++i) {
+ jstring path = env->NewStringUTF(paths[i].c_str());
+ if (path == nullptr) {
+ return nullptr;
+ }
+ env->SetObjectArrayElement(result, i, path);
+ env->DeleteLocalRef(path);
+ if (env->ExceptionCheck()) {
+ return nullptr;
+ }
+ }
+ return result;
+}
+
+void reserve_paimon_spill(JNIEnv* env, jclass, jlong spill_session_handle,
jstring path,
+ jlong bytes) {
+ auto* spill_session =
reinterpret_cast<ExternalSpillSession*>(spill_session_handle);
+ if (spill_session == nullptr || path == nullptr) {
+ throw_java_io_exception(env, "Paimon external spill session or path is
null");
+ return;
+ }
+ const char* path_chars = env->GetStringUTFChars(path, nullptr);
+ if (path_chars == nullptr) {
+ return;
+ }
+ std::string native_path(path_chars);
+ env->ReleaseStringUTFChars(path, path_chars);
+ Status st = spill_session->reserve(native_path, bytes);
+ if (!st.ok()) {
+ throw_java_io_exception(env, st.to_string());
+ }
+}
+
+void update_paimon_spill_accounting(JNIEnv* env, jclass, jlong
spill_session_handle, jstring path,
+ jlong current_bytes_delta, jlong
write_bytes,
+ jlong read_bytes) {
+ auto* spill_session =
reinterpret_cast<ExternalSpillSession*>(spill_session_handle);
+ if (spill_session == nullptr || path == nullptr) {
+ return;
+ }
+ const char* path_chars = env->GetStringUTFChars(path, nullptr);
+ if (path_chars == nullptr) {
+ return;
+ }
+ std::string native_path(path_chars);
+ env->ReleaseStringUTFChars(path, path_chars);
+ spill_session->update_accounting(native_path, current_bytes_delta,
write_bytes, read_bytes);
+}
+
+Status register_paimon_spill_natives(JNIEnv* env, jclass writer_class) {
+ static char get_spill_directories_name[] = "getPaimonSpillDirectories";
+ static char get_spill_directories_signature[] = "(J)[Ljava/lang/String;";
+ static char reserve_spill_name[] = "reservePaimonSpill";
+ static char reserve_spill_signature[] = "(JLjava/lang/String;J)V";
+ static char update_spill_name[] = "updatePaimonSpillAccounting";
+ static char update_spill_signature[] = "(JLjava/lang/String;JJJ)V";
+ static ::JNINativeMethod methods[] = {
+ {get_spill_directories_name, get_spill_directories_signature,
+ reinterpret_cast<void*>(&get_paimon_spill_directories)},
+ {reserve_spill_name, reserve_spill_signature,
+ reinterpret_cast<void*>(&reserve_paimon_spill)},
+ {update_spill_name, update_spill_signature,
+ reinterpret_cast<void*>(&update_paimon_spill_accounting)},
+ };
+ if (env->RegisterNatives(writer_class, methods,
+ static_cast<jint>(sizeof(methods) /
sizeof(methods[0]))) != JNI_OK) {
+ RETURN_IF_ERROR(Jni::Env::GetJniExceptionMsg(
+ env, true, "JNI exception registering Paimon spill native
methods: "));
+ return Status::JniError("Failed to register Paimon spill native
methods");
+ }
+ return Status::OK();
+}
+
+std::atomic<bool>& paimon_jni_close_failed() {
+ static std::atomic<bool> failed {false};
+ return failed;
+}
+
+struct RetainedPaimonResources {
+ std::unique_ptr<PaimonJniMemoryManager> memory_manager;
+ std::unique_ptr<ExternalSpillSession> spill_session;
+};
+
+std::mutex& retained_resources_mutex() {
+ static auto* mutex = new std::mutex();
+ return *mutex;
+}
+
+std::vector<RetainedPaimonResources>& retained_resources() {
+ static auto* resources = new std::vector<RetainedPaimonResources>();
+ return *resources;
+}
+
+void
retain_resources_after_failed_close(std::unique_ptr<PaimonJniMemoryManager>
memory_manager,
+ std::unique_ptr<ExternalSpillSession>
spill_session) {
+ // An unconfirmed Java close means a background Paimon task may still
reference this manager's
+ // native pages or spill callbacks. Quarantine both resources and stop
admitting new writers so
+ // repeated failures cannot accumulate process-lifetime resources without
a bound.
+ paimon_jni_close_failed().store(true, std::memory_order_release);
Review Comment:
[P2] Do not permanently fence the BE after a completed SDK close error —
this one-way process-global flag is set for every Java close exception. In the
ordinary `writer.close()` failure path, `PaimonJniWriter` still terminates the
compaction executor, closes the index and IO manager, clears all writer state,
and only then rethrows; nevertheless every unrelated Paimon write on this BE is
rejected until restart. Distinguish a timed-out/live-callback close from a
completed SDK error, quarantine the affected handles, and bound/observe
retained resources without permanently disabling new writers.
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/GlobalIndexAssigner.java:
##########
@@ -0,0 +1,177 @@
+// 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.paimon.CoreOptions;
+import org.apache.paimon.crosspartition.BucketAssigner;
+import org.apache.paimon.crosspartition.ExistingProcessor;
+import org.apache.paimon.crosspartition.IndexBootstrap;
+import org.apache.paimon.crosspartition.KeyPartPartitionKeyExtractor;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.PartitionKeyExtractor;
+import org.apache.paimon.table.sink.RowPartitionAllPrimaryKeyExtractor;
+import org.apache.paimon.utils.IDMapping;
+import org.apache.paimon.utils.PositiveIntInt;
+import org.apache.paimon.utils.ProjectToRowFunction;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.BiConsumer;
+
+/**
+ * Assigns buckets for a key-dynamic table with a process-local global-key
index.
+ *
+ * <p>Doris gathers key-dynamic writes into one writer, so a single in-memory
index can preserve
+ * Paimon's cross-partition merge semantics without a native state backend.
+ */
+final class GlobalIndexAssigner implements AutoCloseable {
+ private final FileStoreTable table;
+ // TODO: After resolving rocksdbjni allocator compatibility with the Doris
BE jemalloc hook,
+ // use a RocksDB-backed on-disk index to bound Java heap usage for large
tables.
+ private final Map<BinaryRow, PositiveIntInt> keyIndex = new HashMap<>();
Review Comment:
[P1] Keep the key-dynamic index within the query memory budget —
`openKeyDynamicBucketAssigner()` bootstraps the whole table, so
`bootstrapKey()` copies every existing primary key into this map, and
`processNewRecord()` retains every new key as well. KEY_DYNAMIC is gathered to
one writer, while `PaimonJniMemoryManager` charges only native pages, so
`exec_mem_limit` cannot stop this Java-heap growth before the shared embedded
JVM is exhausted. The added 4M-key test explicitly labels this a known bug and
is skipped unless `enablePaimonKnownBugTest` is enabled. Please use a
bounded/spillable index or reject at an accounted query limit before enabling
this path.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2529,6 +2538,20 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
}
_append_external_file_commit_data(req, ¶ms);
+ if (auto pcm = req.runtime_state->paimon_commit_messages(); !pcm.empty()) {
Review Comment:
[P2] Retain rollback ownership until FE accepts these payloads — after
`prepareCommit()` the Java writer is closed and these `RuntimeState` messages
are the only abort metadata left. Each chunk is limited to 8 MiB, but their
aggregate is not, so `validate_report_exec_status_size()` can reject a valid
high-file-count final report locally before FE receives anything. Those
rejection paths only run the Iceberg external-file cleanup, and FE rollback
returns with an empty Paimon payload list, leaving the staged files orphaned.
Give Paimon an acknowledgement-backed cleanup owner and abort on definite
pre-delivery rejection while preserving files for ambiguous delivery.
##########
gensrc/thrift/DataSinks.thrift:
##########
@@ -46,6 +46,7 @@ enum TDataSinkType {
MAXCOMPUTE_TABLE_SINK = 18,
ICEBERG_DELETE_SINK = 19,
ICEBERG_MERGE_SINK = 20,
+ PAIMON_TABLE_SINK = 21,
Review Comment:
[P1] Gate the new sink type with a newly bumped BE capability — an
immediately preceding BE has no `_create_data_sink` case for value 21, but
`PaimonWritePlanProvider` emits this sink unconditionally and this PR leaves
the maximum execution version at the already-advertised value 15. During an
FE-first rolling upgrade, even gather or dynamic/bucket-unaware Paimon writes
that avoid the value-7 exchange protocol are dispatched to old BEs and fail as
an unsupported sink. Allocate a new capability and reject/gate planning until
all executing BEs support it, with a mixed-version gather-write test.
##########
gensrc/thrift/Partitions.thrift:
##########
@@ -45,11 +45,12 @@ enum TPartitionType {
// used for shuffle data by parititon and tablet
OLAP_TABLE_SINK_HASH_PARTITIONED = 6,
- // used for shuffle data by hive parititon
- HIVE_TABLE_SINK_HASH_PARTITIONED = 7,
+ // used for shuffle data by external table sink ownership key. BE execution
+ // versions before 12 reject this type because value 7 had different
semantics.
+ EXTERNAL_TABLE_SINK_HASH_PARTITIONED = 7,
Review Comment:
[P1] Allocate a new wire value and capability for this protocol — value 7
was already `HIVE_TABLE_SINK_HASH_PARTITIONED`, and the version-13 gate cannot
distinguish this feature because the immediately preceding FE/BE already
advertise version 15. During a BE-first upgrade, an old FE sends value 7
without field 5 and the new BE rejects every partitioned Hive write as missing
metadata. In the other direction, an old BE decodes a new Paimon fixed-bucket
request as the legacy Hive ScaleWriter protocol, ignores field 5, and loses the
required identity/one-writer-per-bucket routing. Please preserve value 7 for
legacy payloads and use a fresh enum plus a newly bumped capability version (or
gather until every BE supports it).
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/write/ConnectorWriteDistribution.java:
##########
@@ -0,0 +1,89 @@
+// 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.connector.spi.write;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** Connector-neutral description of the required distribution for a table
write. */
+public final class ConnectorWriteDistribution {
Review Comment:
[P2] Freeze the public members of the new connector SPI types — the API is
bumped to 10, but `ConnectorPluginSurfaceTest` only walks its hand-written
frozen sets, which omit this class, `ConnectorRowLevelDmlRequest`,
`ConnectorRowChangeStyle`, and `ConnectorWriteDistribution.Mode`. The refreshed
baseline therefore records their names only inside provider signatures, not the
factories/getters or enum constants that plugin bytecode links, so removing or
re-signing those members later can pass without a major bump. Add the new
classes/enums to the frozen sets and baseline, or discover public SPI types
automatically.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/ConnectorChangelogPlanBuilder.java:
##########
@@ -0,0 +1,536 @@
+// 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.nereids.rules.analysis;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.analyzer.Scope;
+import org.apache.doris.nereids.analyzer.UnboundAlias;
+import org.apache.doris.nereids.analyzer.UnboundSlot;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.LessThanEqual;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Not;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.WindowExpression;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AnyValue;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.AssertTrue;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.ShortCircuitIf;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.VarcharLiteral;
+import
org.apache.doris.nereids.trees.plans.commands.info.ConnectorChangelogRowChangeSpec;
+import org.apache.doris.nereids.trees.plans.commands.merge.MergeMatchedClause;
+import
org.apache.doris.nereids.trees.plans.commands.merge.MergeNotMatchedClause;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.types.BigIntType;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.TypeCoercionUtils;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+
+/** Builds the operation-column plus full-row projection used by
changelog-oriented connectors. */
+public final class ConnectorChangelogPlanBuilder {
+ public static final String OPERATION_COLUMN = "__DORIS_PAIMON_ROW_KIND__";
+ public static final byte INSERT = 0;
+ public static final byte UPDATE = 1;
+ public static final byte DELETE = 2;
+ private static final String BRANCH_LABEL = "__DORIS_CHANGELOG_BRANCH__";
+
+ private ConnectorChangelogPlanBuilder() {
+ }
+
+ /** Builds a changelog plan for the requested connector row-level
operation. */
+ public static LogicalPlan build(List<Column> schema, List<String>
primaryKeys,
+ ConnectorChangelogRowChangeSpec spec, LogicalPlan child,
CascadesContext context) {
+ if (spec instanceof ConnectorChangelogRowChangeSpec.Update) {
+ return buildUpdate(schema,
(ConnectorChangelogRowChangeSpec.Update) spec, child, context);
+ }
+ if (spec instanceof ConnectorChangelogRowChangeSpec.Delete) {
+ return buildDelete(schema, primaryKeys,
(ConnectorChangelogRowChangeSpec.Delete) spec,
+ child, context);
+ }
+ if (spec instanceof ConnectorChangelogRowChangeSpec.Merge) {
+ return new MergeBuilder(schema, primaryKeys,
+ (ConnectorChangelogRowChangeSpec.Merge) spec, child,
context).build();
+ }
+ throw new AnalysisException("Unsupported connector changelog
specification: "
+ + spec.getClass().getSimpleName());
+ }
+
+ private static LogicalPlan buildUpdate(List<Column> schema,
+ ConnectorChangelogRowChangeSpec.Update update, LogicalPlan child,
+ CascadesContext context) {
+ Map<String, Expression> changes =
Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
+ for (EqualTo assignment : update.getAssignments()) {
+ List<String> parts = ((UnboundSlot)
assignment.left()).getNameParts();
+ String name = parts.get(parts.size() - 1);
+ if (changes.put(name, assignment.right()) != null) {
+ throw new AnalysisException("Duplicate column name in
connector UPDATE: " + name);
+ }
+ }
+ ExpressionAnalyzer analyzer = analyzer(child, context);
+ List<NamedExpression> projects = new ArrayList<>();
+ projects.add(operation(UPDATE));
+ for (Column column : schema) {
+ Expression value = changes.remove(column.getName());
+ if (value == null) {
+ value = targetSlot(update.getTargetNameInPlan(),
column.getName());
+ }
+ projects.add(bindColumn(analyzer, value, column));
+ }
+ if (!changes.isEmpty()) {
+ throw new AnalysisException("Unknown column in connector UPDATE: "
+ + String.join(", ", changes.keySet()));
+ }
+ return new LogicalProject<>(projects, child);
+ }
+
+ private static LogicalPlan buildDelete(List<Column> schema, List<String>
primaryKeys,
+ ConnectorChangelogRowChangeSpec.Delete delete, LogicalPlan child,
+ CascadesContext context) {
+ ExpressionAnalyzer analyzer = analyzer(child, context);
+ List<NamedExpression> projects = new ArrayList<>();
+ projects.add(operation(DELETE));
+ for (Column column : schema) {
+ projects.add(bindColumn(analyzer,
+ targetSlot(delete.getTargetNameInPlan(),
column.getName()), column));
+ }
+ LogicalProject<LogicalPlan> project = new LogicalProject<>(projects,
child);
+ if (!delete.shouldDeduplicateTargetRows()) {
+ return project;
+ }
+ Set<String> keys = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ keys.addAll(primaryKeys);
+ List<Expression> groupBy = new ArrayList<>();
+ List<NamedExpression> outputs = new ArrayList<>();
+ Slot operation = project.getOutput().get(0);
+ groupBy.add(operation);
+ outputs.add(operation);
+ for (int i = 0; i < schema.size(); i++) {
+ Column column = schema.get(i);
+ Slot value = project.getOutput().get(i + 1);
+ if (keys.contains(column.getName())) {
+ groupBy.add(value);
+ outputs.add(value);
+ } else {
+ outputs.add(new Alias(new AnyValue(value), column.getName()));
+ }
+ }
+ return new LogicalAggregate<>(groupBy, outputs, project);
+ }
+
+ private static Alias operation(byte value) {
+ return new Alias(new TinyIntLiteral(value), OPERATION_COLUMN);
+ }
+
+ private static UnboundSlot targetSlot(List<String> qualifier, String
column) {
+ List<String> parts = new ArrayList<>(qualifier);
+ parts.add(column);
+ return new UnboundSlot(parts);
+ }
+
+ private static ExpressionAnalyzer analyzer(LogicalPlan plan,
CascadesContext context) {
+ return new ExpressionAnalyzer(plan, new Scope(plan.getOutput()),
context, true, false);
+ }
+
+ private static Alias bindColumn(ExpressionAnalyzer analyzer, Expression
expression, Column column) {
+ Expression value = analyzer.analyze(expression);
+ value = TypeCoercionUtils.castIfNotSameType(value,
DataType.fromCatalogType(column.getType()));
+ return new Alias(value, column.getName());
+ }
+
+ private static final class MergeBuilder {
+ private final List<Column> schema;
+ private final List<String> primaryKeys;
+ private final ConnectorChangelogRowChangeSpec.Merge merge;
+ private final LogicalPlan child;
+ private final ExpressionAnalyzer analyzer;
+
+ private MergeBuilder(List<Column> schema, List<String> primaryKeys,
+ ConnectorChangelogRowChangeSpec.Merge merge, LogicalPlan child,
+ CascadesContext context) {
+ this.schema = schema;
+ this.primaryKeys = primaryKeys;
+ this.merge = merge;
+ this.child = child;
+ this.analyzer = analyzer(child, context);
+ }
+
+ private LogicalPlan build() {
+ if (primaryKeys.isEmpty()) {
+ throw new AnalysisException("Connector MERGE requires a
primary-key table");
+ }
+ Alias branch = bindBranchLabel();
+ Slot branchSlot = branch.toSlot();
+ List<NamedExpression> branchOutputs = new
ArrayList<>(child.getOutput());
+ branchOutputs.add(branch);
+ LogicalPlan selected = new LogicalProject<>(branchOutputs, child);
+ selected = new LogicalFilter<>(
+ ImmutableSet.of(new Not(new
org.apache.doris.nereids.trees.expressions.IsNull(branchSlot))),
+ selected);
+ List<List<Expression>> branches = buildBranchProjections();
+ if (!merge.getNotMatchedClauses().isEmpty()) {
+ validateNotMatchedPrimaryKeys(branches);
+ }
+ List<NamedExpression> output = new ArrayList<>();
+ for (int column = 0; column <= schema.size(); column++) {
+ DataType type = column == 0
+ ? org.apache.doris.nereids.types.TinyIntType.INSTANCE
+ : DataType.fromCatalogType(schema.get(column -
1).getType());
+ String name = column == 0 ? OPERATION_COLUMN :
schema.get(column - 1).getName();
+ Expression value = new NullLiteral(type);
+ for (int index = branches.size() - 1; index >= 0; index--) {
+ Expression branchValue =
TypeCoercionUtils.castIfNotSameType(
+ branches.get(index).get(column), type);
+ value = new ShortCircuitIf(new EqualTo(branchSlot, new
IntegerLiteral(index)),
Review Comment:
[P1] Keep forced-short-circuit IFs out of generic boolean rewrites — this
also wraps boolean target columns, but `CaseWhenToCompoundPredicate` matches
`If.class` with `Class.isInstance`, so it matches `ShortCircuitIf` too. For a
literal-valued branch it rewrites the expression to ordinary `AND`/`OR` and
drops `AlwaysShortCircuit`; with the default `short_circuit_evaluation=false`,
an inactive later expression such as `assert_true(false, ...)` can still
execute and fail the MERGE. Exclude `AlwaysShortCircuit` expressions from that
rewrite (or preserve the marker) and add a full-plan boolean-branch test.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java:
##########
@@ -1153,6 +1190,223 @@ public void dropTable(ConnectorSession session,
ConnectorTableHandle handle) {
LOG.info("dropped Paimon table {}", id);
}
+ // ==================== DDL: Column evolution ====================
+
+ @Override
+ public void addColumn(ConnectorSession session, ConnectorTableHandle
handle,
+ ConnectorColumn column, ConnectorColumnPosition position) {
+ PaimonTableHandle paimonHandle = (PaimonTableHandle) handle;
+ List<DataField> fields = loadRemoteFields(paimonHandle);
+ Map<String, DataField> fieldsByName = indexFieldsByName(fields);
+ rejectDuplicateColumn(fieldsByName.keySet(), column.getName());
Review Comment:
[P1] Reject reserved metadata names before applying column evolution —
CREATE already blocks `__paimon_file_path` and `__paimon_row_index`, but the
new ADD/ADD COLUMNS/RENAME paths pass those names to Paimon. After the remote
ALTER succeeds, `afterExternalDdl()` refreshes the table and both
`buildTableSchema()` and `buildColumnHandles()` reject the physical collision,
so the altered table can no longer be loaded for normal reads. Reuse the
case-insensitive reserved-name check for every newly introduced column name and
cover both names and case variants in column-evolution tests.
--
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]
