From 52a3e59bfddb4fbdfd8a472a2c59ce1b64bf67dd Mon Sep 17 00:00:00 2001 From: moke-HU <25567926679@qq.com> Date: Fri, 14 Aug 2026 11:24:13 +0800 Subject: [PATCH 1/7] feat(paimon): upgrade connector runtime to 2.0.0 --- .../apache/doris/paimon/PaimonJniScanner.java | 26 ++++++++-- .../doris/paimon/PaimonJniScannerTest.java | 37 +++++++------- .../fe-connector-paimon-hive-shade/pom.xml | 18 +++++++ .../org.apache.paimon.factories.Factory | 2 + fe/fe-connector/fe-connector-paimon/pom.xml | 16 ++++++ .../connector/paimon/PaimonReaderOptions.java | 28 ++++++++--- .../paimon/PaimonScanPlanProvider.java | 14 +++--- .../connector/paimon/FakePaimonTable.java | 37 ++++++++++++++ .../paimon/PaimonBackendBoundTableTest.java | 50 +++++++++---------- .../PaimonConnectorMetadataPartitionTest.java | 8 +-- ...nnectorMetadataPartitionViewCacheTest.java | 2 +- .../PaimonConnectorMetadataReadAuthTest.java | 3 +- .../paimon/PaimonReaderOptionsTest.java | 40 ++++++++++----- .../paimon/PaimonScanMetricsTest.java | 2 +- .../paimon/PaimonScanPlanProviderTest.java | 14 ++++-- fe/pom.xml | 2 +- 16 files changed, 214 insertions(+), 85 deletions(-) create mode 100644 fe/fe-connector/fe-connector-paimon-hive-shade/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory diff --git a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java index 1dbc045d2e47a3..533be38efcbb4c 100644 --- a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java +++ b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java @@ -41,6 +41,8 @@ import org.apache.paimon.table.system.SystemTableLoader; import org.apache.paimon.types.DataType; import org.apache.paimon.types.TimestampType; +import org.apache.paimon.utils.ChainTableUtils; +import org.apache.paimon.utils.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -703,14 +705,14 @@ private static Table applyManifestParallelismBound( FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) table; FileStoreTable main = applyManifestParallelismBound( pair.wrapped(), safeBound, materializeAbsent); - FileStoreTable fallback = applyManifestParallelismBound( - pair.fallback(), safeBound, materializeAbsent); - if (main == pair.wrapped() && fallback == pair.fallback()) { + FileStoreTable other = applyManifestParallelismBound( + pair.other(), safeBound, materializeAbsent); + if (main == pair.wrapped() && other == pair.other()) { return table; } // Each branch owns an independent planner setting; a smaller sibling is not an // execution ceiling and must never throttle the other branch. - return new FallbackReadFileStoreTable(main, fallback); + return new FallbackReadFileStoreTable(main, other, isWrappedFirst(pair)); } if (table instanceof DelegatedFileStoreTable) { @@ -773,6 +775,20 @@ private static FileStoreTable applyManifestParallelismBound( (Table) table, safeBound, materializeAbsent); } + static boolean isWrappedFirst(FallbackReadFileStoreTable table) { + Map options = table.options(); + // Match FileStoreTableFactory's construction order. Paimon does not expose wrappedFirst. + if (ChainTableUtils.isChainTable(options)) { + return true; + } + if (!StringUtils.isNullOrWhitespaceOnly( + options.get(CoreOptions.SCAN_FALLBACK_BRANCH.key()))) { + return true; + } + return StringUtils.isNullOrWhitespaceOnly( + options.get(CoreOptions.SCAN_PRIMARY_BRANCH.key())); + } + private static FileStoreTable unwrapSystemPlanningSource(FileStoreTable table) { FileStoreTable current = table; // System wrappers dispatch fallback reads only when the fallback pair is their direct @@ -801,7 +817,7 @@ private static void validateSerializedReaderOptions(Table table) { validateSerializedAsyncThreshold(table.options().get(CoreOptions.FILE_READER_ASYNC_THRESHOLD.key())); validateSerializedSplitTargetSize(table.options().get(CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key())); if (table instanceof FallbackReadFileStoreTable) { - validateSerializedReaderOptions(((FallbackReadFileStoreTable) table).fallback()); + validateSerializedReaderOptions(((FallbackReadFileStoreTable) table).other()); } if (table instanceof DelegatedFileStoreTable) { validateSerializedReaderOptions(((DelegatedFileStoreTable) table).wrapped()); diff --git a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java index 8fe6523ed74a2d..c72c863d8f3750 100644 --- a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java +++ b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java @@ -159,7 +159,7 @@ public void testOldFeSerializedFallbackZeroReadBatchIsRejected() throws Exceptio Collections.singletonMap(CoreOptions.READ_BATCH_SIZE.key(), "0")); Map params = createBaseParams(); params.put("serialized_table", Base64.getUrlEncoder().withoutPadding().encodeToString( - InstantiationUtil.serializeObject(new FallbackReadFileStoreTable(main, fallback)))); + InstantiationUtil.serializeObject(new FallbackReadFileStoreTable(main, fallback, true)))); PaimonJniScanner scanner = new PaimonJniScanner(1024, params); Method initTable = PaimonJniScanner.class.getDeclaredMethod("initTable"); initTable.setAccessible(true); @@ -181,7 +181,8 @@ public void testOldFeSerializedAsyncThresholdIsRejectedInEveryChild() throws Exc FileStoreTable main = serializableFileStoreTable(Collections.emptyMap()); FileStoreTable fallback = serializableFileStoreTable(Collections.singletonMap( CoreOptions.FILE_READER_ASYNC_THRESHOLD.key(), "2 GB")); - for (Table configuredTable : Arrays.asList(visible, new FallbackReadFileStoreTable(main, fallback))) { + for (Table configuredTable : Arrays.asList( + visible, new FallbackReadFileStoreTable(main, fallback, true))) { Map params = createBaseParams(); params.put("serialized_table", Base64.getUrlEncoder().withoutPadding().encodeToString( InstantiationUtil.serializeObject(configuredTable))); @@ -203,7 +204,7 @@ public void testOldFeSerializedSystemSourceRejectsZeroSplitTarget() throws Excep FileStoreTable fallback = serializableFileStoreTable(Collections.singletonMap( CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "0 B")); Table filesTable = SystemTableLoader.load( - "files", new FallbackReadFileStoreTable(main, fallback)); + "files", new FallbackReadFileStoreTable(main, fallback, true)); Map params = createBaseParams(); params.put("serialized_table", Base64.getUrlEncoder().withoutPadding().encodeToString( InstantiationUtil.serializeObject(filesTable))); @@ -224,7 +225,7 @@ public void testOldFeSerializedSystemSourceRejectsFallbackZeroReadBatch() throws FileStoreTable fallback = serializableFileStoreTable(Collections.singletonMap( CoreOptions.READ_BATCH_SIZE.key(), "0")); Table readerBackedSystemTable = SystemTableLoader.load( - "audit_log", new FallbackReadFileStoreTable(main, fallback)); + "audit_log", new FallbackReadFileStoreTable(main, fallback, true)); Map params = createBaseParams(); params.put("serialized_table", Base64.getUrlEncoder().withoutPadding().encodeToString( InstantiationUtil.serializeObject(readerBackedSystemTable))); @@ -246,10 +247,10 @@ public void testBackendManifestCapReachesHiddenFallbackPlanner() { Collections.singletonMap(CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), "8")); Table safe = PaimonJniScanner.applyBackendManifestParallelism( - new FallbackReadFileStoreTable(main, fallback), "8", 4); + new FallbackReadFileStoreTable(main, fallback, true), "8", 4); Assert.assertTrue(safe instanceof FallbackReadFileStoreTable); - Assert.assertEquals("4", ((FallbackReadFileStoreTable) safe).fallback() + Assert.assertEquals("4", ((FallbackReadFileStoreTable) safe).other() .options().get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); } @@ -260,9 +261,9 @@ public void testAdvertisedFeCapStillChecksSerializedChildren() { Collections.singletonMap(CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), "200")); Table safe = PaimonJniScanner.applyBackendManifestParallelism( - new FallbackReadFileStoreTable(main, fallback), "32", 64); + new FallbackReadFileStoreTable(main, fallback, true), "32", 64); - Assert.assertEquals("32", ((FallbackReadFileStoreTable) safe).fallback() + Assert.assertEquals("32", ((FallbackReadFileStoreTable) safe).other() .options().get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); } @@ -277,11 +278,11 @@ public void testOldFeManifestBackstopKeepsStableMaximum() { Table safeVisible = PaimonJniScanner.applyBackendManifestParallelism( visible, null, 512); Table safeFallback = PaimonJniScanner.applyBackendManifestParallelism( - new FallbackReadFileStoreTable(main, fallback), null, 512); + new FallbackReadFileStoreTable(main, fallback, true), null, 512); Assert.assertEquals("256", safeVisible.options() .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); - Assert.assertEquals("256", ((FallbackReadFileStoreTable) safeFallback).fallback() + Assert.assertEquals("256", ((FallbackReadFileStoreTable) safeFallback).other() .options().get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); } @@ -301,20 +302,20 @@ public void testFallbackManifestParallelismIsCappedPerBranch() { CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), "1")); FileStoreTable fallback = serializableFileStoreTable(Collections.singletonMap( CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), "128")); - Table pair = new FallbackReadFileStoreTable(main, fallback); + Table pair = new FallbackReadFileStoreTable(main, fallback, true); FallbackReadFileStoreTable unchanged = (FallbackReadFileStoreTable) PaimonJniScanner.applyBackendManifestParallelism(pair, "128", 128); Assert.assertEquals("1", unchanged.wrapped().options() .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); - Assert.assertEquals("128", unchanged.fallback().options() + Assert.assertEquals("128", unchanged.other().options() .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); FallbackReadFileStoreTable capped = (FallbackReadFileStoreTable) PaimonJniScanner.applyBackendManifestParallelism(pair, "128", 64); Assert.assertEquals("1", capped.wrapped().options() .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); - Assert.assertEquals("64", capped.fallback().options() + Assert.assertEquals("64", capped.other().options() .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); } @@ -329,7 +330,7 @@ public void testBackendCapTraversesPrivilegeDelegate() { new Class[] {PrivilegeChecker.class}, (proxy, method, args) -> null); FileStoreTable privileged = PrivilegedFileStoreTable.wrap( - new FallbackReadFileStoreTable(main, fallback), checker, + new FallbackReadFileStoreTable(main, fallback, true), checker, Identifier.create("db", "table")); Table safe = PaimonJniScanner.applyBackendManifestParallelism( @@ -342,7 +343,7 @@ public void testBackendCapTraversesPrivilegeDelegate() { FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) planningTable; Assert.assertEquals("1", pair.wrapped().options() .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); - Assert.assertEquals("64", pair.fallback().options() + Assert.assertEquals("64", pair.other().options() .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); } @@ -353,7 +354,7 @@ public void testOldFeSystemWrapperPreservesIndependentFallbackLimits() throws Ex FileStoreTable fallback = serializableFileStoreTable(Collections.singletonMap( CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), "128")); Table wrapper = SystemTableLoader.load( - "partitions", new FallbackReadFileStoreTable(main, fallback)); + "partitions", new FallbackReadFileStoreTable(main, fallback, true)); Table safe = PaimonJniScanner.applyBackendManifestParallelism( wrapper, null, 64); @@ -363,7 +364,7 @@ public void testOldFeSystemWrapperPreservesIndependentFallbackLimits() throws Ex Assert.assertEquals("1", pair.wrapped().options() .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); - Assert.assertEquals("64", pair.fallback().options() + Assert.assertEquals("64", pair.other().options() .get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key())); } @@ -378,7 +379,7 @@ public void testSystemWrapperExposesSafeFallbackBehindPrivilegeDelegate() throws new Class[] {PrivilegeChecker.class}, (proxy, method, args) -> null); FileStoreTable privileged = PrivilegedFileStoreTable.wrap( - new FallbackReadFileStoreTable(main, fallback), checker, + new FallbackReadFileStoreTable(main, fallback, true), checker, Identifier.create("db", "table")); Table wrapper = SystemTableLoader.load("partitions", privileged); diff --git a/fe/fe-connector/fe-connector-paimon-hive-shade/pom.xml b/fe/fe-connector/fe-connector-paimon-hive-shade/pom.xml index 21d3da4c901c9a..a8f95dbeda7f37 100644 --- a/fe/fe-connector/fe-connector-paimon-hive-shade/pom.xml +++ b/fe/fe-connector/fe-connector-paimon-hive-shade/pom.xml @@ -66,6 +66,14 @@ under the License. the shade jar -> child-first duplicate-class hazard, design Risk #5). --> true + + + * + * + org.apache.httpcomponents.client5 httpclient5 @@ -306,6 +314,16 @@ under the License. true ${project.basedir}/target/dependency-reduced-pom.xml + + org.apache.paimon:paimon-hive-connector-3.1 + + + org/apache/paimon/hive/** + + *:* diff --git a/fe/fe-connector/fe-connector-paimon-hive-shade/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/fe/fe-connector/fe-connector-paimon-hive-shade/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory new file mode 100644 index 00000000000000..3df3dcc0842588 --- /dev/null +++ b/fe/fe-connector/fe-connector-paimon-hive-shade/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory @@ -0,0 +1,2 @@ +org.apache.paimon.hive.HiveCatalogFactory +org.apache.paimon.hive.HiveCatalogLockFactory diff --git a/fe/fe-connector/fe-connector-paimon/pom.xml b/fe/fe-connector/fe-connector-paimon/pom.xml index 9b17c987ba3f1d..c8a1d100bf7772 100644 --- a/fe/fe-connector/fe-connector-paimon/pom.xml +++ b/fe/fe-connector/fe-connector-paimon/pom.xml @@ -116,6 +116,22 @@ under the License. org.apache.paimon paimon-core ${paimon.version} + + + + org.apache.paimon + paimon-shade-caffeine-2 + + + org.apache.paimon + paimon-shade-guava-30 + + + org.apache.paimon + paimon-shade-jackson-2 + + - 1.3.1 + 2.0.0 3.4.4 17.0.0 From 640c77f066e19370aef0d753b09209cbb9c971aa Mon Sep 17 00:00:00 2001 From: moke-HU <25567926679@qq.com> Date: Fri, 14 Aug 2026 19:28:01 +0800 Subject: [PATCH 2/7] feat(paimon): add transactional table writes --- be/src/common/config.cpp | 5 + be/src/common/config.h | 4 + .../data_type_variant_v2_serde.cpp | 50 ++ be/src/exec/operator/operator.cpp | 3 + .../operator/paimon_table_sink_operator.cpp | 39 + .../operator/paimon_table_sink_operator.h | 106 +++ .../pipeline/pipeline_fragment_context.cpp | 12 +- .../paimon/jni_paimon_write_backend.cpp | 622 ++++++++++++++ .../writer/paimon/jni_paimon_write_backend.h | 110 +++ .../paimon/paimon_jni_memory_manager.cpp | 304 +++++++ .../writer/paimon/paimon_jni_memory_manager.h | 81 ++ .../writer/paimon/paimon_table_writer.cpp | 193 +++++ .../sink/writer/paimon/paimon_table_writer.h | 102 +++ .../sink/writer/paimon/paimon_write_backend.h | 84 ++ be/src/format/jni/jni_data_bridge.cpp | 43 + be/src/format/jni/jni_data_bridge.h | 3 + be/src/runtime/runtime_state.cpp | 42 +- be/src/runtime/runtime_state.h | 13 +- ...data_type_variant_v2_serde_output_test.cpp | 30 + .../format/table/paimon_jni_reader_test.cpp | 65 ++ .../runtime_state_block_budget_test.cpp | 26 + fe/be-java-extensions/paimon-scanner/pom.xml | 5 + .../doris/paimon/DorisMemorySegmentPool.java | 57 ++ .../doris/paimon/PaimonArrowConverter.java | 479 +++++++++++ .../doris/paimon/PaimonColumnValue.java | 81 +- .../doris/paimon/PaimonCommitCodec.java | 222 +++++ .../apache/doris/paimon/PaimonJniWriter.java | 781 ++++++++++++++++++ .../doris/paimon/PaimonWriteSchema.java | 131 +++ .../paimon/PaimonArrowConverterTest.java | 334 ++++++++ .../doris/paimon/PaimonColumnValueTest.java | 47 ++ .../doris/paimon/PaimonCommitCodecTest.java | 149 ++++ .../doris/paimon/PaimonJniScannerTest.java | 5 + .../doris/paimon/PaimonJniWriterTest.java | 195 +++++ .../doris/paimon/PaimonWriteSchemaTest.java | 200 +++++ .../connector/paimon/PaimonConnector.java | 10 +- .../paimon/PaimonConnectorMetadata.java | 23 + .../paimon/PaimonConnectorTransaction.java | 402 +++++++++ .../paimon/PaimonScanPlanProvider.java | 2 +- .../connector/paimon/PaimonTypeMapping.java | 15 +- .../connector/paimon/PaimonWriteBinding.java | 98 +++ .../paimon/PaimonWritePlanProvider.java | 312 +++++++ .../PaimonConnectorTransactionTest.java | 185 +++++ .../paimon/PaimonSchemaBuilderTest.java | 29 + .../paimon/PaimonTypeMappingReadTest.java | 22 + .../paimon/PaimonWritePlanProviderTest.java | 310 +++++++ .../apache/doris/qe/AbstractJobProcessor.java | 5 +- .../java/org/apache/doris/qe/Coordinator.java | 8 +- .../org/apache/doris/qe/QeProcessorImpl.java | 3 +- .../doris/qe/runtime/LoadProcessor.java | 9 +- .../qe/QeProcessorImplReportAckTest.java | 22 +- .../transaction/CommitDataSerializerTest.java | 39 + gensrc/thrift/DataSinks.thrift | 21 + gensrc/thrift/FrontendService.thrift | 2 + .../test_paimon_ctas_atomicity_negative.out | 9 + .../paimon/test_paimon_write_boundary.out | 18 +- ...test_paimon_ctas_atomicity_negative.groovy | 32 +- .../paimon/test_paimon_write_boundary.groovy | 33 +- 57 files changed, 6168 insertions(+), 64 deletions(-) create mode 100644 be/src/exec/operator/paimon_table_sink_operator.cpp create mode 100644 be/src/exec/operator/paimon_table_sink_operator.h create mode 100644 be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp create mode 100644 be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h create mode 100644 be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.cpp create mode 100644 be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.h create mode 100644 be/src/exec/sink/writer/paimon/paimon_table_writer.cpp create mode 100644 be/src/exec/sink/writer/paimon/paimon_table_writer.h create mode 100644 be/src/exec/sink/writer/paimon/paimon_write_backend.h create mode 100644 fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/DorisMemorySegmentPool.java create mode 100644 fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonArrowConverter.java create mode 100644 fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonCommitCodec.java create mode 100644 fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java create mode 100644 fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java create mode 100644 fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonArrowConverterTest.java create mode 100644 fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonCommitCodecTest.java create mode 100644 fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniWriterTest.java create mode 100644 fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonWriteSchemaTest.java create mode 100644 fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorTransaction.java create mode 100644 fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java create mode 100644 fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWritePlanProvider.java create mode 100644 fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorTransactionTest.java create mode 100644 fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWritePlanProviderTest.java create mode 100644 regression-test/data/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.out diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index d27cf607a36b24..8ca027e6c4ec55 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -1608,6 +1608,11 @@ DEFINE_mInt64(hive_sink_max_file_size, "1073741824"); // 1GB /** Iceberg sink configurations **/ DEFINE_mInt64(iceberg_sink_max_file_size, "1073741824"); // 1GB +/** Paimon sink configurations **/ +DEFINE_mInt64(paimon_jni_writer_memory_pool_limit_bytes, "536870912"); // 512MB +DEFINE_Validator(paimon_jni_writer_memory_pool_limit_bytes, + [](int64_t bytes) -> bool { return bytes > 0; }); + // URI scheme to Doris file type mappings used by paimon-cpp DorisFileSystem. // Each entry uses the format "=", and file_type must be one of: // local, hdfs, s3, http, broker. diff --git a/be/src/common/config.h b/be/src/common/config.h index 470d88819703de..1983712b4b6a56 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1701,6 +1701,10 @@ DECLARE_mInt64(hive_sink_max_file_size); /** Iceberg sink configurations **/ DECLARE_mInt64(iceberg_sink_max_file_size); +/** Paimon sink configurations **/ +// Hard upper bound for Doris-managed Paimon write-buffer memory per JNI writer. +DECLARE_mInt64(paimon_jni_writer_memory_pool_limit_bytes); + /** Paimon file system configurations **/ DECLARE_Strings(paimon_file_system_scheme_mappings); diff --git a/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp b/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp index 4a1d8e70b7bc4c..3ef8142b2afb43 100644 --- a/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp +++ b/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp @@ -18,6 +18,7 @@ #include "core/data_type_serde/data_type_variant_v2_serde.h" #include +#include #include #include @@ -491,6 +492,50 @@ Status write_arrow(const IColumn& column, const NullMap* null_map, Builder& buil return status; } +Status write_arrow_variant(const IColumn& column, const NullMap* null_map, + arrow::StructBuilder& builder, size_t start, size_t end) { + const auto struct_type = std::dynamic_pointer_cast(builder.type()); + if (struct_type == nullptr || builder.num_fields() != 2 || + struct_type->field(0)->name() != "value" || struct_type->field(1)->name() != "metadata") { + return Status::InvalidArgument( + "Variant Arrow output requires struct"); + } + auto* value_builder = dynamic_cast(builder.field_builder(0)); + auto* metadata_builder = dynamic_cast(builder.field_builder(1)); + if (value_builder == nullptr || metadata_builder == nullptr) { + return Status::InvalidArgument( + "Variant Arrow output requires binary value and metadata children"); + } + + Status status = Status::OK(); + visit_variant_v2_values( + column, start, end, forced_nulls(null_map), + [&](size_t) { + if (status.ok()) { + status = checkArrowStatus(builder.AppendNull(), column, builder); + } + }, + [&](size_t, VariantRef value) { + if (!status.ok()) { + return; + } + status = checkArrowStatus(builder.Append(), column, builder); + if (status.ok()) { + status = checkArrowStatus( + value_builder->Append(value.value.data, + cast_set(value.value.size)), + column, *value_builder); + } + if (status.ok()) { + status = checkArrowStatus( + metadata_builder->Append(value.metadata.data, + cast_set(value.metadata.size)), + column, *metadata_builder); + } + }); + return status; +} + } // namespace void DataTypeVariantV2SerDe::to_string(const IColumn& column, size_t row_num, BufferWritable& bw, @@ -553,6 +598,11 @@ Status DataTypeVariantV2SerDe::write_column_to_arrow(const IColumn& column, cons assert_cast(*array_builder), first, last, options); } + if (array_builder->type()->id() == arrow::Type::STRUCT) { + return write_arrow_variant(column, null_map, + assert_cast(*array_builder), first, + last); + } return Status::InvalidArgument("Unsupported arrow type for variant column: {}", array_builder->type()->name()); }); diff --git a/be/src/exec/operator/operator.cpp b/be/src/exec/operator/operator.cpp index cfc9d6645119c8..9a12adaca8e07f 100644 --- a/be/src/exec/operator/operator.cpp +++ b/be/src/exec/operator/operator.cpp @@ -64,6 +64,7 @@ #include "exec/operator/olap_scan_operator.h" #include "exec/operator/olap_table_sink_operator.h" #include "exec/operator/olap_table_sink_v2_operator.h" +#include "exec/operator/paimon_table_sink_operator.h" #include "exec/operator/partition_sort_sink_operator.h" #include "exec/operator/partition_sort_source_operator.h" #include "exec/operator/partitioned_aggregation_sink_operator.h" @@ -830,6 +831,7 @@ DECLARE_OPERATOR(OlapTableSinkV2LocalState) DECLARE_OPERATOR(HiveTableSinkLocalState) DECLARE_OPERATOR(TVFTableSinkLocalState) DECLARE_OPERATOR(IcebergTableSinkLocalState) +DECLARE_OPERATOR(PaimonTableSinkLocalState) DECLARE_OPERATOR(SpillIcebergTableSinkLocalState) DECLARE_OPERATOR(IcebergDeleteSinkLocalState) DECLARE_OPERATOR(IcebergMergeSinkLocalState) @@ -955,6 +957,7 @@ template class AsyncWriterSink; template class AsyncWriterSink; template class AsyncWriterSink; template class AsyncWriterSink; +template class AsyncWriterSink; template class AsyncWriterSink; template class AsyncWriterSink; template class AsyncWriterSink; diff --git a/be/src/exec/operator/paimon_table_sink_operator.cpp b/be/src/exec/operator/paimon_table_sink_operator.cpp new file mode 100644 index 00000000000000..c1386be558ec62 --- /dev/null +++ b/be/src/exec/operator/paimon_table_sink_operator.cpp @@ -0,0 +1,39 @@ +// 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/operator/paimon_table_sink_operator.h" + +#include "common/logging.h" + +namespace doris { + +Status PaimonTableSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) { + return Base::init(state, info); +} + +Status PaimonTableSinkOperatorX::sink_impl(RuntimeState* state, Block* in_block, bool eos) { + auto& local_state = get_local_state(state); + SCOPED_TIMER(local_state.exec_time_counter()); + COUNTER_UPDATE(local_state.rows_input_counter(), static_cast(in_block->rows())); + + // Delegate to AsyncWriterSink → PaimonTableWriter for this pipeline instance. + // Each pipeline instance has its own writer session; partition and bucket + // routing is handled internally by the Paimon SDK inside IPaimonWriter::write(). + return local_state.sink(state, in_block, eos); +} + +} // namespace doris diff --git a/be/src/exec/operator/paimon_table_sink_operator.h b/be/src/exec/operator/paimon_table_sink_operator.h new file mode 100644 index 00000000000000..3a2f885720348c --- /dev/null +++ b/be/src/exec/operator/paimon_table_sink_operator.h @@ -0,0 +1,106 @@ +// 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. + +#pragma once + +#include + +#include +#include + +#include "common/status.h" +#include "core/block/block.h" +#include "exec/operator/operator.h" +#include "exec/sink/writer/paimon/paimon_table_writer.h" +#include "runtime/runtime_state.h" + +namespace doris { + +/// Paimon table sink operator — simple pass-through to AsyncWriterSink. +/// +/// Each pipeline instance (LocalState) owns one PaimonTableWriter, which in +/// turn owns one IPaimonWriteBackend + IPaimonWriter. Pipeline parallelism +/// determines the number of concurrent Paimon writer sessions per table. FE's +/// Paimon write provider currently requires GATHER so bucket assignment sees +/// one ordered input stream. +/// +/// Partition and bucket routing is performed internally by the Paimon Java SDK +/// through JNI. Doris does not compute partition values or bucket ids; it +/// passes complete Blocks through the backend to the SDK, +/// where each row is routed via getPartition(row) + getBucket(row). +/// +/// This mirrors Iceberg's approach: IcebergTableSinkOperatorX delegates to +/// AsyncWriterSink, with partition routing inside +/// VIcebergTableWriter::write(). +class PaimonTableSinkOperatorX; + +class PaimonTableSinkLocalState final + : public AsyncWriterSink { +public: + using Base = AsyncWriterSink; + using Parent = PaimonTableSinkOperatorX; + ENABLE_FACTORY_CREATOR(PaimonTableSinkLocalState); + PaimonTableSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state) + : Base(parent, state) {} + Status init(RuntimeState* state, LocalSinkStateInfo& info) override; + Status open(RuntimeState* state) override { + SCOPED_TIMER(exec_time_counter()); + SCOPED_TIMER(_open_timer); + return Base::open(state); + } + + friend class PaimonTableSinkOperatorX; +}; + +class PaimonTableSinkOperatorX final : public DataSinkOperatorX { +public: + using Base = DataSinkOperatorX; + PaimonTableSinkOperatorX(ObjectPool* pool, int operator_id, const RowDescriptor& row_desc, + const std::vector& t_output_expr) + : Base(operator_id, 0, 0), + _row_desc(row_desc), + _t_output_expr(t_output_expr), + _pool(pool) {} + + Status init(const TDataSink& thrift_sink) override { + RETURN_IF_ERROR(Base::init(thrift_sink)); + DCHECK(thrift_sink.__isset.paimon_table_sink); + RETURN_IF_ERROR(VExpr::create_expr_trees(_t_output_expr, _output_vexpr_ctxs)); + return Status::OK(); + } + + Status prepare(RuntimeState* state) override { + RETURN_IF_ERROR(Base::prepare(state)); + RETURN_IF_ERROR(VExpr::prepare(_output_vexpr_ctxs, state, _row_desc)); + return VExpr::open(_output_vexpr_ctxs, state); + } + + Status sink_impl(RuntimeState* state, Block* in_block, bool eos) override; + +private: + friend class PaimonTableSinkLocalState; + template + requires(std::is_base_of_v) + friend class AsyncWriterSink; + + const RowDescriptor& _row_desc; + VExprContextSPtrs _output_vexpr_ctxs; + const std::vector& _t_output_expr; + ObjectPool* _pool = nullptr; +}; + +} // namespace doris diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp b/be/src/exec/pipeline/pipeline_fragment_context.cpp index 2fa064c8a68e09..abf8db9c6fa102 100644 --- a/be/src/exec/pipeline/pipeline_fragment_context.cpp +++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp @@ -88,6 +88,7 @@ #include "exec/operator/olap_scan_operator.h" #include "exec/operator/olap_table_sink_operator.h" #include "exec/operator/olap_table_sink_v2_operator.h" +#include "exec/operator/paimon_table_sink_operator.h" #include "exec/operator/partition_sort_sink_operator.h" #include "exec/operator/partition_sort_source_operator.h" #include "exec/operator/partitioned_aggregation_sink_operator.h" @@ -1352,6 +1353,14 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS } break; } + case TDataSinkType::PAIMON_TABLE_SINK: { + if (!thrift_sink.__isset.paimon_table_sink) { + return Status::InternalError("Missing paimon table sink."); + } + _sink = std::make_shared(pool, next_sink_operator_id(), row_desc, + output_exprs); + break; + } case TDataSinkType::ICEBERG_DELETE_SINK: { if (!thrift_sink.__isset.iceberg_delete_sink) { return Status::InternalError("Missing iceberg delete sink."); @@ -2585,7 +2594,8 @@ void PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r PrintThriftNetworkAddress(req.coord_addr), e.what()); } - const bool requires_external_file_ack = params.__isset.iceberg_commit_datas; + const bool requires_external_file_ack = + params.__isset.iceberg_commit_datas || params.__isset.paimon_commit_messages; if (rpc_status.ok() && requires_external_file_ack && (!res.__isset.external_file_commit_data_accepted || !res.external_file_commit_data_accepted)) { diff --git a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp new file mode 100644 index 00000000000000..149835b68c9210 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp @@ -0,0 +1,622 @@ +// 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 +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "common/check.h" +#include "common/logging.h" +#include "core/data_type/data_type_agg_state.h" +#include "core/data_type/data_type_array.h" +#include "core/data_type/data_type_map.h" +#include "core/data_type/data_type_struct.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 "storage/options.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"; + +std::atomic& paimon_jni_close_failed() { + static auto* failed = new std::atomic(false); + return *failed; +} + +std::mutex& retained_memory_managers_mutex() { + static auto* mutex = new std::mutex(); + return *mutex; +} + +std::vector>& retained_memory_managers() { + static auto* managers = new std::vector>(); + return *managers; +} + +void retain_memory_after_failed_close(std::unique_ptr manager) { + paimon_jni_close_failed().store(true, std::memory_order_release); + if (manager == nullptr) { + return; + } + std::lock_guard lock(retained_memory_managers_mutex()); + retained_memory_managers().emplace_back(std::move(manager)); +} + +Status convert_to_paimon_arrow_type(const DataTypePtr& origin_type, + std::shared_ptr* result, + const std::string& timezone) { + const DataTypePtr type = get_serialized_type(origin_type); + switch (type->get_primitive_type()) { + case TYPE_VARIANT: + // Paimon consumes the lossless Variant V2 representation. Keeping both children non-null + // distinguishes a SQL NULL struct from a non-null Variant value. + *result = arrow::struct_({arrow::field("value", arrow::binary(), false), + arrow::field("metadata", arrow::binary(), false)}); + return Status::OK(); + case TYPE_ARRAY: { + const auto& array_type = assert_cast(*remove_nullable(type)); + std::shared_ptr element_type; + RETURN_IF_ERROR(convert_to_paimon_arrow_type(array_type.get_nested_type(), &element_type, + timezone)); + *result = std::make_shared(element_type); + return Status::OK(); + } + case TYPE_MAP: { + const auto& map_type = assert_cast(*remove_nullable(type)); + std::shared_ptr key_type; + std::shared_ptr value_type; + RETURN_IF_ERROR(convert_to_paimon_arrow_type(map_type.get_key_type(), &key_type, timezone)); + RETURN_IF_ERROR( + convert_to_paimon_arrow_type(map_type.get_value_type(), &value_type, timezone)); + *result = std::make_shared(key_type, value_type); + return Status::OK(); + } + case TYPE_STRUCT: { + const auto& struct_type = assert_cast(*remove_nullable(type)); + std::vector> fields; + fields.reserve(struct_type.get_elements().size()); + for (size_t i = 0; i < struct_type.get_elements().size(); ++i) { + const DataTypePtr& element = struct_type.get_element(i); + std::shared_ptr field_type; + RETURN_IF_ERROR(convert_to_paimon_arrow_type(element, &field_type, timezone)); + fields.push_back(arrow::field(struct_type.get_element_name(i), field_type, + element->is_nullable())); + } + *result = arrow::struct_(std::move(fields)); + return Status::OK(); + } + default: + return convert_to_arrow_type(origin_type, result, timezone); + } +} + +Status get_paimon_arrow_schema_from_block(const Block& block, + std::shared_ptr* result) { + std::vector> fields; + fields.reserve(block.columns()); + for (const auto& type_and_name : block) { + std::shared_ptr arrow_type; + RETURN_IF_ERROR(convert_to_paimon_arrow_type(type_and_name.type, &arrow_type, "")); + fields.push_back(create_arrow_field_with_metadata( + type_and_name.name, arrow_type, type_and_name.type->is_nullable(), + type_and_name.type->get_primitive_type())); + } + *result = arrow::schema(std::move(fields)); + return Status::OK(); +} +} // 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(env), JNI_VERSION_1_8); + if (result == JNI_EDETACHED) { + result = jvm->AttachCurrentThread(reinterpret_cast(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() { + Status st = close(); + if (!st.ok()) { + LOG(WARNING) << "Failed to close Paimon JNI backend during destruction: " << st.to_string(); + } +} + +Status JniPaimonWriteBackend::close() { + if (_jni_writer_obj == nullptr && _jni_writer_cls == nullptr) { + _memory_manager.reset(); + _opened = false; + return Status::OK(); + } + + JNIEnv* env = nullptr; + Status env_status = _get_jni_env(&env); + if (!env_status.ok()) { + bool java_users_may_exist = _jni_writer_obj != nullptr; + // JNI global references cannot be released without an environment. + // Deliberately abandon the handles so the Java writer remains alive. + _jni_writer_obj = nullptr; + _jni_writer_cls = nullptr; + if (java_users_may_exist) { + retain_memory_after_failed_close(std::move(_memory_manager)); + } else { + _memory_manager.reset(); + } + _opened = false; + return env_status; + } + + Status close_status = Status::OK(); + if (_jni_writer_obj != nullptr) { + _refresh_memory_profile(); + if (_close_id == nullptr) { + close_status = Status::InternalError("PaimonJniWriter.close method is unavailable"); + } else { + env->CallVoidMethod(_jni_writer_obj, _close_id); + close_status = _check_jni_exception(env, "close PaimonJniWriter"); + } + env->DeleteGlobalRef(_jni_writer_obj); + _jni_writer_obj = nullptr; + } + if (_jni_writer_cls != nullptr) { + env->DeleteGlobalRef(_jni_writer_cls); + _jni_writer_cls = nullptr; + } + + if (close_status.ok()) { + _memory_manager.reset(); + } else { + if (_memory_manager != nullptr) { + LOG(WARNING) + << "Retaining Paimon JNI native memory after an unconfirmed Java close: limit=" + << PrettyPrinter::print_bytes(_memory_manager->memory_limit()) << ", peak=" + << PrettyPrinter::print_bytes(_memory_manager->native_peak_allocated_bytes()); + } + // Paimon may still have asynchronous flush or compaction tasks using + // MemorySegments backed by these pages. Retain ownership until process + // exit and reject new writers below. Retention is therefore limited to + // writers which were already open when the first close failure occurred. + retain_memory_after_failed_close(std::move(_memory_manager)); + } + _opened = false; + return close_status; +} + +Status JniPaimonWriteBackend::_check_jni_exception(JNIEnv* env, const std::string& method_name) { + if (env->ExceptionCheck()) { + Status st = + Jni::Env::GetJniExceptionMsg(env, true, "JNI exception in " + method_name + ": "); + LOG(WARNING) << st.to_string(); + return st; + } + return Status::OK(); +} + +Status JniPaimonWriteBackend::_load_writer_class(JNIEnv* env, jclass* writer_class) { + jclass loader_class = env->FindClass(SCANNER_LOADER_CLASS); + RETURN_IF_ERROR(_check_jni_exception(env, "find ScannerLoader")); + + jmethodID loader_constructor = env->GetMethodID(loader_class, "", "()V"); + jmethodID get_loaded_class = env->GetMethodID(loader_class, "getLoadedClass", + "(Ljava/lang/String;)Ljava/lang/Class;"); + RETURN_IF_ERROR(_check_jni_exception(env, "resolve ScannerLoader methods")); + + jobject loader = env->NewObject(loader_class, loader_constructor); + jstring class_name = env->NewStringUTF(PAIMON_JNI_WRITER_CLASS); + auto* loaded_class = + static_cast(env->CallObjectMethod(loader, get_loaded_class, class_name)); + RETURN_IF_ERROR(_check_jni_exception(env, "load PaimonJniWriter")); + + *writer_class = loaded_class; + env->DeleteLocalRef(class_name); + env->DeleteLocalRef(loader); + env->DeleteLocalRef(loader_class); + return Status::OK(); +} + +static jobject _to_java_options(JNIEnv* env, const std::map& options) { + jclass map_cls = env->FindClass("java/util/HashMap"); + jmethodID map_ctor = env->GetMethodID(map_cls, "", "()V"); + jmethodID put_method = env->GetMethodID( + map_cls, "put", "(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;"); + + jobject map_obj = env->NewObject(map_cls, map_ctor); + for (const auto& kv : options) { + jstring key = env->NewStringUTF(kv.first.c_str()); + jstring val = env->NewStringUTF(kv.second.c_str()); + env->CallObjectMethod(map_obj, put_method, key, val); + env->DeleteLocalRef(key); + env->DeleteLocalRef(val); + } + env->DeleteLocalRef(map_cls); + return map_obj; +} + +Status JniPaimonWriteBackend::abort_prepared_commit( + const TPaimonTableSink& sink, const std::vector& commit_messages) { + if (commit_messages.empty()) { + return Status::OK(); + } + + JNIEnv* env = nullptr; + RETURN_IF_ERROR(_get_jni_env(&env)); + + jclass writer_class = nullptr; + RETURN_IF_ERROR(_load_writer_class(env, &writer_class)); + jmethodID abort_id = + env->GetStaticMethodID(writer_class, "abortPreparedCommit", + "(Ljava/lang/String;Ljava/util/Map;Ljava/lang/String;[[B)V"); + Status method_status = _check_jni_exception(env, "resolve abortPreparedCommit"); + if (!method_status.ok()) { + env->DeleteLocalRef(writer_class); + return method_status; + } + + const std::map empty_config; + jstring serialized_table = env->NewStringUTF(sink.serialized_table.c_str()); + jobject hadoop_config = + _to_java_options(env, sink.__isset.hadoop_config ? sink.hadoop_config : empty_config); + jstring commit_user = env->NewStringUTF(sink.commit_user.c_str()); + jclass byte_array_class = env->FindClass("[B"); + auto payloads = env->NewObjectArray(static_cast(commit_messages.size()), + byte_array_class, nullptr); + Status allocation_status = _check_jni_exception(env, "allocate abortPreparedCommit arguments"); + if (allocation_status.ok()) { + for (size_t i = 0; i < commit_messages.size(); ++i) { + const auto& message = commit_messages[i]; + DORIS_CHECK(message.__isset.payload); + auto payload = env->NewByteArray(static_cast(message.payload.size())); + env->SetByteArrayRegion(payload, 0, static_cast(message.payload.size()), + reinterpret_cast(message.payload.data())); + env->SetObjectArrayElement(payloads, static_cast(i), payload); + env->DeleteLocalRef(payload); + } + allocation_status = _check_jni_exception(env, "populate abortPreparedCommit payloads"); + } + + Status abort_status = allocation_status; + if (abort_status.ok()) { + env->CallStaticVoidMethod(writer_class, abort_id, serialized_table, hadoop_config, + commit_user, payloads); + abort_status = _check_jni_exception(env, "abort prepared Paimon commit"); + } + + if (payloads != nullptr) { + env->DeleteLocalRef(payloads); + } + if (byte_array_class != nullptr) { + env->DeleteLocalRef(byte_array_class); + } + if (commit_user != nullptr) { + env->DeleteLocalRef(commit_user); + } + if (hadoop_config != nullptr) { + env->DeleteLocalRef(hadoop_config); + } + if (serialized_table != nullptr) { + env->DeleteLocalRef(serialized_table); + } + env->DeleteLocalRef(writer_class); + return abort_status; +} + +Status JniPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* state, + RuntimeProfile* profile) { + if (paimon_jni_close_failed().load(std::memory_order_acquire)) { + return Status::InternalError( + "Paimon JNI writes are disabled on this BE because a previous Java writer close " + "failed; restart the BE to reclaim retained native memory safely"); + } + _sink = sink; + DORIS_CHECK(sink.__isset.column_names); + DORIS_CHECK(sink.__isset.write_mode); + DORIS_CHECK(sink.__isset.serialized_table); + DORIS_CHECK(!sink.serialized_table.empty()); + DORIS_CHECK(sink.__isset.transaction_id); + DORIS_CHECK(sink.transaction_id > 0); + DORIS_CHECK(sink.__isset.commit_user); + DORIS_CHECK(!sink.commit_user.empty()); + DORIS_CHECK(profile != nullptr); + + RETURN_IF_ERROR(PaimonJniMemoryManager::create(state, &_memory_manager)); + RuntimeProfile* jni_profile = profile->create_child("JniPaimonWriteBackend", true, true); + _native_page_memory_limit = ADD_COUNTER(jni_profile, "NativePageMemoryLimit", TUnit::BYTES); + _native_page_memory_peak = ADD_COUNTER(jni_profile, "NativePageMemoryPeak", TUnit::BYTES); + + JNIEnv* env = nullptr; + RETURN_IF_ERROR(_get_jni_env(&env)); + + // Step 1: Load PaimonJniWriter class through ScannerLoader (Paimon jars are + // not on the default application classpath, so FindClass won't work). + jclass local_cls = nullptr; + RETURN_IF_ERROR(_load_writer_class(env, &local_cls)); + _jni_writer_cls = static_cast(env->NewGlobalRef(local_cls)); + env->DeleteLocalRef(local_cls); + RETURN_IF_ERROR(PaimonJniMemoryManager::register_natives(env, _jni_writer_cls)); + + // Step 2: Cache JNI method IDs for write, prepareCommit, abort, close. + jmethodID open_id = env->GetMethodID( + _jni_writer_cls, "open", + "(Ljava/lang/String;Ljava/util/Map;[Ljava/lang/String;JLjava/lang/String;ZLjava/lang/" + "String;Ljava/lang/String;JJ)V"); + _write_id = env->GetMethodID(_jni_writer_cls, "write", "(Ljava/nio/ByteBuffer;)V"); + _prepare_commit_id = env->GetMethodID(_jni_writer_cls, "prepareCommit", "()[[B"); + _abort_id = env->GetMethodID(_jni_writer_cls, "abort", "()V"); + _close_id = env->GetMethodID(_jni_writer_cls, "close", "()V"); + RETURN_IF_ERROR(_check_jni_exception(env, "GetMethodID")); + + // Step 3: Create the Java PaimonJniWriter instance. + jmethodID ctor_id = env->GetMethodID(_jni_writer_cls, "", "()V"); + jobject local_obj = env->NewObject(_jni_writer_cls, ctor_id); + RETURN_IF_ERROR(_check_jni_exception(env, "NewObject")); + _jni_writer_obj = env->NewGlobalRef(local_obj); + env->DeleteLocalRef(local_obj); + + // Step 4: Build Java arguments and call PaimonJniWriter.open(). + const std::map empty_config; + jstring j_serialized_table = env->NewStringUTF(sink.serialized_table.c_str()); + jobject j_hadoop_config = + _to_java_options(env, sink.__isset.hadoop_config ? sink.hadoop_config : empty_config); + jstring j_commit_user = env->NewStringUTF(sink.commit_user.c_str()); + jstring j_time_zone = env->NewStringUTF(state->timezone().c_str()); + std::vector spill_directories; + for (const auto& store_path : state->exec_env()->store_paths()) { + spill_directories.push_back(store_path.path + "/" + + std::string(PAIMON_JNI_WRITER_IO_TMP_DIR)); + } + DORIS_CHECK(!spill_directories.empty()); + jstring j_spill_directories = env->NewStringUTF(join(spill_directories, ":").c_str()); + + jclass string_cls = env->FindClass("java/lang/String"); + jobjectArray j_cols = + env->NewObjectArray(static_cast(sink.column_names.size()), string_cls, nullptr); + for (size_t i = 0; i < sink.column_names.size(); ++i) { + jstring str = env->NewStringUTF(sink.column_names[i].c_str()); + env->SetObjectArrayElement(j_cols, static_cast(i), str); + env->DeleteLocalRef(str); + } + + env->CallVoidMethod(_jni_writer_obj, open_id, j_serialized_table, j_hadoop_config, j_cols, + static_cast(sink.transaction_id), j_commit_user, + static_cast(sink.write_mode == TPaimonWriteMode::OVERWRITE), + j_time_zone, j_spill_directories, + static_cast(_memory_manager->memory_limit()), + reinterpret_cast(_memory_manager.get())); + Status st = _check_jni_exception(env, "open"); + + env->DeleteLocalRef(j_serialized_table); + env->DeleteLocalRef(j_hadoop_config); + env->DeleteLocalRef(j_commit_user); + env->DeleteLocalRef(j_time_zone); + env->DeleteLocalRef(j_spill_directories); + env->DeleteLocalRef(j_cols); + env->DeleteLocalRef(string_cls); + + if (st.ok()) { + _opened = true; + _refresh_memory_profile(); + LOG(INFO) << "Paimon JNI writer memory limit: " + << PrettyPrinter::print_bytes(_memory_manager->memory_limit()) + << ", local_sink_count=" << std::max(1, state->num_local_sink()); + } + return st; +} + +Status JniPaimonWriteBackend::create_writer( // NOLINT(readability-make-member-function-const) + std::unique_ptr* writer) { + DORIS_CHECK(_opened); + *writer = std::make_unique(_jni_writer_obj, _write_id, _prepare_commit_id, + _abort_id, std::make_unique>(), + _sink); + return Status::OK(); +} + +JniPaimonWriter::JniPaimonWriter(jobject jni_writer_obj, jmethodID write_id, + jmethodID prepare_commit_id, jmethodID abort_id, + std::unique_ptr> arrow_pool, + TPaimonTableSink sink) + : _jni_writer_obj(jni_writer_obj), + _write_id(write_id), + _prepare_commit_id(prepare_commit_id), + _abort_id(abort_id), + _arrow_pool(std::move(arrow_pool)), + _sink(std::move(sink)) {} + +Status JniPaimonWriter::_write_projected_block(RuntimeState* state, Block& block) { + if (block.rows() == 0) { + return Status::OK(); + } + + // Use Thrift column_names as the authoritative schema source for both + // Arrow schema construction and Java-side write type derivation. + DORIS_CHECK(_sink.__isset.column_names); + DORIS_CHECK_EQ(_sink.column_names.size(), block.columns()); + for (size_t i = 0; i < _sink.column_names.size(); ++i) { + block.get_by_position(i).name = _sink.column_names[i]; + } + + // Pipeline: Doris Block → Arrow Schema → Arrow RecordBatch → IPC Stream → JNI direct buffer + // + // Step 1: Build Arrow schema from the projected Block. + // Paimon write timestamps are transported as civil-time fields. The Java writer uses the + // pinned Paimon target type to preserve NTZ values or convert LTZ values with the session zone. + // Variant V2 is transported losslessly as its value/metadata pair, including nested Variant. + std::shared_ptr arrow_schema; + RETURN_IF_ERROR(get_paimon_arrow_schema_from_block(block, &arrow_schema)); + + // Step 2: Convert Doris Block columns to an Arrow RecordBatch. + std::shared_ptr record_batch; + RETURN_IF_ERROR(convert_to_arrow_batch(block, arrow_schema, _arrow_pool.get(), &record_batch, + state->timezone_obj())); + + // Step 3: Serialize the RecordBatch to Arrow IPC Stream format in memory. + auto out_stream_res = arrow::io::BufferOutputStream::Create(4096, _arrow_pool.get()); + if (!out_stream_res.ok()) { + return Status::InternalError("Arrow BufferOutputStream create failed: {}", + out_stream_res.status().ToString()); + } + auto out_stream = *out_stream_res; + + auto writer_res = arrow::ipc::MakeStreamWriter(out_stream, arrow_schema); + if (!writer_res.ok()) { + return Status::InternalError("Arrow StreamWriter create failed: {}", + writer_res.status().ToString()); + } + auto ipc_writer = *writer_res; + if (!ipc_writer->WriteRecordBatch(*record_batch).ok()) { + return Status::InternalError("Arrow WriteRecordBatch failed"); + } + if (!ipc_writer->Close().ok()) { + return Status::InternalError("Arrow StreamWriter close failed"); + } + + auto buffer_res = out_stream->Finish(); + if (!buffer_res.ok()) { + return Status::InternalError("Arrow output stream finish failed: {}", + buffer_res.status().ToString()); + } + std::shared_ptr buffer = *buffer_res; + + // Step 4: Wrap the IPC buffer in a JNI direct ByteBuffer (zero-copy) and + // call PaimonJniWriter.write(ByteBuffer). Java side reads the Arrow IPC + // stream via ArrowStreamReader. + JNIEnv* env = nullptr; + RETURN_IF_ERROR(_get_jni_env(&env)); + + jobject direct_buffer = + env->NewDirectByteBuffer(buffer->mutable_data(), static_cast(buffer->size())); + RETURN_IF_ERROR(Jni::Env::GetJniExceptionMsg( + env, false, "JNI exception in NewDirectByteBuffer for PaimonJniWriter::write: ")); + + env->CallVoidMethod(_jni_writer_obj, _write_id, direct_buffer); + Status write_status = + Jni::Env::GetJniExceptionMsg(env, false, "JNI exception in JniPaimonWriter::write: "); + env->DeleteLocalRef(direct_buffer); + return write_status; +} + +Status JniPaimonWriter::write(RuntimeState* state, Block& block) { + return _write_projected_block(state, block); +} + +Status JniPaimonWriter::prepare_commit(std::vector& messages) { + JNIEnv* env = nullptr; + RETURN_IF_ERROR(_get_jni_env(&env)); + + // Call PaimonJniWriter.prepareCommit() which returns byte[][] — + // each element is a DPCM-framed serialized CommitMessage chunk produced + // by PaimonCommitCodec.encode(). + jobject j_payloads_obj = env->CallObjectMethod(_jni_writer_obj, _prepare_commit_id); + Status st = Jni::Env::GetJniExceptionMsg(env, false, "JNI exception in prepareCommit: "); + if (!st.ok()) { + return st; + } + + if (j_payloads_obj == nullptr) { + return Status::InternalError("PaimonJniWriter.prepareCommit returned null"); + } + + // Unpack the byte[][] into TPaimonCommitMessage structs for FE transport. + auto* j_payloads = static_cast(j_payloads_obj); + jsize num_payloads = env->GetArrayLength(j_payloads); + + for (jsize i = 0; i < num_payloads; ++i) { + auto j_bytes = static_cast(env->GetObjectArrayElement(j_payloads, i)); + if (j_bytes == nullptr) { + env->DeleteLocalRef(j_payloads); + return Status::InternalError("PaimonJniWriter.prepareCommit returned a null payload"); + } + jsize len = env->GetArrayLength(j_bytes); + if (len == 0) { + env->DeleteLocalRef(j_bytes); + env->DeleteLocalRef(j_payloads); + return Status::InternalError("PaimonJniWriter.prepareCommit returned an empty payload"); + } + jbyte* bytes = env->GetByteArrayElements(j_bytes, nullptr); + if (bytes == nullptr) { + env->DeleteLocalRef(j_bytes); + env->DeleteLocalRef(j_payloads); + RETURN_IF_ERROR(Jni::Env::GetJniExceptionMsg( + env, false, "JNI exception while reading Paimon commit payload: ")); + return Status::InternalError("Failed to read Paimon commit payload"); + } + std::string payload(reinterpret_cast(bytes), static_cast(len)); + TPaimonCommitMessage msg; + msg.__set_payload(payload); + messages.emplace_back(std::move(msg)); + env->ReleaseByteArrayElements(j_bytes, bytes, JNI_ABORT); + env->DeleteLocalRef(j_bytes); + } + env->DeleteLocalRef(j_payloads); + return Status::OK(); +} + +Status JniPaimonWriter::abort() { + JNIEnv* env = nullptr; + RETURN_IF_ERROR(_get_jni_env(&env)); + env->CallVoidMethod(_jni_writer_obj, _abort_id); + return Jni::Env::GetJniExceptionMsg(env, true, "JNI exception in abort: "); +} + +void JniPaimonWriteBackend::_refresh_memory_profile() { + if (_memory_manager == nullptr) { + return; + } + COUNTER_SET(_native_page_memory_limit, _memory_manager->memory_limit()); + COUNTER_SET(_native_page_memory_peak, _memory_manager->native_peak_allocated_bytes()); +} + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h new file mode 100644 index 00000000000000..46cea7da85f7a6 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h @@ -0,0 +1,110 @@ +// 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. + +#pragma once + +#include +#include + +#include +#include + +#include "common/status.h" +#include "exec/sink/writer/paimon/paimon_jni_memory_manager.h" +#include "exec/sink/writer/paimon/paimon_write_backend.h" +#include "format/parquet/arrow_memory_pool.h" +#include "runtime/runtime_profile.h" + +namespace doris { + +class RuntimeState; + +/// JNI backend that owns the Java PaimonJniWriter object and its JNI method +/// handles. Creates lightweight JniPaimonWriter adapters that share this +/// backend's JVM connection. +/// +/// Each JniPaimonWriteBackend corresponds to one Java PaimonJniWriter +/// instance; the JniPaimonWriter adapters are thin wrappers that delegate +/// write/prepare_commit/abort calls through the cached JNI method IDs. JNI-only +/// memory ownership and Profile counters stay here and are not part of the +/// common backend contract. +class JniPaimonWriteBackend final : public IPaimonWriteBackend { +public: + ~JniPaimonWriteBackend() override; + + Status open(const TPaimonTableSink& sink, RuntimeState* state, + RuntimeProfile* profile) override; + Status create_writer(std::unique_ptr* writer) override; + Status close() override; + + /// Abort prepared files without retaining the original Java writer object. + static Status abort_prepared_commit(const TPaimonTableSink& sink, + const std::vector& commit_messages); + +private: + static Status _check_jni_exception(JNIEnv* env, const std::string& method_name); + static Status _load_writer_class(JNIEnv* env, jclass* writer_class); + void _refresh_memory_profile(); + + // JNI global references — live for the duration of this backend. + jclass _jni_writer_cls = nullptr; + jobject _jni_writer_obj = nullptr; + + // Cached JNI method IDs for the PaimonJniWriter Java methods. + jmethodID _write_id = nullptr; + jmethodID _prepare_commit_id = nullptr; + jmethodID _abort_id = nullptr; + jmethodID _close_id = nullptr; + + TPaimonTableSink _sink; + std::unique_ptr _memory_manager; + RuntimeProfile::Counter* _native_page_memory_limit = nullptr; + RuntimeProfile::Counter* _native_page_memory_peak = nullptr; + bool _opened = false; +}; + +/// Lightweight C++ adapter that delegates to the shared JNI backend. +/// +/// Owns the Arrow memory pool used for Block → Arrow IPC conversion. +/// Each JniPaimonWriter is created by JniPaimonWriteBackend::create_writer() +/// and shares the backend's JNI method IDs and Java writer object reference. +class JniPaimonWriter final : public IPaimonWriter { +public: + JniPaimonWriter(jobject jni_writer_obj, jmethodID write_id, jmethodID prepare_commit_id, + jmethodID abort_id, std::unique_ptr> arrow_pool, + TPaimonTableSink sink); + + Status write(RuntimeState* state, Block& block) override; + Status prepare_commit(std::vector& messages) override; + Status abort() override; + +private: + /// Convert Block → Arrow RecordBatch → IPC Stream, then pass to Java via JNI direct buffer. + Status _write_projected_block(RuntimeState* state, Block& block); + + // Shared JNI state (owned by JniPaimonWriteBackend, not this adapter). + jobject _jni_writer_obj; + jmethodID _write_id; + jmethodID _prepare_commit_id; + jmethodID _abort_id; + + // Arrow resources owned by this writer adapter. + std::unique_ptr> _arrow_pool; + TPaimonTableSink _sink; +}; + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.cpp b/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.cpp new file mode 100644 index 00000000000000..63eae9904e8c73 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.cpp @@ -0,0 +1,304 @@ +// 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_jni_memory_manager.h" + +#include +#include +#include +#include + +#include "common/check.h" +#include "common/config.h" +#include "common/exception.h" +#include "common/logging.h" +#include "core/allocator.h" +#include "runtime/memory/mem_tracker_limiter.h" +#include "runtime/query_context.h" +#include "runtime/runtime_state.h" +#include "runtime/thread_context.h" +#include "util/defer_op.h" +#include "util/jni-util.h" +#include "util/pretty_printer.h" + +namespace doris { + +class PaimonJniMemoryManager::Impl { +public: + Impl(std::shared_ptr resource_context, int64_t memory_limit) + : _resource_context(std::move(resource_context)), _memory_limit(memory_limit) { + DORIS_CHECK(_resource_context != nullptr); + DORIS_CHECK(_memory_limit > 0); + } + + ~Impl() { + // Java may retain direct buffers until its writer is closed. Release + // every outstanding page here as the final native ownership boundary. + try { + release_all_pages(); + } catch (const std::exception& e) { + LOG(WARNING) << "Failed to release Paimon JNI native memory: " << e.what(); + } catch (...) { + LOG(WARNING) << "Failed to release Paimon JNI native memory: unknown exception"; + } + } + + jobject allocate_page(JNIEnv* env, jint bytes) { + if (bytes <= 0) { + throw Exception(Status::InvalidArgument( + "Paimon JNI memory page size must be positive, actual={}", bytes)); + } + + // Reserve the writer-local budget before entering the allocator. This + // prevents concurrent JNI callbacks from transiently allocating past + // the configured cap and only discovering it after query accounting + // or the system allocator has already rejected the request. + { + std::lock_guard lock(_mutex); + if (bytes > _memory_limit - _native_allocated_bytes - _native_reserved_bytes) { + throw Exception(Status::Error( + "Paimon JNI write buffer exceeded its {} native memory limit", + PrettyPrinter::print_bytes(_memory_limit))); + } + _native_reserved_bytes += bytes; + } + bool reservation_committed = false; + Defer rollback_reservation {[&]() { + if (!reservation_committed) { + std::lock_guard lock(_mutex); + _native_reserved_bytes -= bytes; + } + }}; + + // Allocate and account while attached to the query's resource + // context. The callback can run on a JVM-created thread, so merely + // relying on the calling BE thread's context would bypass query + // memory accounting. + void* address = with_resource_context([&]() { + enable_thread_catch_bad_alloc++; + Defer restore_bad_alloc_catch {[&]() { enable_thread_catch_bad_alloc--; }}; + void* allocated = _allocator.alloc(static_cast(bytes)); + try { + std::lock_guard lock(_mutex); + _allocations.emplace_back(allocated, static_cast(bytes)); + _native_reserved_bytes -= bytes; + _native_allocated_bytes += bytes; + _native_peak_allocated_bytes = + std::max(_native_peak_allocated_bytes, _native_allocated_bytes); + reservation_committed = true; + } catch (...) { + _allocator.free(allocated, static_cast(bytes)); + throw; + } + return allocated; + }); + + // NewDirectByteBuffer does not copy memory; Paimon will read/write the + // page directly. If JNI rejects the address, undo the native + // allocation and its accounting entry before returning. + jobject buffer = env->NewDirectByteBuffer(address, bytes); + if (buffer == nullptr || env->ExceptionCheck()) { + remove_and_free_page(address, static_cast(bytes)); + return nullptr; + } + return buffer; + } + + int64_t memory_limit() const { return _memory_limit; } + + int64_t native_peak_allocated_bytes() const { + std::lock_guard lock(_mutex); + return _native_peak_allocated_bytes; + } + +private: + template + auto with_resource_context(Function&& function) + -> decltype(std::forward(function)()) { + // JNI normally re-enters on an attached async-writer thread. Attach + // Java-created threads explicitly too, so every allocation/free is + // charged to the query rather than to an unrelated thread context. + if (!pthread_context_ptr_init && bthread_self() == 0) { + SCOPED_ATTACH_TASK(_resource_context); + return std::forward(function)(); + } + if (thread_context()->is_attach_task()) { + SCOPED_SWITCH_RESOURCE_CONTEXT(_resource_context); + return std::forward(function)(); + } + SCOPED_ATTACH_TASK(_resource_context); + return std::forward(function)(); + } + + void release_all_pages() { + // Detach ownership from the bookkeeping vector under the lock, then + // free outside the lock. Allocator/free may invoke code that takes + // unrelated locks and must not block page accounting readers. + std::vector> allocations; + { + std::lock_guard lock(_mutex); + allocations.swap(_allocations); + _native_allocated_bytes = 0; + } + if (allocations.empty()) { + return; + } + + with_resource_context([&]() { + for (const auto& [address, bytes] : allocations) { + _allocator.free(address, bytes); + } + std::vector>().swap(allocations); + }); + } + + void remove_and_free_page(void* address, size_t bytes) { + // Roll back a page whose Java direct-buffer wrapper could not be + // created. The address is removed under the same lock used by the + // normal accounting path, while the potentially expensive free is + // performed after releasing it. + { + std::lock_guard lock(_mutex); + auto it = std::find_if( + _allocations.begin(), _allocations.end(), + [&](const auto& allocation) { return allocation.first == address; }); + if (it != _allocations.end()) { + _allocations.erase(it); + _native_allocated_bytes -= bytes; + } + } + with_resource_context([&]() { _allocator.free(address, bytes); }); + } + + // Query resource context used for all native allocator operations. + std::shared_ptr _resource_context; + // Immutable per-writer cap, calculated by PaimonJniMemoryManager::create. + const int64_t _memory_limit; + // Doris allocator used instead of JVM/Arrow allocation so native pages are + // visible to Doris' memory accounting and allocator hooks. + Allocator _allocator; + // Protects the allocation list and both usage counters. JNI callbacks and + // Java close/finalizer paths may arrive concurrently. + mutable std::mutex _mutex; + // Every entry is (native address, size) and remains here until released. + std::vector> _allocations; + // Bytes reserved by callbacks which have passed the local limit check but + // have not yet completed their allocator call. + int64_t _native_reserved_bytes = 0; + // Committed and high-water native page usage, respectively. + int64_t _native_allocated_bytes = 0; + int64_t _native_peak_allocated_bytes = 0; +}; + +namespace { + +jobject allocate_paimon_memory_page(JNIEnv* env, jclass, jlong manager_handle, jint bytes) { + // This is called from PaimonJniWriter's Java memory pool. The handle is + // the native manager address passed when the writer is opened; ownership + // stays with the C++ writer/backend, so this callback must never delete it. + auto* manager = reinterpret_cast(manager_handle); + if (manager == nullptr) { + jclass exception_class = env->FindClass("java/lang/IllegalStateException"); + env->ThrowNew(exception_class, "Paimon JNI memory manager is null"); + env->DeleteLocalRef(exception_class); + return nullptr; + } + try { + return manager->allocate_page(env, bytes); + } catch (const std::exception& e) { + jclass exception_class = env->FindClass("java/lang/OutOfMemoryError"); + env->ThrowNew(exception_class, e.what()); + env->DeleteLocalRef(exception_class); + return nullptr; + } +} + +} // namespace + +PaimonJniMemoryManager::PaimonJniMemoryManager(std::unique_ptr impl) + : _impl(std::move(impl)) {} + +PaimonJniMemoryManager::~PaimonJniMemoryManager() = default; + +Status PaimonJniMemoryManager::create(RuntimeState* state, + std::unique_ptr* manager) { + DORIS_CHECK(state != nullptr); + DORIS_CHECK(manager != nullptr); + if (state->query_mem_tracker() == nullptr) { + return Status::InternalError( + "Paimon JNI writer cannot size its write buffer without a query tracker"); + } + if (state->get_query_ctx() == nullptr) { + return Status::InternalError( + "Paimon JNI writer cannot allocate native memory without QueryContext"); + } + + // A query can create multiple local sink instances. Divide its budget + // before applying the configured cap so one writer cannot consume the + // entire query allowance. + const int64_t writer_count = std::max(1, state->num_local_sink()); + const int64_t query_limit = state->query_mem_tracker()->limit(); + const int64_t query_share = query_limit > 0 ? query_limit / writer_count : query_limit; + const int64_t configured_memory_limit = config::paimon_jni_writer_memory_pool_limit_bytes; + const int64_t memory_limit = query_share > 0 ? std::min(query_share, configured_memory_limit) + : configured_memory_limit; + if (memory_limit <= 0) { + return Status::Error( + "Paimon JNI writer has insufficient memory budget: query_limit={}, " + "local_sink_count={}, write_buffer_limit={}", + PrettyPrinter::print_bytes(query_limit), writer_count, + PrettyPrinter::print_bytes(memory_limit)); + } + + // ResourceContext is retained by Impl for the manager's whole lifetime; + // this is what keeps asynchronous JNI callbacks associated with the query. + auto impl = std::make_unique(state->get_query_ctx()->resource_ctx(), memory_limit); + *manager = std::unique_ptr(new PaimonJniMemoryManager(std::move(impl))); + return Status::OK(); +} + +Status PaimonJniMemoryManager::register_natives(JNIEnv* env, jclass writer_class) { + // Keep the JNI surface minimal: Java asks native code only for a page; + // all ownership, limits, and cleanup stay in PaimonJniMemoryManager. + static char allocate_name[] = "allocatePaimonMemoryPage"; + static char allocate_signature[] = "(JI)Ljava/nio/ByteBuffer;"; + static ::JNINativeMethod methods[] = { + {allocate_name, allocate_signature, + reinterpret_cast(&allocate_paimon_memory_page)}, + }; + if (env->RegisterNatives(writer_class, methods, + static_cast(sizeof(methods) / sizeof(methods[0]))) != JNI_OK) { + RETURN_IF_ERROR(Jni::Env::GetJniExceptionMsg( + env, true, "JNI exception registering Paimon memory native methods: ")); + return Status::JniError("Failed to register Paimon memory native methods"); + } + return Status::OK(); +} + +jobject PaimonJniMemoryManager::allocate_page(JNIEnv* env, jint bytes) { + return _impl->allocate_page(env, bytes); +} + +int64_t PaimonJniMemoryManager::memory_limit() const { + return _impl->memory_limit(); +} + +int64_t PaimonJniMemoryManager::native_peak_allocated_bytes() const { + return _impl->native_peak_allocated_bytes(); +} + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.h b/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.h new file mode 100644 index 00000000000000..0d4818cd6dce42 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.h @@ -0,0 +1,81 @@ +// 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. + +#pragma once + +#include + +#include +#include + +#include "common/status.h" + +namespace doris { + +class RuntimeState; + +/// Owns the Doris-side native memory used by one Java Paimon writer. +/// +/// Paimon's sort/merge buffers are Java objects, but their page storage is +/// requested through a JNI callback. This manager is the bridge for that +/// callback: it allocates each page with Doris' allocator, exposes the page as +/// a direct ByteBuffer, tracks it until the writer is closed, and releases all +/// pages in its destructor. The native writer/backend therefore keeps this +/// manager alive for at least as long as the Java writer can access its +/// callback handle. +/// +/// The limit is a per-writer budget. It is derived from the query memory +/// limit and the number of local sink instances, then capped by the global +/// Paimon JNI configuration. The manager accounts only for pages allocated +/// by this callback; Java heap and other Paimon-managed memory remain under +/// their respective runtimes. +class PaimonJniMemoryManager { +public: + ~PaimonJniMemoryManager(); + + /// Construct a manager whose budget is sized from the query context. + /// + /// The query must provide both a memory tracker and QueryContext. The + /// latter supplies the ResourceContext used whenever allocation/freeing + /// crosses into a JNI-created or asynchronous thread. + static Status create(RuntimeState* state, std::unique_ptr* manager); + /// Register the static JNI callback used by PaimonJniWriter. + static Status register_natives(JNIEnv* env, jclass writer_class); + + /// Allocate one native page and return it as a direct ByteBuffer. + /// + /// On failure this method leaves no accounting entry behind and reports + /// the error through the JNI environment. The returned buffer remains + /// valid until the manager is destroyed (or allocation of that page is + /// rolled back because NewDirectByteBuffer failed). + jobject allocate_page(JNIEnv* env, jint bytes); + + /// Return the immutable per-writer native page budget in bytes. + int64_t memory_limit() const; + + /// Return the high-water mark of native pages allocated by this manager. + int64_t native_peak_allocated_bytes() const; + +private: + class Impl; + + explicit PaimonJniMemoryManager(std::unique_ptr impl); + + std::unique_ptr _impl; +}; + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/paimon_table_writer.cpp b/be/src/exec/sink/writer/paimon/paimon_table_writer.cpp new file mode 100644 index 00000000000000..66238cc621760a --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_table_writer.cpp @@ -0,0 +1,193 @@ +// 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 "exec/sink/writer/paimon/jni_paimon_write_backend.h" +#include "runtime/runtime_state.h" + +namespace doris { + +PaimonTableWriter::PaimonTableWriter(TDataSink t_sink, const VExprContextSPtrs& output_exprs, + std::shared_ptr dep, + std::shared_ptr fin_dep) + : AsyncResultWriter(output_exprs, std::move(dep), std::move(fin_dep)), + _t_sink(std::move(t_sink)) { + DCHECK(_t_sink.__isset.paimon_table_sink); +} + +Status PaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) { + _state = state; + _operator_profile = profile; + + // Register profile counters + _written_rows_counter = ADD_COUNTER(_operator_profile, "WrittenRows", TUnit::UNIT); + _written_bytes_counter = ADD_COUNTER(_operator_profile, "WrittenBytes", TUnit::BYTES); + _send_data_timer = ADD_TIMER(_operator_profile, "SendDataTime"); + _project_timer = ADD_CHILD_TIMER(_operator_profile, "ProjectTime", "SendDataTime"); + _file_store_write_timer = + ADD_CHILD_TIMER(_operator_profile, "FileStoreWriteTime", "SendDataTime"); + _open_timer = ADD_TIMER(_operator_profile, "OpenTime"); + _close_timer = ADD_TIMER(_operator_profile, "CloseTime"); + _prepare_commit_timer = ADD_TIMER(_operator_profile, "PrepareCommitTime"); + _commit_payload_count = ADD_COUNTER(_operator_profile, "CommitPayloadCount", TUnit::UNIT); + _commit_payload_bytes_counter = + ADD_COUNTER(_operator_profile, "CommitPayloadBytes", TUnit::BYTES); + + SCOPED_TIMER(_open_timer); + + // Step 1: Create the JNI backend that owns the Java Paimon SDK writer. + _backend = std::make_unique(); + // 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=JNI, 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(_projection_block(block, &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: Convert Block → Arrow IPC → direct buffer → 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 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(); + } + } + } + + // Record message metrics before backend shutdown, but retain local ownership until + // every Java SDK user has stopped successfully. + if (status.ok() && !messages.empty()) { + messages.front().__set_row_count(_written_rows); + COUNTER_UPDATE(_commit_payload_count, static_cast(messages.size())); + for (const auto& msg : messages) { + DORIS_CHECK(msg.__isset.payload); + COUNTER_UPDATE(_commit_payload_bytes_counter, static_cast(msg.payload.size())); + } + } + + // 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()) { + if (status.ok()) { + status = close_st; + } else { + LOG(WARNING) << "Paimon backend close also failed: " << close_st.to_string(); + } + } + } + + // A clean backend close is the ownership boundary. On any failure, the FE must never + // observe these messages; use an independent committer to clean prepared files because + // the original Java writer is already closed (or its close outcome is unsafe). + if (!status.ok() && !messages.empty()) { + WARN_IF_ERROR( + JniPaimonWriteBackend::abort_prepared_commit(_t_sink.paimon_table_sink, messages), + "failed to abort Paimon files after backend close failure"); + } + + _backend.reset(); + + if (!status.ok() || messages.empty()) { + return status; + } + + // Transfer payload ownership only after backend shutdown. If the report budget rejects + // the transfer, abort immediately. If FE later explicitly rejects the final report, the + // callback reads the same RuntimeState payloads and aborts them without retaining a second copy. + Status publish_status = _state->add_paimon_commit_messages(messages); + if (!publish_status.ok()) { + WARN_IF_ERROR( + JniPaimonWriteBackend::abort_prepared_commit(_t_sink.paimon_table_sink, messages), + "failed to abort Paimon files after report-budget rejection"); + return publish_status; + } + + RuntimeState* cleanup_state = _state; + TPaimonTableSink cleanup_sink = _t_sink.paimon_table_sink; + _state->add_rejected_external_file_report_cleanup([cleanup_state, + cleanup_sink = std::move(cleanup_sink)] { + std::vector rejected_messages; + cleanup_state->append_paimon_commit_messages(&rejected_messages); + WARN_IF_ERROR(JniPaimonWriteBackend::abort_prepared_commit(cleanup_sink, rejected_messages), + "failed to abort Paimon files after final report rejection"); + }); + + LOG(INFO) << "Paimon writer closed: " << messages.size() + << " commit messages, total rows=" << _written_rows; + return status; +} + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/paimon_table_writer.h b/be/src/exec/sink/writer/paimon/paimon_table_writer.h new file mode 100644 index 00000000000000..850d5dea4b001f --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_table_writer.h @@ -0,0 +1,102 @@ +// 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. + +#pragma once + +#include + +#include +#include + +#include "common/status.h" +#include "core/block/block.h" +#include "exec/sink/writer/async_result_writer.h" +#include "exec/sink/writer/paimon/paimon_write_backend.h" +#include "exprs/vexpr_fwd.h" +#include "runtime/runtime_profile.h" + +namespace doris { + +class RuntimeState; + +/// Each PaimonTableSinkLocalState owns one PaimonTableWriter, which in turn +/// owns one IPaimonWriteBackend and one IPaimonWriter. Pipeline parallelism +/// therefore determines the number of independent Paimon writer sessions; +/// each writer session delegates partition and bucket routing to the Paimon +/// Java SDK through JNI. FE currently places Paimon writes on GATHER to keep +/// dynamic-bucket assignment single-writer correct. +/// +/// Doris does NOT compute partition values or bucket ids — it passes complete +/// Blocks through the JNI backend to the Paimon SDK, which +/// internally computes partition values, bucket ids, and routes rows to the +/// correct file writers. +/// +/// Architecture: +/// PaimonTableSinkOperatorX +/// │ sink_impl() → AsyncWriterSink::sink() (no routing) +/// ▼ +/// PaimonTableWriter (one per LocalState / pipeline instance) +/// │ owns IPaimonWriteBackend (JNI) +/// │ └─ create_writer() → IPaimonWriter +/// │ write() +/// │ → JNI backend: Block → Arrow IPC → Java Paimon SDK +/// │ → Paimon SDK owns row normalization, routing, buffering, +/// │ file writing, and compaction +/// ▼ +/// close() → prepareCommit() → CommitMessage[] +/// +/// Commit flow (BE only prepares messages; FE is the commit coordinator): +/// close() → writer->prepare_commit() +/// → collect TPaimonCommitMessage[] (DPCM-framed serialized messages) +/// → RuntimeState::add_paimon_commit_messages() +/// → RPC to FE Coordinator → PaimonTransaction +class PaimonTableWriter final : public AsyncResultWriter { +public: + PaimonTableWriter(TDataSink t_sink, const VExprContextSPtrs& output_exprs, + std::shared_ptr dep, std::shared_ptr fin_dep); + + ~PaimonTableWriter() override = default; + + Status open(RuntimeState* state, RuntimeProfile* profile) override; + + Status write(RuntimeState* state, Block& block) override; + + Status close(Status status) override; + +private: + TDataSink _t_sink; + RuntimeState* _state = nullptr; + + // Backend owns the JNI connection and creates the writer adapter. + // Both are scoped to this PaimonTableWriter (one per LocalState). + std::unique_ptr _backend; + std::unique_ptr _writer; + + // Profile counters + RuntimeProfile::Counter* _written_rows_counter = nullptr; + RuntimeProfile::Counter* _written_bytes_counter = nullptr; + RuntimeProfile::Counter* _send_data_timer = nullptr; + RuntimeProfile::Counter* _project_timer = nullptr; + RuntimeProfile::Counter* _file_store_write_timer = nullptr; + RuntimeProfile::Counter* _open_timer = nullptr; + RuntimeProfile::Counter* _close_timer = nullptr; + RuntimeProfile::Counter* _prepare_commit_timer = nullptr; + RuntimeProfile::Counter* _commit_payload_count = nullptr; + RuntimeProfile::Counter* _commit_payload_bytes_counter = nullptr; +}; + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/paimon_write_backend.h b/be/src/exec/sink/writer/paimon/paimon_write_backend.h new file mode 100644 index 00000000000000..ade863aec8cde1 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_write_backend.h @@ -0,0 +1,84 @@ +// 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. + +#pragma once + +#include + +#include +#include + +#include "common/status.h" +#include "core/block/block.h" + +namespace doris { + +class RuntimeState; +class RuntimeProfile; + +/// Writer contract implemented by one SDK writer adapter. Each +/// PaimonTableWriter owns one IPaimonWriter, which delegates to the +/// Paimon Java SDK through JNI. Partition and bucket routing happens +/// inside the SDK. +/// +/// Lifecycle: created by IPaimonWriteBackend::create_writer() after the +/// backend is opened; used for the duration of one pipeline instance. +class IPaimonWriter { +public: + virtual ~IPaimonWriter() = default; + + /// Write a projected Block to the Paimon SDK. + /// For the JNI path: Block → Arrow IPC → direct buffer → Java. + virtual Status write(RuntimeState* state, Block& block) = 0; + + /// Flush all buffered data, close files, and collect serialized commit + /// messages (DPCM-framed). Called once at EOS. + virtual Status prepare_commit(std::vector& messages) = 0; + + /// Discard written data files on error. Called when write or prepare_commit fails. + virtual Status abort() = 0; +}; + +/// Backend boundary for creating writers through the Paimon Java SDK. +/// +/// The backend owns the JVM class reference, method IDs, and Java writer object. +/// +/// Each backend creates one or more IPaimonWriter adapters that share the +/// same underlying connection. Snapshot commit is deliberately excluded from +/// this boundary: BE only prepares commit messages (byte payloads), while FE +/// PaimonTransaction is the single commit coordinator. +class IPaimonWriteBackend { +public: + virtual ~IPaimonWriteBackend() = default; + + /// Initialize the backend connection. For JNI this loads the writer class, + /// creates the Java object, and calls PaimonJniWriter.open(). + virtual Status open(const TPaimonTableSink& sink, RuntimeState* state, + RuntimeProfile* profile) = 0; + + /// Create a lightweight writer adapter that delegates to this backend. + virtual Status create_writer(std::unique_ptr* writer) = 0; + + /// Stop all SDK users and release backend resources. + /// + /// A successful return is the ownership boundary after which native memory + /// backing SDK buffers can be reclaimed safely. Callers must not publish + /// prepared commit messages until this succeeds. + virtual Status close() = 0; +}; + +} // namespace doris diff --git a/be/src/format/jni/jni_data_bridge.cpp b/be/src/format/jni/jni_data_bridge.cpp index 9dc935e0b62da5..51b4d0e7089c01 100644 --- a/be/src/format/jni/jni_data_bridge.cpp +++ b/be/src/format/jni/jni_data_bridge.cpp @@ -29,6 +29,7 @@ #include "core/column/column_string.h" #include "core/column/column_struct.h" #include "core/column/column_varbinary.h" +#include "core/column/variant_v2/column_variant_v2.h" #include "core/data_type/data_type_array.h" #include "core/data_type/data_type_map.h" #include "core/data_type/data_type_nullable.h" @@ -38,6 +39,7 @@ #include "core/data_type/primitive_type.h" #include "core/types.h" #include "core/value/decimalv2_value.h" +#include "core/value/variant/variant_batch_builder.h" #include "util/string_util.h" #include "util/url_coding.h" @@ -148,6 +150,10 @@ Status JniDataBridge::fill_column(TableMetaAddress& address, ColumnPtr& doris_co case PrimitiveType::TYPE_VARBINARY: status = _fill_varbinary_column(address, data_column, num_rows); break; + case PrimitiveType::TYPE_VARIANT: + status = _fill_variant_column(address, data_column, static_cast(null_map_ptr), + num_rows); + break; default: status = Status::InvalidArgument("Unsupported type {} in jni scanner", data_type->get_name()); @@ -179,6 +185,37 @@ Status JniDataBridge::_fill_varbinary_column(TableMetaAddress& address, return Status::OK(); } +Status JniDataBridge::_fill_variant_column(TableMetaAddress& address, + MutableColumnPtr& doris_column, const bool* null_map, + size_t num_rows) { + ColumnPtr values = ColumnVarbinary::create(); + ColumnPtr metadatas = ColumnVarbinary::create(); + const DataTypePtr binary_type = std::make_shared(); + RETURN_IF_ERROR(fill_column(address, values, binary_type, num_rows)); + RETURN_IF_ERROR(fill_column(address, metadatas, binary_type, num_rows)); + + RETURN_IF_CATCH_EXCEPTION({ + const auto& value_column = assert_cast(*values); + const auto& metadata_column = assert_cast(*metadatas); + VariantBatchBuilder builder; + for (size_t row_index = 0; row_index < num_rows; ++row_index) { + auto row = builder.begin_row(); + if (null_map[row_index]) { + row.add_null(); + } else { + const StringRef value = value_column.get_data_at(row_index); + const StringRef metadata = metadata_column.get_data_at(row_index); + row.add_value({.metadata = {.data = metadata.data, .size = metadata.size}, + .value = value}); + } + row.finish(); + } + VariantBatchBuilder batch = builder.finish_batch(); + assert_cast(*doris_column).insert_encoded_batch(batch); + }); + return Status::OK(); +} + Status JniDataBridge::_fill_string_column(TableMetaAddress& address, MutableColumnPtr& doris_column, size_t num_rows) { auto& string_col = static_cast(*doris_column); @@ -358,6 +395,8 @@ std::string JniDataBridge::get_jni_type(const DataTypePtr& data_type) { } case TYPE_VARBINARY: return "varbinary"; + case TYPE_VARIANT: + return "struct"; // bitmap, hll, quantile_state, jsonb are transferred as strings via JNI case TYPE_BITMAP: [[fallthrough]]; @@ -433,6 +472,8 @@ std::string JniDataBridge::get_jni_type_with_different_string(const DataTypePtr& << assert_cast(remove_nullable(data_type).get())->len() << ")"; return buffer.str(); + case TYPE_VARIANT: + return "struct"; case TYPE_DECIMALV2: { buffer << "decimalv2(" << DecimalV2Value::PRECISION << "," << DecimalV2Value::SCALE << ")"; return buffer.str(); @@ -506,6 +547,8 @@ std::string JniDataBridge::encode_schema_values(const std::vector& std::string JniDataBridge::get_jni_type_with_encoded_struct_fields(const DataTypePtr& data_type) { switch (data_type->get_primitive_type()) { + case TYPE_VARIANT: + return "struct<$dmFsdWU=:varbinary,$bWV0YWRhdGE=:varbinary>"; case TYPE_STRUCT: { const auto* type_struct = assert_cast(remove_nullable(data_type).get()); diff --git a/be/src/format/jni/jni_data_bridge.h b/be/src/format/jni/jni_data_bridge.h index e037ffec3d4d5a..5a0eea8aa55b26 100644 --- a/be/src/format/jni/jni_data_bridge.h +++ b/be/src/format/jni/jni_data_bridge.h @@ -154,6 +154,9 @@ class JniDataBridge { static Status _fill_varbinary_column(TableMetaAddress& address, MutableColumnPtr& doris_column, size_t num_rows); + static Status _fill_variant_column(TableMetaAddress& address, MutableColumnPtr& doris_column, + const bool* null_map, size_t num_rows); + static Status _fill_array_column(TableMetaAddress& address, MutableColumnPtr& doris_column, const DataTypePtr& data_type, size_t num_rows); diff --git a/be/src/runtime/runtime_state.cpp b/be/src/runtime/runtime_state.cpp index 380bc8f8f72081..c02c7478a5a471 100644 --- a/be/src/runtime/runtime_state.cpp +++ b/be/src/runtime/runtime_state.cpp @@ -72,18 +72,52 @@ Status RuntimeState::add_iceberg_commit_datas(TIcebergCommitData iceberg_commit_ thrift_limit > report_envelope_headroom ? thrift_limit - report_envelope_headroom : 0; std::lock_guard budget_lock(_external_file_report_state->mutex); // Parallel task states share this budget because FE receives their vectors in one fragment report. - if (_external_file_report_state->iceberg_serialized_bytes + serialized_size + sizeof(uint32_t) > + if (_external_file_report_state->serialized_commit_bytes + serialized_size + sizeof(uint32_t) > commit_data_limit) { return Status::InternalError( "Iceberg commit metadata exceeds the Thrift report limit; reduce output file " "count"); } std::lock_guard data_lock(_iceberg_commit_datas_mutex); - _external_file_report_state->iceberg_serialized_bytes += serialized_size + sizeof(uint32_t); + _external_file_report_state->serialized_commit_bytes += serialized_size + sizeof(uint32_t); _iceberg_commit_datas.emplace_back(std::move(iceberg_commit_data)); return Status::OK(); } +Status RuntimeState::add_paimon_commit_messages(std::vector commit_messages) { + if (commit_messages.empty()) { + return Status::OK(); + } + + ThriftSerializer serializer(false, 256); + size_t messages_size = 0; + for (auto& message : commit_messages) { + uint32_t serialized_size = 0; + uint8_t* buffer = nullptr; + RETURN_IF_ERROR(serializer.serialize(&message, &serialized_size, &buffer)); + messages_size += serialized_size + sizeof(uint32_t); + } + + constexpr size_t report_envelope_headroom = 1024 * 1024; + const size_t thrift_limit = coordinator_thrift_message_limit(); + const size_t commit_data_limit = + thrift_limit > report_envelope_headroom ? thrift_limit - report_envelope_headroom : 0; + std::lock_guard budget_lock(_external_file_report_state->mutex); + if (messages_size > + commit_data_limit - + std::min(commit_data_limit, _external_file_report_state->serialized_commit_bytes)) { + return Status::InternalError( + "Paimon commit metadata exceeds the Thrift report limit; reduce output file " + "count"); + } + std::lock_guard data_lock(_paimon_commit_messages_mutex); + _external_file_report_state->serialized_commit_bytes += messages_size; + _paimon_commit_messages.insert(_paimon_commit_messages.end(), + std::make_move_iterator(commit_messages.begin()), + std::make_move_iterator(commit_messages.end())); + return Status::OK(); +} + size_t RuntimeState::coordinator_thrift_message_limit() const { int32_t effective_thrift_limit = std::max(config::thrift_max_message_size, 0); if (_query_options.__isset.coordinator_thrift_max_message_size && @@ -115,6 +149,10 @@ void RuntimeState::append_external_file_commit_data(TReportExecStatusParams* par params->mc_commit_datas.insert(params->mc_commit_datas.end(), commit_datas.begin(), commit_datas.end()); } + append_paimon_commit_messages(¶ms->paimon_commit_messages); + if (!params->paimon_commit_messages.empty()) { + params->__isset.paimon_commit_messages = true; + } } void RuntimeState::add_rejected_external_file_report_cleanup(std::function cleanup) { diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h index a3cfc5e4cad782..44e054a593070f 100644 --- a/be/src/runtime/runtime_state.h +++ b/be/src/runtime/runtime_state.h @@ -84,7 +84,7 @@ class ExternalFileReportState { private: std::mutex mutex; - size_t iceberg_serialized_bytes = 0; + size_t serialized_commit_bytes = 0; bool ownership_may_have_transferred = false; std::vector> rejected_report_cleanups; }; @@ -547,6 +547,14 @@ class RuntimeState { Status add_iceberg_commit_datas(TIcebergCommitData iceberg_commit_data); + void append_paimon_commit_messages(std::vector* output) const { + std::lock_guard lock(_paimon_commit_messages_mutex); + output->insert(output->end(), _paimon_commit_messages.begin(), + _paimon_commit_messages.end()); + } + + Status add_paimon_commit_messages(std::vector commit_messages); + size_t coordinator_thrift_message_limit() const; void append_external_file_commit_data(TReportExecStatusParams* params, bool final_report) const; @@ -1012,6 +1020,9 @@ class RuntimeState { mutable std::mutex _mc_commit_datas_mutex; std::vector _mc_commit_datas; + mutable std::mutex _paimon_commit_messages_mutex; + std::vector _paimon_commit_messages; + std::vector> _op_id_to_local_state; std::unique_ptr _sink_local_state; diff --git a/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp b/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp index 9f5e6054a1fba7..da30c0a040c1fc 100644 --- a/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp +++ b/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp @@ -15,7 +15,9 @@ // specific language governing permissions and limitations // under the License. +#include #include +#include #include #include @@ -251,6 +253,34 @@ TEST(DataTypeVariantV2SerdeOutputTest, DormantDirectClassExists) { EXPECT_EQ(serde.get_name(), "Variant"); } +TEST(DataTypeVariantV2SerdeOutputTest, VariantStructArrowPreservesBinaryEncoding) { + DataTypeVariantV2SerDe serde; + auto documents = encoded_json({R"({"id":7,"tags":["doris"]})", R"([1,true,null])"}); + auto value_builder = std::make_shared(); + auto metadata_builder = std::make_shared(); + auto arrow_type = arrow::struct_({arrow::field("value", arrow::binary(), false), + arrow::field("metadata", arrow::binary(), false)}); + arrow::StructBuilder builder(arrow_type, arrow::default_memory_pool(), + {value_builder, metadata_builder}); + NullMap forced_nulls {0, 1}; + + ASSERT_TRUE(serde.write_column_to_arrow(*documents, &forced_nulls, &builder, 0, + documents->size(), cctz::utc_time_zone()) + .ok()); + std::shared_ptr output; + ASSERT_TRUE(builder.Finish(&output).ok()); + ASSERT_EQ(output->length(), 2); + EXPECT_FALSE(output->IsNull(0)); + EXPECT_TRUE(output->IsNull(1)); + + const auto& values = assert_cast(*output->field(0)); + const auto& metadatas = assert_cast(*output->field(1)); + const VariantRef expected = documents->get_value_ref(0); + EXPECT_EQ(values.GetView(0), std::string_view(expected.value.data, expected.value.size)); + EXPECT_EQ(metadatas.GetView(0), + std::string_view(expected.metadata.data, expected.metadata.size)); +} + TEST(DataTypeVariantV2SerdeOutputTest, SqlScalarsFollowLegacyOutputAndDataFormatsUseJson) { DataTypeVariantV2SerDe serde; auto strings = typed_strings({std::string_view("a\"\n"), std::string_view(""), std::nullopt, diff --git a/be/test/format/table/paimon_jni_reader_test.cpp b/be/test/format/table/paimon_jni_reader_test.cpp index 969713429c29d6..c2e5eb4224145a 100644 --- a/be/test/format/table/paimon_jni_reader_test.cpp +++ b/be/test/format/table/paimon_jni_reader_test.cpp @@ -19,10 +19,21 @@ #include +#include +#include #include +#include #include #include +#include "core/assert_cast.h" +#include "core/column/column_nullable.h" +#include "core/column/variant_v2/column_variant_v2.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_variant_v2.h" +#include "core/value/variant/variant_batch_builder.h" +#include "exprs/function/parse/variant_string_parse.h" +#include "format/jni/jni_data_bridge.h" #include "gen_cpp/PlanNodes_types.h" #include "runtime/runtime_state.h" @@ -40,6 +51,13 @@ TFileRangeDesc make_legacy_paimon_jni_range() { return range; } +struct JavaVarbinaryEntry { + int64_t length; + uint64_t address; +}; + +static_assert(sizeof(JavaVarbinaryEntry) == 16); + TEST(LegacyPaimonJniReaderTest, GeneratesMissingOrEmptySerializedTableCacheKey) { const auto range = make_legacy_paimon_jni_range(); TFileScanRangeParams scan_params; @@ -61,5 +79,52 @@ TEST(LegacyPaimonJniReaderTest, GeneratesMissingOrEmptySerializedTableCacheKey) EXPECT_NE(missing_key, empty_key); } +TEST(LegacyPaimonJniReaderTest, PublishesVariantV2BinaryStructSchema) { + const DataTypePtr type = std::make_shared(); + EXPECT_EQ(JniDataBridge::get_jni_type_with_different_string(type), + "struct"); + EXPECT_EQ(JniDataBridge::get_jni_type_with_encoded_struct_fields(type), + "struct<$dmFsdWU=:varbinary,$bWV0YWRhdGE=:varbinary>"); +} + +TEST(LegacyPaimonJniReaderTest, DecodesVariantV2BinaryStructFromJavaMetadata) { + const std::string json = R"({"id":7,"tags":["doris"]})"; + JsonStringToVariantEncoder encoder; + encoder.add_json({json.data(), json.size()}); + VariantBatchBuilder source = encoder.finish_batch(); + const VariantRef expected = source.value_at(0); + + std::array outer_nulls {0, 1}; + std::array child_nulls {0, 1}; + std::array value_entries { + JavaVarbinaryEntry {static_cast(expected.value.size), + reinterpret_cast(expected.value.data)}, + JavaVarbinaryEntry {0, 0}}; + std::array metadata_entries { + JavaVarbinaryEntry {static_cast(expected.metadata.size), + reinterpret_cast(expected.metadata.data)}, + JavaVarbinaryEntry {0, 0}}; + std::array metadata {reinterpret_cast(outer_nulls.data()), + reinterpret_cast(child_nulls.data()), + reinterpret_cast(value_entries.data()), + reinterpret_cast(child_nulls.data()), + reinterpret_cast(metadata_entries.data())}; + + const DataTypePtr type = make_nullable(std::make_shared()); + ColumnPtr result = type->create_column(); + JniDataBridge::TableMetaAddress address(reinterpret_cast(metadata.data())); + ASSERT_TRUE(JniDataBridge::fill_column(address, result, type, 2).ok()); + + const auto& nullable = assert_cast(*result); + EXPECT_EQ(nullable.get_null_map_data(), (NullMap {0, 1})); + const auto& variants = assert_cast(nullable.get_nested_column()); + ASSERT_EQ(variants.size(), 2); + const VariantRef actual = variants.get_value_ref(0); + EXPECT_EQ(actual.value, expected.value); + EXPECT_EQ(std::string_view(actual.metadata.data, actual.metadata.size), + std::string_view(expected.metadata.data, expected.metadata.size)); + EXPECT_TRUE(variants.get_value_ref(1).is_null()); +} + } // namespace } // namespace doris diff --git a/be/test/runtime/runtime_state_block_budget_test.cpp b/be/test/runtime/runtime_state_block_budget_test.cpp index 5a384378ec382b..cb55639b6cd120 100644 --- a/be/test/runtime/runtime_state_block_budget_test.cpp +++ b/be/test/runtime/runtime_state_block_budget_test.cpp @@ -60,6 +60,27 @@ TEST(RuntimeStateIcebergCommitDataTest, SharesTheReportBudgetAcrossParallelTasks EXPECT_FALSE(second_status.ok()); } +TEST(RuntimeStateIcebergCommitDataTest, SharesTheReportBudgetWithPaimon) { + RuntimeState iceberg_state; + RuntimeState paimon_state; + auto budget = std::make_shared(); + iceberg_state.set_external_file_report_state(budget); + paimon_state.set_external_file_report_state(budget); + const int32_t saved_limit = config::thrift_max_message_size; + config::thrift_max_message_size = 1024 * 1024 + 512; + TIcebergCommitData iceberg_data; + iceberg_data.__set_file_path(std::string(300, 'x')); + TPaimonCommitMessage paimon_data; + paimon_data.__set_payload(std::string(300, 'x')); + + Status iceberg_status = iceberg_state.add_iceberg_commit_datas(iceberg_data); + Status paimon_status = paimon_state.add_paimon_commit_messages({std::move(paimon_data)}); + + config::thrift_max_message_size = saved_limit; + EXPECT_TRUE(iceberg_status.ok()) << iceberg_status; + EXPECT_FALSE(paimon_status.ok()); +} + TEST(RuntimeStateIcebergCommitDataTest, UsesTheSmallerCoordinatorThriftLimit) { RuntimeState state; const int32_t saved_limit = config::thrift_max_message_size; @@ -91,6 +112,9 @@ TEST(RuntimeStateIcebergCommitDataTest, PeriodicReportOmitsExternalCommitData) { ASSERT_TRUE(state.add_iceberg_commit_datas(iceberg_data).ok()); TMCCommitData mc_data; state.add_mc_commit_datas(mc_data); + TPaimonCommitMessage paimon_data; + paimon_data.__set_payload("paimon-commit"); + ASSERT_TRUE(state.add_paimon_commit_messages({std::move(paimon_data)}).ok()); TReportExecStatusParams periodic_params; state.append_external_file_commit_data(&periodic_params, false); @@ -98,12 +122,14 @@ TEST(RuntimeStateIcebergCommitDataTest, PeriodicReportOmitsExternalCommitData) { EXPECT_FALSE(periodic_params.__isset.hive_partition_updates); EXPECT_FALSE(periodic_params.__isset.iceberg_commit_datas); EXPECT_FALSE(periodic_params.__isset.mc_commit_datas); + EXPECT_FALSE(periodic_params.__isset.paimon_commit_messages); TReportExecStatusParams final_params; state.append_external_file_commit_data(&final_params, true); EXPECT_TRUE(final_params.__isset.hive_partition_updates); EXPECT_TRUE(final_params.__isset.iceberg_commit_datas); EXPECT_TRUE(final_params.__isset.mc_commit_datas); + EXPECT_TRUE(final_params.__isset.paimon_commit_messages); } TEST(RuntimeStateIcebergCommitDataTest, RetainsFileCleanupUntilReportAcknowledgement) { diff --git a/fe/be-java-extensions/paimon-scanner/pom.xml b/fe/be-java-extensions/paimon-scanner/pom.xml index fa7c27e4e98319..7f415dae329855 100644 --- a/fe/be-java-extensions/paimon-scanner/pom.xml +++ b/fe/be-java-extensions/paimon-scanner/pom.xml @@ -61,6 +61,11 @@ under the License. paimon-format + + org.apache.arrow + arrow-vector + + + org.apache.paimon paimon-format - test + + org.apache.paimon + paimon-vortex-format + +