diff --git a/be/src/format/jni/jni_data_bridge.cpp b/be/src/format/jni/jni_data_bridge.cpp index 1a88f4c317b28a..9dc935e0b62da5 100644 --- a/be/src/format/jni/jni_data_bridge.cpp +++ b/be/src/format/jni/jni_data_bridge.cpp @@ -38,6 +38,8 @@ #include "core/data_type/primitive_type.h" #include "core/types.h" #include "core/value/decimalv2_value.h" +#include "util/string_util.h" +#include "util/url_coding.h" namespace doris { @@ -490,6 +492,55 @@ std::string JniDataBridge::get_jni_type_with_different_string(const DataTypePtr& } } +std::string JniDataBridge::encode_schema_values(const std::vector& values) { + std::vector encoded_values; + encoded_values.reserve(values.size()); + for (const auto& value : values) { + std::string encoded; + base64_encode(value, &encoded); + // Prefix every element so an empty Base64 token remains distinct from an empty list. + encoded_values.emplace_back("$" + encoded); + } + return join(encoded_values, ","); +} + +std::string JniDataBridge::get_jni_type_with_encoded_struct_fields(const DataTypePtr& data_type) { + switch (data_type->get_primitive_type()) { + case TYPE_STRUCT: { + const auto* type_struct = + assert_cast(remove_nullable(data_type).get()); + std::ostringstream buffer; + buffer << "struct<"; + for (int i = 0; i < type_struct->get_elements().size(); ++i) { + if (i != 0) { + buffer << ","; + } + std::string encoded_name; + base64_encode(type_struct->get_element_name(i), &encoded_name); + // '$' versions the nested-name token and cannot collide with Base64 or grammar + // delimiters; Java rejects an unversioned name in the encoded schema path. + buffer << "$" << encoded_name << ":" + << get_jni_type_with_encoded_struct_fields(type_struct->get_element(i)); + } + buffer << ">"; + return buffer.str(); + } + case TYPE_ARRAY: { + const auto* type_array = + assert_cast(remove_nullable(data_type).get()); + return "array<" + get_jni_type_with_encoded_struct_fields(type_array->get_nested_type()) + + ">"; + } + case TYPE_MAP: { + const auto* type_map = assert_cast(remove_nullable(data_type).get()); + return "map<" + get_jni_type_with_encoded_struct_fields(type_map->get_key_type()) + "," + + get_jni_type_with_encoded_struct_fields(type_map->get_value_type()) + ">"; + } + default: + return get_jni_type_with_different_string(data_type); + } +} + Status JniDataBridge::_fill_column_meta(const ColumnPtr& doris_column, const DataTypePtr& data_type, std::vector& meta_data) { auto logical_type = data_type->get_primitive_type(); diff --git a/be/src/format/jni/jni_data_bridge.h b/be/src/format/jni/jni_data_bridge.h index d3deca8a85076d..e037ffec3d4d5a 100644 --- a/be/src/format/jni/jni_data_bridge.h +++ b/be/src/format/jni/jni_data_bridge.h @@ -140,6 +140,12 @@ class JniDataBridge { */ static std::string get_jni_type_with_different_string(const DataTypePtr& data_type); + /** Encodes every list element independently so schema delimiters remain unambiguous. */ + static std::string encode_schema_values(const std::vector& values); + + /** Encodes STRUCT field names inside a JNI type descriptor. */ + static std::string get_jni_type_with_encoded_struct_fields(const DataTypePtr& data_type); + private: // Column fill helpers for various types static Status _fill_string_column(TableMetaAddress& address, MutableColumnPtr& doris_column, diff --git a/be/src/format/table/paimon_jni_reader.cpp b/be/src/format/table/paimon_jni_reader.cpp index e03e28186a29f8..15f81fe6fcab79 100644 --- a/be/src/format/table/paimon_jni_reader.cpp +++ b/be/src/format/table/paimon_jni_reader.cpp @@ -54,10 +54,14 @@ PaimonJniReader::PaimonJniReader(const std::vector& file_slot_d [&]() { std::vector column_names; std::vector column_types; + std::vector encoded_column_types; for (const auto& desc : file_slot_descs) { column_names.emplace_back(desc->col_name()); column_types.emplace_back( JniDataBridge::get_jni_type_with_different_string(desc->type())); + encoded_column_types.emplace_back( + JniDataBridge::get_jni_type_with_encoded_struct_fields( + desc->type())); } const auto& paimon_params = range.table_format_params.paimon_params; std::map params; @@ -70,6 +74,12 @@ PaimonJniReader::PaimonJniReader(const std::vector& file_slot_d } params["required_fields"] = join(column_names, ","); params["columns_types"] = join(column_types, "#"); + // V1 and V2 must publish the same safe schema protocol because session + // routing can select either producer for the same Paimon scanner. + params["required_fields_base64"] = + JniDataBridge::encode_schema_values(column_names); + params["columns_types_base64"] = + JniDataBridge::encode_schema_values(encoded_column_types); params["time_zone"] = state->timezone(); if (range_params->__isset.serialized_table) { params["serialized_table"] = range_params->serialized_table; diff --git a/be/src/format_v2/jni/jni_table_reader.cpp b/be/src/format_v2/jni/jni_table_reader.cpp index 1c691fc3129c95..b58696fe4ad895 100644 --- a/be/src/format_v2/jni/jni_table_reader.cpp +++ b/be/src/format_v2/jni/jni_table_reader.cpp @@ -496,11 +496,16 @@ Status JniTableReader::_set_open_scanner_batch_size(size_t batch_size) { } void JniTableReader::_prepare_jni_scanner_schema() { + const bool publish_encoded_schema = publishes_encoded_schema(); std::vector required_fields; std::vector column_types; + std::vector encoded_column_types; std::vector replace_types; required_fields.reserve(_jni_columns.size()); column_types.reserve(_jni_columns.size()); + if (publish_encoded_schema) { + encoded_column_types.reserve(_jni_columns.size()); + } replace_types.reserve(_jni_columns.size()); _jni_block_template.clear(); _jni_block_template.reserve(_jni_columns.size()); @@ -511,6 +516,10 @@ void JniTableReader::_prepare_jni_scanner_schema() { required_fields.push_back(column.java_name); column_types.push_back( JniDataBridge::get_jni_type_with_different_string(column.transfer_type)); + if (publish_encoded_schema) { + encoded_column_types.push_back( + JniDataBridge::get_jni_type_with_encoded_struct_fields(column.transfer_type)); + } replace_types.push_back(column.replace_type); has_replace_type = has_replace_type || column.replace_type != "not_replace"; _jni_block_template.insert( @@ -518,6 +527,14 @@ void JniTableReader::_prepare_jni_scanner_schema() { } _scanner_params["required_fields"] = join(required_fields, ","); _scanner_params["columns_types"] = join(column_types, "#"); + if (publish_encoded_schema) { + // Only Paimon consumes the paired payload. Keeping it capability-gated avoids recursively + // encoding nested types for every split of unrelated V2 JNI connectors. + _scanner_params["required_fields_base64"] = + JniDataBridge::encode_schema_values(required_fields); + _scanner_params["columns_types_base64"] = + JniDataBridge::encode_schema_values(encoded_column_types); + } if (has_replace_type) { _scanner_params["replace_string"] = join(replace_types, ","); } diff --git a/be/src/format_v2/jni/jni_table_reader.h b/be/src/format_v2/jni/jni_table_reader.h index 5e48270515f996..df7edc92f985fb 100644 --- a/be/src/format_v2/jni/jni_table_reader.h +++ b/be/src/format_v2/jni/jni_table_reader.h @@ -81,6 +81,7 @@ class JniTableReader : public TableReader { virtual Status _close_jni_scanner(); virtual Status _set_open_scanner_batch_size(size_t batch_size); virtual bool supports_batch_size_update_after_open() const { return true; } + virtual bool publishes_encoded_schema() const { return false; } virtual Status _open_jni_scanner(); bool _reserve_split_profile_publication(); const std::vector& jni_columns() const { return _jni_columns; } diff --git a/be/src/format_v2/jni/paimon_jni_reader.h b/be/src/format_v2/jni/paimon_jni_reader.h index 26767458b427db..04dd3f0be99337 100644 --- a/be/src/format_v2/jni/paimon_jni_reader.h +++ b/be/src/format_v2/jni/paimon_jni_reader.h @@ -51,6 +51,7 @@ class PaimonJniReader final : public format::JniTableReader { Status validate_scan_range(const TFileRangeDesc& range) const override; Status build_scanner_params(std::map* params) const override; bool supports_batch_size_update_after_open() const override { return false; } + bool publishes_encoded_schema() const override { return true; } private: static std::string build_default_io_manager_tmp_dirs(const std::vector& store_paths); diff --git a/be/test/format_v2/jni/jni_table_reader_test.cpp b/be/test/format_v2/jni/jni_table_reader_test.cpp index 95eda4a557e790..2eee5548ad7e31 100644 --- a/be/test/format_v2/jni/jni_table_reader_test.cpp +++ b/be/test/format_v2/jni/jni_table_reader_test.cpp @@ -27,6 +27,9 @@ #include #include +#include "core/data_type/data_type_string.h" +#include "core/data_type/data_type_struct.h" +#include "format/jni/jni_data_bridge.h" #include "io/io_common.h" namespace doris::format { @@ -109,6 +112,40 @@ Status init_reader(FakeJniTableReader* reader, const std::shared_ptr( + DataTypes {std::make_shared(), std::make_shared(), + std::make_shared()}, + Strings {"hash#name", "region,code", "colon:name"}); + + // Keep standard Base64 padding in the wire-contract expectation; Java decodes these tokens + // verbatim, and field names whose length is not divisible by three require trailing '=' bytes. + EXPECT_EQ(JniDataBridge::get_jni_type_with_encoded_struct_fields(type), + "struct<$aGFzaCNuYW1l:string,$cmVnaW9uLGNvZGU=:string,$Y29sb246bmFtZQ==:string>"); +} + +TEST(JniTableReaderTest, GenericConnectorDoesNotPublishPaimonEncodedSchema) { + FakeJniTableReader reader; + reader._jni_columns = {JniTableReader::JniColumn { + .java_name = "region,code", + // Keep the aggregate complete because BE UT treats omitted JNI column fields as errors. + .output_type = std::make_shared(), + .transfer_type = std::make_shared(), + }}; + + reader._prepare_jni_scanner_schema(); + + EXPECT_FALSE(reader._scanner_params.contains("required_fields_base64")); + EXPECT_FALSE(reader._scanner_params.contains("columns_types_base64")); +} + TEST(JniTableReaderTest, CancellationStopsBeforeFetchingAnotherJavaBatch) { auto io_ctx = std::make_shared(); FakeJniTableReader reader; diff --git a/be/test/format_v2/jni/paimon_jni_reader_test.cpp b/be/test/format_v2/jni/paimon_jni_reader_test.cpp index 9c46aba05c76e1..921b6e70ab909f 100644 --- a/be/test/format_v2/jni/paimon_jni_reader_test.cpp +++ b/be/test/format_v2/jni/paimon_jni_reader_test.cpp @@ -23,6 +23,7 @@ #include #include +#include "core/data_type/data_type_string.h" #include "format_v2/table_reader.h" #include "gen_cpp/PlanNodes_types.h" #include "runtime/runtime_state.h" @@ -161,6 +162,21 @@ TEST(PaimonJniReaderTest, ScanLevelOptionsOverrideLegacySplitFallbacks) { EXPECT_EQ(params["hadoop.source"], "scan"); } +TEST(PaimonJniReaderTest, PublishesEncodedSchemaForQuotedIdentifiers) { + PaimonJniReader reader; + reader._jni_columns = {JniTableReader::JniColumn { + .java_name = "region,code", + // Keep the aggregate complete because BE UT treats omitted JNI column fields as errors. + .output_type = std::make_shared(), + .transfer_type = std::make_shared(), + }}; + + reader._prepare_jni_scanner_schema(); + + EXPECT_EQ(reader._scanner_params.at("required_fields_base64"), "$cmVnaW9uLGNvZGU="); + EXPECT_TRUE(reader._scanner_params.contains("columns_types_base64")); +} + TEST(PaimonJniReaderTest, UsesStableRuntimeBatchSizeBeforeAndAfterOpen) { TQueryOptions query_options; query_options.__set_batch_size(8160); diff --git a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ColumnType.java b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ColumnType.java index dfd40b1541e877..983681d24dcb72 100644 --- a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ColumnType.java +++ b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ColumnType.java @@ -17,8 +17,10 @@ package org.apache.doris.common.jni.vec; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; +import java.util.Base64; import java.util.Collections; import java.util.List; import java.util.regex.Matcher; @@ -277,6 +279,15 @@ private static int findNextNestedField(String commaSplitFields) { } public static ColumnType parseType(String columnName, String hiveType) { + return parseType(columnName, hiveType, false); + } + + public static ColumnType parseTypeWithEncodedStructFields(String columnName, String hiveType) { + return parseType(columnName, hiveType, true); + } + + private static ColumnType parseType(String columnName, String hiveType, boolean encodedStructFields) { + String originalType = hiveType.trim(); String lowerCaseType = hiveType.toLowerCase(); Type type = Type.UNSUPPORTED; int length = -1; @@ -397,7 +408,7 @@ public static ColumnType parseType(String columnName, String hiveType) { if (lowerCaseType.indexOf("<") == 5 && lowerCaseType.lastIndexOf(">") == lowerCaseType.length() - 1) { ColumnType nestedType = parseType("element", - lowerCaseType.substring(6, lowerCaseType.length() - 1)); + originalType.substring(6, originalType.length() - 1), encodedStructFields); ColumnType arrayType = new ColumnType(columnName, Type.ARRAY); arrayType.setChildTypes(Collections.singletonList(nestedType)); return arrayType; @@ -405,12 +416,13 @@ public static ColumnType parseType(String columnName, String hiveType) { } else if (lowerCaseType.startsWith("map")) { if (lowerCaseType.indexOf("<") == 3 && lowerCaseType.lastIndexOf(">") == lowerCaseType.length() - 1) { - String keyValue = lowerCaseType.substring(4, lowerCaseType.length() - 1); + String keyValue = originalType.substring(4, originalType.length() - 1); int index = findNextNestedField(keyValue); if (index != keyValue.length() && index != 0) { - ColumnType keyType = parseType("key", keyValue.substring(0, index).trim()); + ColumnType keyType = parseType( + "key", keyValue.substring(0, index).trim(), encodedStructFields); ColumnType valueType = - parseType("value", keyValue.substring(index + 1).trim()); + parseType("value", keyValue.substring(index + 1).trim(), encodedStructFields); ColumnType mapType = new ColumnType(columnName, Type.MAP); mapType.setChildTypes(Arrays.asList(keyType, valueType)); return mapType; @@ -419,16 +431,31 @@ public static ColumnType parseType(String columnName, String hiveType) { } else if (lowerCaseType.startsWith("struct")) { if (lowerCaseType.indexOf("<") == 6 && lowerCaseType.lastIndexOf(">") == lowerCaseType.length() - 1) { - String listFields = lowerCaseType.substring(7, lowerCaseType.length() - 1); + String listFields = originalType.substring(7, originalType.length() - 1); ArrayList fields = new ArrayList<>(); ArrayList names = new ArrayList<>(); while (listFields.length() > 0) { int index = findNextNestedField(listFields); int pivot = listFields.indexOf(':'); if (pivot > 0 && pivot < listFields.length() - 1) { - fields.add(parseType(listFields.substring(0, pivot), - listFields.substring(pivot + 1, index))); - names.add(listFields.substring(0, pivot)); + String fieldName = listFields.substring(0, pivot); + if (encodedStructFields) { + if (!fieldName.startsWith("$")) { + throw new IllegalArgumentException( + "Encoded JNI struct field is missing its version marker"); + } + // The encoded producer removes every grammar delimiter from + // names before parsing, then restores the exact UTF-8 spelling. + fieldName = new String(Base64.getDecoder().decode(fieldName.substring(1)), + StandardCharsets.UTF_8); + } else { + // The public legacy grammar historically normalized STRUCT keys + // to lowercase; only the versioned Paimon payload preserves spelling. + fieldName = fieldName.toLowerCase(); + } + fields.add(parseType(fieldName, + listFields.substring(pivot + 1, index), encodedStructFields)); + names.add(fieldName); listFields = listFields.substring(Math.min(index + 1, listFields.length())); } else { break; diff --git a/fe/be-java-extensions/java-common/src/test/java/org/apache/doris/common/jni/JniScannerTest.java b/fe/be-java-extensions/java-common/src/test/java/org/apache/doris/common/jni/JniScannerTest.java index 74683955411801..9e9b15fc510c38 100644 --- a/fe/be-java-extensions/java-common/src/test/java/org/apache/doris/common/jni/JniScannerTest.java +++ b/fe/be-java-extensions/java-common/src/test/java/org/apache/doris/common/jni/JniScannerTest.java @@ -19,15 +19,27 @@ import org.apache.doris.common.jni.utils.OffHeap; +import org.apache.doris.common.jni.vec.ColumnType; import org.apache.doris.common.jni.vec.VectorTable; import org.junit.Assert; import org.junit.Test; import java.io.IOException; +import java.util.Arrays; import java.util.HashMap; public class JniScannerTest { + @Test + public void testUnencodedStructFieldNamesRemainLowerCase() { + ColumnType structType = ColumnType.parseType( + "value", "struct>"); + + Assert.assertEquals(Arrays.asList("mixed", "upper"), structType.getChildNames()); + Assert.assertEquals(Arrays.asList("nested"), + structType.getChildTypes().get(1).getChildNames()); + } + @Test public void testMockJniScanner() throws IOException { OffHeap.setTesting(); 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 3279aa3df746af..59cc8d13abc3d1 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 @@ -46,16 +46,21 @@ import java.lang.management.ThreadMXBean; import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Paths; import java.util.Arrays; +import java.util.Base64; import java.util.HashMap; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Optional; import java.util.TimeZone; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.IntSupplier; +import java.util.function.LongSupplier; import java.util.stream.Collectors; public class PaimonJniScanner extends JniScanner { @@ -68,6 +73,11 @@ public class PaimonJniScanner extends JniScanner { static final String JNI_IO_MANAGER_TMP_DIR = "paimon.jni.io_manager.tmp_dir"; static final String JNI_IO_MANAGER_IMPL_CLASS = "paimon.jni.io_manager.impl_class"; private static final AtomicInteger ACTIVE_SCANNERS = new AtomicInteger(); + // Scanner profiles are collected per split, but the asynchronous reader pool is JVM-global. + // Share one sample so profile collection does not allocate ThreadInfo for every scanner. + private static final CachedThreadCounter ASYNC_READER_THREAD_COUNTER = new CachedThreadCounter( + TimeUnit.SECONDS.toNanos(1L), System::nanoTime, + () -> countThreadsByNamePrefix(ASYNC_READER_THREAD_NAME_PREFIX)); private final Map params; private final Map hadoopOptionParams; @@ -92,18 +102,23 @@ public class PaimonJniScanner extends JniScanner { public PaimonJniScanner(int batchSize, Map params) { this.classLoader = this.getClass().getClassLoader(); - if (LOG.isDebugEnabled()) { - LOG.debug("params:{}", params); - } this.params = params; - String[] requiredFields = splitParam(params.get("required_fields"), ","); - String[] requiredTypes = splitParam(params.get("columns_types"), "#"); + boolean encodedSchema = usesEncodedSchema(params); + String[] requiredFields = requiredFields(params); + String[] requiredTypes = requiredTypes(params); Preconditions.checkArgument(requiredFields.length == requiredTypes.length, "required_fields size %s does not match columns_types size %s", requiredFields.length, requiredTypes.length); ColumnType[] columnTypes = new ColumnType[requiredTypes.length]; for (int i = 0; i < requiredTypes.length; i++) { - columnTypes[i] = ColumnType.parseType(requiredFields[i], requiredTypes[i]); + columnTypes[i] = encodedSchema + ? ColumnType.parseTypeWithEncodedStructFields(requiredFields[i], requiredTypes[i]) + : ColumnType.parseType(requiredFields[i], requiredTypes[i]); + } + if (LOG.isDebugEnabled()) { + // The raw map may carry storage or JDBC credentials; diagnostics must only use + // values derived from a fixed non-sensitive whitelist. + LOG.debug(buildDebugSummary(batchSize, requiredFields.length)); } paimonSplit = params.get("paimon_split"); paimonPredicate = params.get("paimon_predicate"); @@ -483,7 +498,56 @@ private static long nonNegative(long value) { } private static int currentAsyncReaderThreadCount() { - return countThreadsByNamePrefix(ASYNC_READER_THREAD_NAME_PREFIX); + return ASYNC_READER_THREAD_COUNTER.get(); + } + + static String buildDebugSummary(int batchSize, int requiredFieldCount) { + return "Paimon JNI scanner configuration: batchSize=" + batchSize + + ", requiredFieldCount=" + requiredFieldCount; + } + + static String[] requiredFields(Map params) { + String encodedFields = params.get("required_fields_base64"); + if (encodedFields == null) { + return splitParam(params.get("required_fields"), ","); + } + if (encodedFields.isEmpty()) { + return new String[0]; + } + // Each identifier is encoded independently, so delimiters in quoted identifiers cannot + // change field cardinality. The legacy parameter remains the rolling-upgrade fallback. + return decodeSchemaValues(encodedFields); + } + + static String[] requiredTypes(Map params) { + String encodedTypes = params.get("columns_types_base64"); + return encodedTypes == null + ? splitParam(params.get("columns_types"), "#") + : decodeSchemaValues(encodedTypes); + } + + private static boolean usesEncodedSchema(Map params) { + boolean hasFields = params.containsKey("required_fields_base64"); + boolean hasTypes = params.containsKey("columns_types_base64"); + // Both halves describe one schema version; accepting a mixed pair would reintroduce the + // cardinality and nested-name ambiguity this protocol is intended to remove. + Preconditions.checkArgument(hasFields == hasTypes, + "required_fields_base64 and columns_types_base64 must be provided together"); + return hasFields; + } + + private static String[] decodeSchemaValues(String encodedValues) { + if (encodedValues.isEmpty()) { + return new String[0]; + } + return Arrays.stream(encodedValues.split(",", -1)) + .map(encoded -> { + // A marker on every token preserves list arity when the encoded value itself is empty. + Preconditions.checkArgument(encoded.startsWith("$"), + "Encoded JNI schema token is missing its version marker"); + return new String(Base64.getDecoder().decode(encoded.substring(1)), StandardCharsets.UTF_8); + }) + .toArray(String[]::new); } static int countThreadsByNamePrefix(String threadNamePrefix) { @@ -498,6 +562,36 @@ static int countThreadsByNamePrefix(String threadNamePrefix) { return count; } + static final class CachedThreadCounter { + private final long ttlNanos; + private final LongSupplier ticker; + private final IntSupplier sampler; + private volatile long lastSampleNanos = Long.MIN_VALUE; + private volatile int cachedValue; + + CachedThreadCounter(long ttlNanos, LongSupplier ticker, IntSupplier sampler) { + this.ttlNanos = ttlNanos; + this.ticker = ticker; + this.sampler = sampler; + } + + int get() { + long now = ticker.getAsLong(); + long sampledAt = lastSampleNanos; + if (sampledAt != Long.MIN_VALUE && now - sampledAt < ttlNanos) { + return cachedValue; + } + synchronized (this) { + now = ticker.getAsLong(); + if (lastSampleNanos == Long.MIN_VALUE || now - lastSampleNanos >= ttlNanos) { + cachedValue = sampler.getAsInt(); + lastSampleNanos = now; + } + return cachedValue; + } + } + } + private void markScannerOpenedForMetrics() { if (!scannerCounted) { scannerCounted = true; 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 d43f251f58deb1..e04bb4375675e1 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 @@ -17,6 +17,15 @@ package org.apache.doris.paimon; +import org.apache.doris.common.jni.vec.ColumnType; + +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Configurator; +import org.apache.logging.log4j.core.config.Property; import org.apache.paimon.data.InternalRow; import org.apache.paimon.disk.BufferFileReader; import org.apache.paimon.disk.BufferFileWriter; @@ -34,14 +43,18 @@ import java.io.IOException; import java.lang.reflect.Field; import java.lang.reflect.Proxy; +import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.Base64; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; public class PaimonJniScannerTest { @Rule @@ -52,6 +65,104 @@ public void testConstructorAcceptsEmptyProjection() { new PaimonJniScanner(128, createBaseParams()); } + @Test + public void testConstructorDecodesDelimiterSafeProjectionNames() { + Map params = createBaseParams(); + params.put("required_fields", "legacy,value,is,ambiguous"); + params.put("required_fields_base64", encodeFields( + "region,code", "hash#name", "地区 名")); + params.put("columns_types", "string#string#string"); + params.put("columns_types_base64", encodeFields("string", "string", "string")); + + PaimonJniScanner scanner = new PaimonJniScanner(128, params); + + Assert.assertArrayEquals(new String[] {"region,code", "hash#name", "地区 名"}, + PaimonJniScanner.requiredFields(params)); + Assert.assertEquals("3", scanner.getStatistics().get("gauge:PaimonJniRequiredFieldCount")); + Assert.assertEquals(0, PaimonJniScanner.getFieldIndex( + Arrays.asList("region,code", "hash#name", "地区 名"), "REGION,CODE")); + } + + @Test + public void testConstructorDecodesDelimiterSafeNestedFieldNamesAndTypes() { + Map params = createBaseParams(); + params.put("required_fields_base64", encodeFields("nested#value")); + params.put("columns_types", "legacy#type#framing"); + params.put("columns_types_base64", encodeFields( + "struct<$aGFzaCNuYW1l:string,$cmVnaW9uLGNvZGU:string,$Y29sb246bmFtZQ:string>")); + + InspectablePaimonJniScanner scanner = new InspectablePaimonJniScanner(128, params); + + Assert.assertEquals(1, scanner.requiredTypes().length); + ColumnType structType = scanner.requiredTypes()[0]; + Assert.assertTrue(structType.isStruct()); + Assert.assertEquals(Arrays.asList("hash#name", "region,code", "colon:name"), + structType.getChildNames()); + } + + @Test + public void testConstructorDistinguishesEmptyIdentifierFromEmptyProjection() { + Map oneEmptyField = createBaseParams(); + oneEmptyField.put("required_fields_base64", "$"); + oneEmptyField.put("columns_types_base64", "$c3RyaW5n"); + + new PaimonJniScanner(128, oneEmptyField); + + Assert.assertArrayEquals(new String[] {""}, PaimonJniScanner.requiredFields(oneEmptyField)); + Map noFields = createBaseParams(); + noFields.put("required_fields_base64", ""); + noFields.put("columns_types_base64", ""); + Assert.assertEquals(0, PaimonJniScanner.requiredFields(noFields).length); + new PaimonJniScanner(128, noFields); + } + + @Test + public void testDebugSummaryNeverIncludesRawScannerParams() { + Map params = createBaseParams(); + params.put("hadoop.fs.s3a.secret.key", "FAKE_SECRET_MARKER"); + params.put("paimon.jdbc.url", "jdbc:test?password=FAKE_PASSWORD_MARKER"); + + Logger logger = (Logger) LogManager.getLogger(PaimonJniScanner.class); + Level originalLevel = logger.getLevel(); + CapturingAppender appender = new CapturingAppender(); + appender.start(); + logger.addAppender(appender); + Configurator.setLevel(PaimonJniScanner.class.getName(), Level.DEBUG); + try { + new PaimonJniScanner(128, params); + } finally { + logger.removeAppender(appender); + appender.stop(); + Configurator.setLevel(PaimonJniScanner.class.getName(), originalLevel); + } + + String captured = String.join("\n", appender.messages); + Assert.assertTrue(captured.contains("batchSize=128, requiredFieldCount=0")); + Assert.assertFalse(captured.contains("FAKE_SECRET_MARKER")); + Assert.assertFalse(captured.contains("FAKE_PASSWORD_MARKER")); + Assert.assertFalse(captured.contains("hadoop.fs.s3a.secret.key")); + Assert.assertFalse(captured.contains("paimon.jdbc.url")); + } + + @Test + public void testThreadCountSampleIsSharedWithinTtl() { + AtomicLong ticker = new AtomicLong(1000L); + AtomicInteger sampleCalls = new AtomicInteger(); + PaimonJniScanner.CachedThreadCounter counter = new PaimonJniScanner.CachedThreadCounter( + 100L, ticker::get, () -> { + sampleCalls.incrementAndGet(); + return 7; + }); + + Assert.assertEquals(7, counter.get()); + Assert.assertEquals(7, counter.get()); + Assert.assertEquals(1, sampleCalls.get()); + + ticker.addAndGet(100L); + Assert.assertEquals(7, counter.get()); + Assert.assertEquals(2, sampleCalls.get()); + } + @Test public void testIOManagerOptionHelpers() throws Exception { Map params = createBaseParams(); @@ -264,6 +375,12 @@ private Map createBaseParams() { return params; } + private String encodeFields(String... fields) { + return Arrays.stream(fields) + .map(field -> "$" + Base64.getEncoder().encodeToString(field.getBytes(StandardCharsets.UTF_8))) + .collect(java.util.stream.Collectors.joining(",")); + } + private void setTableOptions(PaimonJniScanner scanner, Map options) throws Exception { Table table = (Table) Proxy.newProxyInstance( Table.class.getClassLoader(), new Class[] {Table.class}, (proxy, method, args) -> { @@ -337,6 +454,29 @@ public void close() { } } + private static class CapturingAppender extends AbstractAppender { + private final CopyOnWriteArrayList messages = new CopyOnWriteArrayList<>(); + + CapturingAppender() { + super("paimon-test-capture", null, null, false, Property.EMPTY_ARRAY); + } + + @Override + public void append(LogEvent event) { + messages.add(event.getMessage().getFormattedMessage()); + } + } + + private static class InspectablePaimonJniScanner extends PaimonJniScanner { + InspectablePaimonJniScanner(int batchSize, Map params) { + super(batchSize, params); + } + + ColumnType[] requiredTypes() { + return types; + } + } + @Test public void testGetFieldIndexMatchesMixedCaseColumns() { Assert.assertEquals(1, PaimonJniScanner.getFieldIndex(Arrays.asList("data", "mIxEd_COL", "PART"), diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java index 83e778ac2f0ced..96411f0730c122 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java @@ -652,16 +652,31 @@ public void replayAlterCatalogProps(CatalogLog log, Map oldPrope CatalogIf catalog = idToCatalog.get(log.getCatalogId()); if (catalog instanceof ExternalCatalog) { Map newProps = log.getNewProps(); - ((ExternalCatalog) catalog).tryModifyCatalogProps(newProps); if (!isReplay) { + boolean tentativelyMutated = false; try { - ((ExternalCatalog) catalog).checkProperties(); - } catch (DdlException ddlException) { - if (oldProperties != null) { + ExternalCatalog externalCatalog = (ExternalCatalog) catalog; + boolean validatedWithoutMutation = externalCatalog.validatePropertiesBeforeUpdate( + oldProperties, newProps); + if (!validatedWithoutMutation) { + externalCatalog.tryModifyCatalogProps(newProps); + tentativelyMutated = true; + externalCatalog.checkProperties(); + } + } catch (Exception validationException) { + // Only legacy validators publish a tentative candidate. Detached validators + // leave the live CatalogProperty untouched while concurrent initialization runs. + if (oldProperties != null && tentativelyMutated) { ((ExternalCatalog) catalog).rollBackCatalogProps(oldProperties); } - throw ddlException; + if (validationException instanceof DdlException) { + throw (DdlException) validationException; + } + throw new DdlException("Invalid catalog properties: " + + validationException.getMessage(), validationException); } + } else { + ((ExternalCatalog) catalog).tryModifyCatalogProps(newProps); } if (newProps.containsKey(METADATA_REFRESH_INTERVAL_SEC)) { long catalogId = catalog.getId(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java index 71b7a7f5344714..ce855994b766bd 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java @@ -470,8 +470,12 @@ private void buildMetaCache() { // check if all required properties are set when creating catalog public void checkProperties() throws DdlException { + checkProperties(catalogProperty); + } + + protected void checkProperties(CatalogProperty property) throws DdlException { // check refresh parameter of catalog - Map properties = catalogProperty.getProperties(); + Map properties = property.getProperties(); if (properties.containsKey(CatalogMgr.METADATA_REFRESH_INTERVAL_SEC)) { try { int metadataRefreshIntervalSec = Integer.parseInt( @@ -485,7 +489,7 @@ public void checkProperties() throws DdlException { } // check schema.cache.ttl-second parameter - String schemaCacheTtlSecond = catalogProperty.getOrDefault(SCHEMA_CACHE_TTL_SECOND, null); + String schemaCacheTtlSecond = property.getOrDefault(SCHEMA_CACHE_TTL_SECOND, null); if (java.util.Objects.nonNull(schemaCacheTtlSecond) && NumberUtils.toInt(schemaCacheTtlSecond, CACHE_NO_TTL) < CACHE_TTL_DISABLE_CACHE) { throw new DdlException( @@ -493,6 +497,16 @@ public void checkProperties() throws DdlException { } } + /** + * Validate an ALTER candidate without publishing it to this catalog. A true return value + * declares that the connector performed complete detached validation; false retains the + * legacy tentative-mutation path for connectors that have not implemented this contract. + */ + public boolean validatePropertiesBeforeUpdate( + Map currentProperties, Map updatedProperties) throws DdlException { + return false; + } + /** * eg: * ( diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonPartitionInfoLoader.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonPartitionInfoLoader.java index c29a359b9592d1..9053aae1920841 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonPartitionInfoLoader.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonPartitionInfoLoader.java @@ -22,22 +22,21 @@ import org.apache.doris.datasource.CacheException; import org.apache.doris.datasource.NameMapping; import org.apache.doris.datasource.paimon.PaimonPartitionInfo; +import org.apache.doris.datasource.paimon.PaimonReaderOptions; import org.apache.doris.datasource.paimon.PaimonUtil; import org.apache.commons.collections4.CollectionUtils; +import org.apache.paimon.catalog.CatalogUtils; import org.apache.paimon.partition.Partition; import org.apache.paimon.table.Table; import java.util.List; /** - * Loads partition info for a snapshot projection from the base Paimon table and catalog metadata. + * Loads partition info for a snapshot projection from the effective Paimon table handle. */ public final class PaimonPartitionInfoLoader { - private final PaimonTableLoader tableLoader; - - public PaimonPartitionInfoLoader(PaimonTableLoader tableLoader) { - this.tableLoader = tableLoader; + public PaimonPartitionInfoLoader() { } public PaimonPartitionInfo load(NameMapping nameMapping, Table paimonTable, List partitionColumns) @@ -46,7 +45,10 @@ public PaimonPartitionInfo load(NameMapping nameMapping, Table paimonTable, List return PaimonPartitionInfo.EMPTY; } try { - List paimonPartitions = tableLoader.catalog(nameMapping).getPaimonPartitions(nameMapping); + // Catalog.listPartitions reloads the raw physical table and loses Doris catalog/relation + // copies. Enumerate the already merged handle so validation and planning see one table. + PaimonReaderOptions.validateEffectivePlanningTable(paimonTable); + List paimonPartitions = CatalogUtils.listPartitionsFromFileSystem(paimonTable); boolean legacyPartitionName = PaimonUtil.isLegacyPartitionName(paimonTable); return PaimonUtil.generatePartitionInfo(partitionColumns, paimonPartitions, legacyPartitionName); } catch (Exception e) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java index 3409df1e42c079..365826afefb0d3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java @@ -36,6 +36,7 @@ import org.apache.paimon.table.Table; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; @@ -136,8 +137,9 @@ public Table getPaimonTable(NameMapping nameMapping, String branch, String query } return executionAuthenticator.execute(() -> { Table table = catalog.getTable(identifier); - Map tableOptions = - paimonProperties.getTableOptionsForCopy(table.options()); + Map tableOptions = paimonProperties.getTableOptionsForCopy(); + // Relation options are applied after this cached handle is returned. Defer final + // validation so a safe relation value can override an unsafe physical value. return tableOptions.isEmpty() ? table : table.copy(tableOptions); }); } catch (Exception e) { @@ -164,14 +166,32 @@ public Map getPaimonOptionsMap() { @Override public void checkProperties() throws DdlException { - super.checkProperties(); - CacheSpec.checkBooleanProperty(catalogProperty.getOrDefault(PAIMON_TABLE_CACHE_ENABLE, null), + checkProperties(catalogProperty, catalogProperty.getProperties()); + } + + @Override + public boolean validatePropertiesBeforeUpdate( + Map currentProperties, Map updatedProperties) throws DdlException { + Map candidateProperties = currentProperties == null + ? new HashMap<>() : new HashMap<>(currentProperties); + candidateProperties.putAll(updatedProperties); + checkProperties(new CatalogProperty(null, candidateProperties), updatedProperties); + return true; + } + + private void checkProperties(CatalogProperty property, Map strictlyValidatedProperties) + throws DdlException { + super.checkProperties(property); + CacheSpec.checkBooleanProperty(property.getOrDefault(PAIMON_TABLE_CACHE_ENABLE, null), PAIMON_TABLE_CACHE_ENABLE); - CacheSpec.checkLongProperty(catalogProperty.getOrDefault(PAIMON_TABLE_CACHE_TTL_SECOND, null), + CacheSpec.checkLongProperty(property.getOrDefault(PAIMON_TABLE_CACHE_TTL_SECOND, null), -1L, PAIMON_TABLE_CACHE_TTL_SECOND); - CacheSpec.checkLongProperty(catalogProperty.getOrDefault(PAIMON_TABLE_CACHE_CAPACITY, null), + CacheSpec.checkLongProperty(property.getOrDefault(PAIMON_TABLE_CACHE_CAPACITY, null), 0L, PAIMON_TABLE_CACHE_CAPACITY); - catalogProperty.checkMetaStoreAndStorageProperties(AbstractPaimonProperties.class); + // Validate only newly supplied dynamic options on ALTER. This lets an old image containing + // a formerly accepted option survive an unrelated update while still rejecting new writes. + PaimonReaderOptions.validateCatalogProperties(strictlyValidatedProperties); + property.checkMetaStoreAndStorageProperties(AbstractPaimonProperties.class); } @Override diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java index 1d08ba1274e256..979c61abdc56a4 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java @@ -66,7 +66,7 @@ public PaimonExternalMetaCache(ExecutorService refreshExecutor) { super(ENGINE, refreshExecutor); tableLoader = new PaimonTableLoader(); latestSnapshotProjectionLoader = new PaimonLatestSnapshotProjectionLoader( - new PaimonPartitionInfoLoader(tableLoader), this::getPaimonSchemaCacheValue); + new PaimonPartitionInfoLoader(), this::getPaimonSchemaCacheValue); tableEntry = registerEntry(MetaCacheEntryDef.of(ENTRY_TABLE, NameMapping.class, PaimonTableCacheValue.class, this::loadTableCacheValue, defaultEntryCacheSpec(), MetaCacheEntryInvalidation.forNameMapping(nameMapping -> nameMapping))); @@ -89,6 +89,10 @@ public PaimonSnapshotCacheValue getSnapshotCache(ExternalTable dorisTable) { return tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue(); } + public PaimonSnapshotCacheValue loadSnapshotProjection(ExternalTable dorisTable, Table effectiveTable) { + return latestSnapshotProjectionLoader.load(dorisTable.getOrBuildNameMapping(), effectiveTable); + } + public PaimonSchemaCacheValue getPaimonSchemaCacheValue(NameMapping nameMapping, long schemaId) { SchemaCacheValue schemaCacheValue = schemaEntry.get(nameMapping.getCtlId()) .get(new PaimonSchemaCacheKey(nameMapping, schemaId)); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java index 6aeacc9058d115..96dff680734b98 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java @@ -178,6 +178,19 @@ private PaimonSnapshotCacheValue getPaimonSnapshotCacheValue(Optional resolvedOptions = scanParams.get().getOrResolveMapParams( + options -> PaimonScanParams.resolveOptions(baseTable, options)); + Table effectiveTable = PaimonScanParams.applyOptions(baseTable, resolvedOptions); + if (PaimonScanParams.hasOnlyReaderOptions(resolvedOptions)) { + // Reader tuning cannot change snapshot metadata. Reuse the memoized projection so + // a per-query batch-size change does not enumerate every partition again. + return PaimonUtils.getLatestSnapshotCacheValue(this); + } + // The shared latest cache was built from the catalog-scoped handle. Relation options + // need their own projection so partition enumeration uses the final safe table copy. + return PaimonUtils.loadSnapshotProjection(this, effectiveTable); } else if (scanParams.isPresent() && scanParams.get().isBranch()) { try { Table baseTable = getBasePaimonTable(); @@ -236,7 +249,11 @@ public BaseAnalysisTask createAnalysisTask(AnalysisInfo info) { public long fetchRowCount() { makeSureInitialized(); long rowCount = 0; - List splits = getBasePaimonTable().newReadBuilder().newScan().plan().splits(); + Table effectiveTable = getBasePaimonTable(); + // Statistics and row-count cache planning run before ScanNode and must not reach an + // unsafe manifest executor, even when the foreground relation later supplies an override. + PaimonReaderOptions.validateEffectiveTable(effectiveTable); + List splits = effectiveTable.newReadBuilder().newScan().plan().splits(); for (Split split : splits) { rowCount += split.rowCount(); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonReaderOptions.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonReaderOptions.java new file mode 100644 index 00000000000000..da38441b74a0b1 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonReaderOptions.java @@ -0,0 +1,210 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.datasource.paimon; + +import com.google.common.collect.ImmutableSet; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.options.ConfigOption; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +import org.apache.paimon.table.DelegatedFileStoreTable; +import org.apache.paimon.table.FallbackReadFileStoreTable; +import org.apache.paimon.table.Table; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +/** Validation shared by catalog-scoped and relation-scoped Paimon reader tuning. */ +public final class PaimonReaderOptions { + public static final String TABLE_OPTION_PREFIX = "paimon.table-option."; + public static final int MIN_READ_BATCH_SIZE = 1; + public static final int MAX_READ_BATCH_SIZE = 65536; + public static final long MIN_ASYNC_THRESHOLD_BYTES = 1024L * 1024L; + public static final long MAX_ASYNC_THRESHOLD_BYTES = 1024L * 1024L * 1024L; + + // Keep this list to batch-read controls consumed by Doris' Paimon scan path. Context selectors, + // streaming-source settings, storage layout, and write options are unsafe after schema binding. + private static final Set SUPPORTED_OPTIONS = ImmutableSet.of( + CoreOptions.READ_BATCH_SIZE.key(), + CoreOptions.FILE_READER_ASYNC_THRESHOLD.key(), + CoreOptions.FILE_INDEX_READ_ENABLED.key(), + CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), + CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(), + CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), + CoreOptions.SCAN_PLAN_SORT_PARTITION.key()); + + // These settings do not alter the selected snapshot or manifest projection, so relation-local + // copies can reuse the memoized partition projection while still planning splits from the copy. + private static final Set METADATA_NEUTRAL_OPTIONS = ImmutableSet.of( + CoreOptions.READ_BATCH_SIZE.key(), + CoreOptions.FILE_READER_ASYNC_THRESHOLD.key(), + CoreOptions.FILE_INDEX_READ_ENABLED.key(), + CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), + CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key()); + + private PaimonReaderOptions() { + } + + public static Set supportedOptions() { + return SUPPORTED_OPTIONS; + } + + public static Set metadataNeutralOptions() { + return METADATA_NEUTRAL_OPTIONS; + } + + public static void validate(String key, String value) { + if (!SUPPORTED_OPTIONS.contains(key)) { + throw new IllegalArgumentException("Unsupported Paimon dynamic reader option '" + key + + "'. Supported options are " + SUPPORTED_OPTIONS); + } + + if (CoreOptions.READ_BATCH_SIZE.key().equals(key)) { + int batchSize = parse(key, value, CoreOptions.READ_BATCH_SIZE); + // A zero batch can make Paimon's vectorized reader report success without + // advancing input; the upper bound also prevents one relation from over-allocating. + requireRange(key, batchSize, MIN_READ_BATCH_SIZE, MAX_READ_BATCH_SIZE); + } else if (CoreOptions.FILE_READER_ASYNC_THRESHOLD.key().equals(key)) { + MemorySize threshold = parse(key, value, CoreOptions.FILE_READER_ASYNC_THRESHOLD); + // Bound the trigger on both sides so a query cannot fan out tiny async reads or + // silently disable asynchronous reading with an effectively infinite threshold. + requireRange(key, threshold.getBytes(), + MIN_ASYNC_THRESHOLD_BYTES, MAX_ASYNC_THRESHOLD_BYTES); + } else if (CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key().equals(key)) { + MemorySize targetSize = parse(key, value, CoreOptions.SOURCE_SPLIT_TARGET_SIZE); + // A split-size option represents byte capacity; non-positive values silently defeat + // Paimon's bin packing and turn every data file into a separate Doris scan range. + requireRange(key, targetSize.getBytes(), 1, Long.MAX_VALUE); + } else if (CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key().equals(key)) { + parse(key, value, CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST); + } else if (CoreOptions.SCAN_MANIFEST_PARALLELISM.key().equals(key)) { + validateManifestParallelism(value); + } else if (CoreOptions.FILE_INDEX_READ_ENABLED.key().equals(key)) { + parse(key, value, CoreOptions.FILE_INDEX_READ_ENABLED); + } else { + parse(key, value, CoreOptions.SCAN_PLAN_SORT_PARTITION); + } + } + + public static void validateCatalogProperties(Map properties) { + properties.forEach((key, value) -> { + if (!key.toLowerCase(Locale.ROOT).startsWith(TABLE_OPTION_PREFIX)) { + return; + } + String optionKey = key.substring(TABLE_OPTION_PREFIX.length()); + if (optionKey.isEmpty()) { + throw new IllegalArgumentException( + "Paimon table option name must not be empty after prefix " + TABLE_OPTION_PREFIX); + } + validate(optionKey, value); + }); + } + + public static Map compatibleCatalogOptions(Map properties) { + Map compatibleOptions = new LinkedHashMap<>(); + properties.forEach((key, value) -> { + if (!key.toLowerCase(Locale.ROOT).startsWith(TABLE_OPTION_PREFIX)) { + return; + } + String optionKey = key.substring(TABLE_OPTION_PREFIX.length()); + try { + validate(optionKey, value); + compatibleOptions.put(optionKey, value); + } catch (IllegalArgumentException ignored) { + // Images written before the reader-only allowlist may contain arbitrary Paimon + // options. Keep the catalog loadable, but never apply an unsafe legacy option. + } + }); + return Collections.unmodifiableMap(compatibleOptions); + } + + public static void validateEffectiveTableOptions(Map options) { + SUPPORTED_OPTIONS.stream() + .filter(options::containsKey) + .forEach(key -> validate(key, options.get(key))); + } + + public static void validateEffectiveTable(Table table) { + validateEffectiveTableOptions(table.options()); + if (table instanceof FallbackReadFileStoreTable) { + // The fallback scan plans its private child independently, so the visible main options + // cannot prove that every manifest executor input is safe. + validateEffectiveTable(((FallbackReadFileStoreTable) table).fallback()); + } + if (table instanceof DelegatedFileStoreTable) { + // Privilege and other supported delegates can hide a fallback planner behind their + // own visible main options, so traverse the complete planning-handle chain. + validateEffectiveTable(((DelegatedFileStoreTable) table).wrapped()); + } + } + + public static void validateEffectivePlanningTable(Table table) { + // Partition projection never opens a data reader. Revalidating batch/async values here + // would reject a raw handle even when the final relation copy safely overrides them. + validateIfPresent(table.options(), CoreOptions.SCAN_MANIFEST_PARALLELISM.key()); + validateIfPresent(table.options(), CoreOptions.SCAN_PLAN_SORT_PARTITION.key()); + if (table instanceof FallbackReadFileStoreTable) { + validateEffectivePlanningTable(((FallbackReadFileStoreTable) table).fallback()); + } + if (table instanceof DelegatedFileStoreTable) { + validateEffectivePlanningTable(((DelegatedFileStoreTable) table).wrapped()); + } + } + + private static void validateIfPresent(Map options, String key) { + if (options.containsKey(key)) { + validate(key, options.get(key)); + } + } + + private static void validateManifestParallelism(String value) { + if (value == null) { + return; + } + int parallelism; + try { + parallelism = Integer.parseInt(value); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Invalid value for Paimon option '" + + CoreOptions.SCAN_MANIFEST_PARALLELISM.key() + "': " + value, e); + } + int maximum = Runtime.getRuntime().availableProcessors(); + // Paimon replaces a JVM-static manifest executor above its CPU-sized default, so every + // option source must stay within that capacity to avoid cross-query thread-pool mutation. + requireRange(CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), parallelism, 1, maximum); + } + + private static T parse(String key, String value, ConfigOption option) { + try { + return new Options(Collections.singletonMap(key, value)).get(option); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException("Invalid value for Paimon option '" + key + "': " + + e.getMessage(), e); + } + } + + private static void requireRange(String key, long value, long minimum, long maximum) { + if (value < minimum || value > maximum) { + throw new IllegalArgumentException("Paimon option '" + key + "' must be between " + + minimum + " and " + maximum + ", but was " + value); + } + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java index 5152f63775cc95..b2279023e3e726 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java @@ -45,18 +45,20 @@ public final class PaimonScanParams { "doris.internal.paimon.file-creation-time-millis"; private static final String PINNED_EMPTY_SCAN = "doris.internal.paimon.empty-scan"; - private static final Set QUERY_OPTION_KEYS = ImmutableSet.of( - CoreOptions.SCAN_MODE.key(), - CoreOptions.SCAN_TIMESTAMP.key(), - CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), - CoreOptions.SCAN_WATERMARK.key(), - CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(), - CoreOptions.SCAN_CREATION_TIME_MILLIS.key(), - CoreOptions.SCAN_SNAPSHOT_ID.key(), - CoreOptions.SCAN_TAG_NAME.key(), - CoreOptions.SCAN_VERSION.key(), - CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), - CoreOptions.SCAN_PLAN_SORT_PARTITION.key()); + private static final Set QUERY_OPTION_KEYS = ImmutableSet.builder() + .add(CoreOptions.SCAN_MODE.key()) + .add(CoreOptions.SCAN_TIMESTAMP.key()) + .add(CoreOptions.SCAN_TIMESTAMP_MILLIS.key()) + .add(CoreOptions.SCAN_WATERMARK.key()) + .add(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key()) + .add(CoreOptions.SCAN_CREATION_TIME_MILLIS.key()) + .add(CoreOptions.SCAN_SNAPSHOT_ID.key()) + .add(CoreOptions.SCAN_TAG_NAME.key()) + .add(CoreOptions.SCAN_VERSION.key()) + .add(CoreOptions.SCAN_MANIFEST_PARALLELISM.key()) + .add(CoreOptions.SCAN_PLAN_SORT_PARTITION.key()) + .addAll(PaimonReaderOptions.supportedOptions()) + .build(); private static final Set STARTUP_POSITION_KEYS = ImmutableSet.of( CoreOptions.SCAN_TIMESTAMP.key(), @@ -101,6 +103,8 @@ public static void validateOptions(Map options) { throw new IllegalArgumentException("Unsupported Paimon query option(s): " + unsupported); } + PaimonReaderOptions.validateEffectiveTableOptions(options); + String scanMode = options.get(CoreOptions.SCAN_MODE.key()); if ("from-creation-timestamp".equalsIgnoreCase(scanMode) && options.get(CoreOptions.SCAN_CREATION_TIME_MILLIS.key()) == null) { @@ -140,7 +144,11 @@ public static Table applyOptions(Table table, Map options) { .filter(key -> !tableOptions.containsKey(key)) .forEach(key -> isolatedOptions.put(key, null)); } - return table.copy(isolatedOptions); + Table effectiveTable = table.copy(isolatedOptions); + // Validate after every copy so relation options participate in the documented + // relation > catalog > physical precedence before the effective value is judged. + PaimonReaderOptions.validateEffectiveTable(effectiveTable); + return effectiveTable; } private static Set inheritedReadStateKeys() { @@ -177,6 +185,12 @@ public static boolean selectsSchema(Map options) { return hasStartupOptions(options); } + public static boolean hasOnlyReaderOptions(Map options) { + Map tableOptions = userOptions(options); + return !tableOptions.isEmpty() + && PaimonReaderOptions.metadataNeutralOptions().containsAll(tableOptions.keySet()); + } + public static boolean usesStatementSnapshot(Map options) { if (options.keySet().stream().anyMatch(STARTUP_POSITION_KEYS::contains)) { return false; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonSysExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonSysExternalTable.java index 25297de6862e26..8c6fa89807fec6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonSysExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonSysExternalTable.java @@ -36,8 +36,10 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.apache.paimon.table.DataTable; +import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.Table; import org.apache.paimon.table.source.Split; +import org.apache.paimon.table.system.SystemTableLoader; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataTypeRoot; @@ -70,6 +72,7 @@ public class PaimonSysExternalTable extends ExternalTable { private final PaimonExternalTable sourceTable; private final String sysTableType; private volatile Boolean isDataTable; + private volatile FileStoreTable paimonSysDataTable; private volatile Table paimonSysTable; private volatile List fullSchema; private volatile SchemaCacheValue schemaCacheValue; @@ -117,15 +120,24 @@ private static long generateSysTableId(long sourceTableId, String sysTableType) * Note: system tables currently ignore snapshot semantics. */ public Table getSysPaimonTable() { + validateEffectiveDataTable(null); + return getRawSysPaimonTable(); + } + + /** Returns the cached wrapper without validating its hidden data table. */ + public Table getRawSysPaimonTable() { if (paimonSysTable == null) { synchronized (this) { if (paimonSysTable == null) { - PaimonExternalCatalog catalog = (PaimonExternalCatalog) getCatalog(); - paimonSysTable = catalog.getPaimonTable( - sourceTable.getOrBuildNameMapping(), - "main", // branch - sysTableType // queryType: snapshots, binlog, etc. - ); + Table dataTable = sourceTable.getBasePaimonTable(); + if (!(dataTable instanceof FileStoreTable)) { + throw new IllegalArgumentException( + "Paimon system tables require a file-store data table."); + } + paimonSysDataTable = (FileStoreTable) dataTable; + // Build the wrapper from this exact cached handle. Reloading it through the + // catalog can pair validation with one generation and planning with another. + paimonSysTable = createSystemTable(paimonSysDataTable); LOG.info("Created Paimon system table: {} for source table: {}", sysTableType, sourceTable.getName()); } @@ -135,19 +147,47 @@ public Table getSysPaimonTable() { } public Table getSysPaimonTable(TableScanParams scanParams) { - Table table = getSysPaimonTable(); if (scanParams == null || !scanParams.isOptions()) { - return table; + return getSysPaimonTable(); } - Map resolvedOptions = scanParams.getOrResolveMapParams( - options -> PaimonScanParams.resolveOptions(sourceTable.getBasePaimonTable(), options)); + Map resolvedOptions = resolvedOptions(scanParams); if (PaimonScanParams.getPinnedFileCreationTime(resolvedOptions).isPresent()) { // Generic system-table wrappers cannot carry Paimon's manifest-entry predicate. // Reject the fallback instead of silently widening it to the whole pinned snapshot. throw new IllegalArgumentException( "Paimon system tables cannot apply a creation-time file filter."); } - return PaimonScanParams.applyOptions(table, resolvedOptions); + FileStoreTable effectiveDataTable = (FileStoreTable) PaimonScanParams.applyOptions( + getRawSysPaimonDataTable(), resolvedOptions); + return createSystemTable(effectiveDataTable); + } + + public void validateEffectiveDataTable(TableScanParams scanParams) { + Table dataTable = getRawSysPaimonDataTable(); + if (scanParams != null && scanParams.isOptions()) { + // Apply the same relation copy to the data table hidden by ReadonlyTable wrappers. + PaimonScanParams.applyOptions(dataTable, resolvedOptions(scanParams)); + } else { + PaimonReaderOptions.validateEffectiveTable(dataTable); + } + } + + private Map resolvedOptions(TableScanParams scanParams) { + return scanParams.getOrResolveMapParams( + options -> PaimonScanParams.resolveOptions(getRawSysPaimonDataTable(), options)); + } + + private FileStoreTable getRawSysPaimonDataTable() { + getRawSysPaimonTable(); + return paimonSysDataTable; + } + + private Table createSystemTable(FileStoreTable dataTable) { + Table systemTable = SystemTableLoader.load(sysTableType, dataTable); + if (systemTable == null) { + throw new IllegalArgumentException("Unknown Paimon system table '" + sysTableType + "'."); + } + return systemTable; } public List getFullSchema(TableScanParams scanParams) { @@ -187,7 +227,9 @@ private boolean resolveIsDataTable() { if (isDataTable == null) { synchronized (this) { if (isDataTable == null) { - isDataTable = getSysPaimonTable() instanceof DataTable; + // Type inspection happens before relation parameters reach ScanNode. It must not + // reject a hidden physical value that a later OPTIONS copy safely overrides. + isDataTable = getRawSysPaimonTable() instanceof DataTable; } } } @@ -273,7 +315,9 @@ private SchemaCacheValue getOrCreateSchemaCacheValue() { } private List buildFullSchema() { - Table sysTable = getSysPaimonTable(); + // Schema and descriptor construction do not plan a scan. Using the raw wrapper keeps a + // relation-scoped safe override from being replaced by validation of the physical handle. + Table sysTable = getRawSysPaimonTable(); List fields = sysTable.rowType().getFields(); List columns = Lists.newArrayListWithCapacity(fields.size()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtils.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtils.java index dc28c083ca10fa..3503493343d34e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtils.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtils.java @@ -35,6 +35,10 @@ public static PaimonSnapshotCacheValue getLatestSnapshotCacheValue(ExternalTable return paimonExternalMetaCache(dorisTable).getSnapshotCache(dorisTable); } + public static PaimonSnapshotCacheValue loadSnapshotProjection(ExternalTable dorisTable, Table effectiveTable) { + return paimonExternalMetaCache(dorisTable).loadSnapshotProjection(dorisTable, effectiveTable); + } + public static PaimonSnapshotCacheValue getSnapshotCacheValue(Optional snapshot, ExternalTable dorisTable) { if (snapshot.isPresent() && snapshot.get() instanceof PaimonMvccSnapshot) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java index 39e8cfcfbf0185..75b3c1800bc6a1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java @@ -32,6 +32,7 @@ import org.apache.doris.datasource.credentials.CredentialUtils; import org.apache.doris.datasource.credentials.VendedCredentialsFactory; import org.apache.doris.datasource.paimon.PaimonExternalCatalog; +import org.apache.doris.datasource.paimon.PaimonReaderOptions; import org.apache.doris.datasource.paimon.PaimonScanParams; import org.apache.doris.datasource.paimon.PaimonSysExternalTable; import org.apache.doris.datasource.paimon.PaimonUtil; @@ -995,18 +996,32 @@ private Table getProcessedTable() throws UserException { throw new UserException("Can not specify scan params and table snapshot at same time."); } + Table finalTable; if (theScanParams != null && theScanParams.incrementalRead()) { // System table handles are cached, so preserve query isolation by applying dynamic // options to a copied Paimon table instead of changing the shared handle. - return baseTable.copy(getIncrReadParams()); - } - if (theScanParams != null && theScanParams.isOptions()) { + finalTable = baseTable.copy(getIncrReadParams()); + } else if (theScanParams != null && theScanParams.isOptions()) { try { - return source.getPaimonTable(theScanParams); + finalTable = source.getPaimonTable(theScanParams); } catch (IllegalArgumentException e) { throw new UserException(e.getMessage(), e); } + } else { + finalTable = baseTable; + } + try { + // This is the last common boundary before planning and serialization, including scans + // with no relation copy and incremental/system-table paths that bypass applyOptions. + PaimonReaderOptions.validateEffectiveTable(finalTable); + if (source.getExternalTable() instanceof PaimonSysExternalTable) { + // Read-only system wrappers hide the data table that performs manifest planning. + ((PaimonSysExternalTable) source.getExternalTable()) + .validateEffectiveDataTable(theScanParams); + } + } catch (IllegalArgumentException e) { + throw new UserException(e.getMessage(), e); } - return baseTable; + return finalTable; } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonSource.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonSource.java index e8a3fb33d957ed..1ca508b4457e27 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonSource.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonSource.java @@ -86,7 +86,9 @@ private Table resolvePaimonTable(ExternalTable table) { return ((PaimonExternalTable) table).getPaimonTable(snapshot); } if (table instanceof PaimonSysExternalTable) { - return ((PaimonSysExternalTable) table).getSysPaimonTable(); + // Relation options are unavailable during source construction, so defer validation of + // the wrapper's hidden data table until getProcessedTable has the complete precedence chain. + return ((PaimonSysExternalTable) table).getRawSysPaimonTable(); } throw new IllegalArgumentException( "Expected Paimon table but got " + table.getClass().getSimpleName()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractPaimonProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractPaimonProperties.java index 70d16dfebc1937..5ea5a156a0d948 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractPaimonProperties.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractPaimonProperties.java @@ -18,6 +18,7 @@ package org.apache.doris.datasource.property.metastore; import org.apache.doris.common.security.authentication.ExecutionAuthenticator; +import org.apache.doris.datasource.paimon.PaimonReaderOptions; import org.apache.doris.datasource.storage.StorageAdapter; import org.apache.doris.foundation.property.ConnectorProperty; @@ -25,16 +26,12 @@ import lombok.Getter; import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.conf.Configuration; -import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.Catalog; import org.apache.paimon.options.CatalogOptions; -import org.apache.paimon.options.ConfigOption; -import org.apache.paimon.options.FallbackKey; import org.apache.paimon.options.Options; import java.util.Collections; import java.util.HashMap; -import java.util.LinkedHashMap; import java.util.List; import java.util.Locale; import java.util.Map; @@ -61,8 +58,7 @@ public abstract class AbstractPaimonProperties extends MetastoreProperties { private static final String USER_PROPERTY_PREFIX = "paimon."; private static final String DORIS_JNI_PROPERTY_PREFIX = "paimon.jni."; /** The suffix after this prefix is passed to Paimon as a dynamic table option. */ - public static final String TABLE_OPTION_PREFIX = "paimon.table-option."; - private static final SupportedTableOptions SUPPORTED_TABLE_OPTIONS = SupportedTableOptions.build(); + public static final String TABLE_OPTION_PREFIX = PaimonReaderOptions.TABLE_OPTION_PREFIX; private Map tableOptionsMap = Collections.emptyMap(); @@ -146,25 +142,12 @@ public Map getTableOptionsMap() { } /** - * Returns Catalog table options which are not explicitly configured by the Paimon table. - * - *

The comparison is based on Paimon {@link ConfigOption}s so canonical and fallback keys - * follow the same precedence rule. + * Returns Catalog-scoped dynamic reader options to copy onto a Paimon table. + * Catalog options intentionally override physical table values, while a subsequent + * relation-scoped copy can override the Catalog for one relation only. */ - public Map getTableOptionsForCopy(Map currentTableOptions) { - if (tableOptionsMap.isEmpty() || currentTableOptions.isEmpty()) { - return tableOptionsMap; - } - - Options existingOptions = new Options(currentTableOptions); - Map optionsForCopy = new LinkedHashMap<>(); - tableOptionsMap.forEach((key, value) -> { - ConfigOption option = SUPPORTED_TABLE_OPTIONS.find(key); - if (!existingOptions.contains(option)) { - optionsForCopy.put(key, value); - } - }); - return Collections.unmodifiableMap(optionsForCopy); + public Map getTableOptionsForCopy() { + return tableOptionsMap; } public static boolean isTableOptionProperty(String key) { @@ -172,58 +155,7 @@ public static boolean isTableOptionProperty(String key) { } private Map extractTableOptions() { - Map tableOptions = new LinkedHashMap<>(); - origProps.forEach((key, value) -> { - if (isTableOptionProperty(key)) { - String tableOptionKey = key.substring(TABLE_OPTION_PREFIX.length()); - if (StringUtils.isBlank(tableOptionKey)) { - throw new IllegalArgumentException( - "Paimon table option name must not be empty after prefix " + TABLE_OPTION_PREFIX); - } - validateTableOption(tableOptionKey, value); - tableOptions.put(tableOptionKey, value); - } - }); - return Collections.unmodifiableMap(tableOptions); - } - - private void validateTableOption(String key, String value) { - ConfigOption option = SUPPORTED_TABLE_OPTIONS.find(key); - if (option == null) { - throw new IllegalArgumentException("Unsupported Paimon table option '" + key - + "' for the bundled Paimon version"); - } - - try { - new Options(Collections.singletonMap(key, value)).get(option); - } catch (IllegalArgumentException e) { - throw new IllegalArgumentException("Invalid value for Paimon table option '" + key + "': " - + e.getMessage(), e); - } - } - - private static final class SupportedTableOptions { - /** Canonical and fallback option names which support direct lookup. */ - private final Map> exactOptions; - - private SupportedTableOptions(Map> exactOptions) { - this.exactOptions = exactOptions; - } - - private static SupportedTableOptions build() { - Map> exactOptions = new HashMap<>(); - for (ConfigOption option : CoreOptions.getOptions()) { - exactOptions.put(option.key(), option); - for (FallbackKey fallbackKey : option.fallbackKeys()) { - exactOptions.put(fallbackKey.getKey(), option); - } - } - return new SupportedTableOptions(Collections.unmodifiableMap(exactOptions)); - } - - private ConfigOption find(String key) { - return exactOptions.get(key); - } + return PaimonReaderOptions.compatibleCatalogOptions(origProps); } /** diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogMgrTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogMgrTest.java new file mode 100644 index 00000000000000..48c447342fc949 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogMgrTest.java @@ -0,0 +1,191 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.datasource; + +import org.apache.doris.catalog.DatabaseIf; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.DdlException; +import org.apache.doris.datasource.paimon.PaimonExternalCatalog; +import org.apache.doris.datasource.property.metastore.AbstractPaimonProperties; + +import com.google.common.collect.ImmutableMap; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +public class CatalogMgrTest { + + private static void addCatalog(CatalogMgr catalogMgr, ExternalCatalog catalog) throws Exception { + Field idToCatalogField = CatalogMgr.class.getDeclaredField("idToCatalog"); + idToCatalogField.setAccessible(true); + @SuppressWarnings("unchecked") + ConcurrentMap>> idToCatalog = + (ConcurrentMap>>) + idToCatalogField.get(catalogMgr); + idToCatalog.put(catalog.getId(), catalog); + } + + @Test + void testAlterCatalogRollsBackUncheckedValidationFailure() throws Exception { + CatalogMgr catalogMgr = new CatalogMgr(); + ExternalCatalog catalog = Mockito.mock(ExternalCatalog.class); + long catalogId = 42L; + Mockito.when(catalog.getId()).thenReturn(catalogId); + + addCatalog(catalogMgr, catalog); + + Map oldProperties = ImmutableMap.of("read.batch-size", "1024"); + Map newProperties = ImmutableMap.of("read.batch-size", "0"); + CatalogLog log = new CatalogLog(); + log.setCatalogId(catalogId); + log.setNewProps(newProperties); + Mockito.doThrow(new IllegalArgumentException("invalid reader option")) + .when(catalog).checkProperties(); + + DdlException exception = Assertions.assertThrows(DdlException.class, + () -> catalogMgr.replayAlterCatalogProps(log, oldProperties, false)); + + Assertions.assertTrue(exception.getMessage().contains("invalid reader option")); + Mockito.verify(catalog).tryModifyCatalogProps(newProperties); + Mockito.verify(catalog).rollBackCatalogProps(oldProperties); + Mockito.verify(catalog, Mockito.never()).modifyCatalogProps(newProperties); + } + + @Test + void testDetachedValidationNeverPublishesCandidateToConcurrentInitialization() throws Exception { + CatalogMgr catalogMgr = new CatalogMgr(); + Map oldProperties = ImmutableMap.of("read.batch-size", "1024"); + Map newProperties = ImmutableMap.of( + "read.batch-size", "4096", + CatalogMgr.METADATA_REFRESH_INTERVAL_SEC, "invalid"); + LatchingValidationCatalog catalog = new LatchingValidationCatalog(43L, oldProperties); + addCatalog(catalogMgr, catalog); + CatalogLog log = new CatalogLog(); + log.setCatalogId(catalog.getId()); + log.setNewProps(newProperties); + ExecutorService executor = Executors.newSingleThreadExecutor(); + + try { + Future alterResult = executor.submit(() -> { + try { + catalogMgr.replayAlterCatalogProps(log, oldProperties, false); + return null; + } catch (DdlException e) { + return e; + } + }); + Assertions.assertTrue(catalog.validationStarted.await(10, TimeUnit.SECONDS)); + + Assertions.assertThrows(RuntimeException.class, catalog::makeSureInitialized); + DdlException validationFailure = alterResult.get(10, TimeUnit.SECONDS); + + Assertions.assertNotNull(validationFailure); + Assertions.assertEquals(oldProperties, catalog.propertiesSeenByInitialization); + Assertions.assertEquals(oldProperties, catalog.getProperties()); + } finally { + executor.shutdownNow(); + } + } + + @Test + void testReplayKeepsPersistedLegacyPaimonOptionLoadableButInactive() throws Exception { + CatalogMgr catalogMgr = new CatalogMgr(); + Map persistedProperties = new HashMap<>(); + persistedProperties.put("type", "paimon"); + persistedProperties.put("paimon.catalog.type", "filesystem"); + persistedProperties.put("warehouse", "s3://example-bucket/warehouse"); + persistedProperties.put("paimon.table-option.write.batch-size", "2048"); + ReplayCompatiblePaimonCatalog catalog = new ReplayCompatiblePaimonCatalog(44L, persistedProperties); + addCatalog(catalogMgr, catalog); + CatalogLog log = new CatalogLog(); + log.setCatalogId(catalog.getId()); + log.setNewProps(ImmutableMap.of(ExternalCatalog.USE_META_CACHE, "false")); + + catalogMgr.replayAlterCatalogProps(log, persistedProperties, true); + + AbstractPaimonProperties restoredProperties = (AbstractPaimonProperties) + catalog.getCatalogProperty().getMetastoreProperties(); + Assertions.assertEquals("2048", + catalog.getProperties().get("paimon.table-option.write.batch-size")); + Assertions.assertTrue(restoredProperties.getTableOptionsMap().isEmpty()); + } + + private static class LatchingValidationCatalog extends ExternalCatalog { + private final CountDownLatch validationStarted = new CountDownLatch(1); + private final CountDownLatch initializationReadProperties = new CountDownLatch(1); + private volatile Map propertiesSeenByInitialization; + + LatchingValidationCatalog(long id, Map properties) { + super(id, "latching_catalog", InitCatalogLog.Type.TEST, ""); + catalogProperty = new CatalogProperty(null, properties); + } + + @Override + public boolean validatePropertiesBeforeUpdate( + Map currentProperties, Map updatedProperties) { + validationStarted.countDown(); + try { + Assertions.assertTrue(initializationReadProperties.await(10, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + throw new IllegalArgumentException("invalid candidate properties"); + } + + @Override + protected void initLocalObjectsImpl() { + propertiesSeenByInitialization = getProperties(); + initializationReadProperties.countDown(); + throw new IllegalStateException("stop after observing properties"); + } + + @Override + protected List listTableNamesFromRemote(SessionContext ctx, String dbName) { + return Collections.emptyList(); + } + + @Override + public boolean tableExist(SessionContext ctx, String dbName, String tblName) { + return false; + } + } + + private static class ReplayCompatiblePaimonCatalog extends PaimonExternalCatalog { + ReplayCompatiblePaimonCatalog(long id, Map properties) { + super(id, "persisted_paimon_catalog", null, properties, ""); + } + + @Override + public void notifyPropertiesUpdated(Map updatedProps) { + // This test isolates edit-log property restoration from environment-owned cache services. + } + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCacheTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCacheTest.java index 3389eb23918b07..06b32ea9d07815 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCacheTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCacheTest.java @@ -17,6 +17,9 @@ package org.apache.doris.datasource.paimon; +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.Type; +import org.apache.doris.datasource.CacheException; import org.apache.doris.datasource.ExternalCatalog; import org.apache.doris.datasource.NameMapping; import org.apache.doris.datasource.SchemaCacheValue; @@ -26,11 +29,19 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.AppendOnlyFileStoreTable; +import org.apache.paimon.table.CatalogEnvironment; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.IntType; import org.junit.Assert; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; import org.mockito.Mockito; import java.util.Collections; @@ -40,11 +51,13 @@ import java.util.concurrent.Executors; public class PaimonExternalMetaCacheTest { + @Rule + public TemporaryFolder temporaryFolder = new TemporaryFolder(); @Test public void testLatestSnapshotUsesLatestSchemaForPinnedRead() { PaimonLatestSnapshotProjectionLoader loader = new PaimonLatestSnapshotProjectionLoader( - new PaimonPartitionInfoLoader(null), + new PaimonPartitionInfoLoader(), (nameMapping, schemaId) -> new PaimonSchemaCacheValue( Collections.emptyList(), Collections.emptyList(), null)); NameMapping nameMapping = new NameMapping(1L, "db", "table", "remote_db", "remote_table"); @@ -71,6 +84,69 @@ public void testLatestSnapshotUsesLatestSchemaForPinnedRead() { Mockito.verify(pinnedTable).copyWithLatestSchema(); } + @Test + public void testPartitionProjectionRejectsUnsafeEffectiveTableBeforeEnumeration() throws Exception { + FileStoreTable unsafeTable = newPartitionedTable( + "unsafe", Collections.singletonMap("scan.manifest.parallelism", "0")); + PaimonPartitionInfoLoader loader = new PaimonPartitionInfoLoader(); + + try { + loader.load(new NameMapping(1L, "db", "table", "db", "table"), unsafeTable, + Collections.singletonList(new Column("part", Type.INT))); + Assert.fail("partition projection must reject unsafe manifest parallelism before enumeration"); + } catch (CacheException e) { + Assert.assertTrue(e.getMessage().contains("scan.manifest.parallelism")); + } + } + + @Test + public void testPartitionProjectionUsesSafeCopiedTableInsteadOfRawCatalogReload() throws Exception { + FileStoreTable unsafeTable = newPartitionedTable( + "safe_override", Collections.singletonMap("scan.manifest.parallelism", "0")); + FileStoreTable safeTable = unsafeTable.copy( + Collections.singletonMap("scan.manifest.parallelism", "1")); + PaimonPartitionInfoLoader loader = new PaimonPartitionInfoLoader(); + + PaimonPartitionInfo partitionInfo = loader.load( + new NameMapping(1L, "db", "table", "db", "table"), safeTable, + Collections.singletonList(new Column("part", Type.INT))); + + Assert.assertTrue(partitionInfo.getNameToPartition().isEmpty()); + } + + @Test + public void testPartitionProjectionIgnoresReaderOnlyPhysicalOptions() throws Exception { + FileStoreTable table = newPartitionedTable( + "reader_only", Collections.singletonMap("read.batch-size", "0")); + PaimonPartitionInfoLoader loader = new PaimonPartitionInfoLoader(); + + // Partition metadata does not use the data-reader batch size. The final relation copy is + // validated separately before scanning, so a safe override can reuse this projection. + PaimonPartitionInfo partitionInfo = loader.load( + new NameMapping(1L, "db", "table", "db", "table"), table, + Collections.singletonList(new Column("part", Type.INT))); + + Assert.assertTrue(partitionInfo.getNameToPartition().isEmpty()); + } + + private FileStoreTable newPartitionedTable(String name, Map options) throws Exception { + TableSchema schema = new TableSchema( + 0, + java.util.Arrays.asList( + new DataField(0, "id", new IntType()), + new DataField(1, "part", new IntType())), + 1, + Collections.singletonList("part"), + Collections.emptyList(), + options, + null); + return new AppendOnlyFileStoreTable( + LocalFileIO.create(), + new Path(temporaryFolder.newFolder(name).toURI()), + schema, + CatalogEnvironment.empty()); + } + @Test public void testInvalidateTablePrecise() { ExecutorService executor = Executors.newSingleThreadExecutor(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java index 6b0116433514f9..ba5257cd24e256 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java @@ -18,19 +18,37 @@ package org.apache.doris.datasource.paimon; import org.apache.doris.analysis.TableScanParams; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.datasource.ExternalRowCountCache; import org.apache.doris.datasource.mvcc.MvccSnapshot; import org.apache.doris.datasource.mvcc.MvccUtil; +import org.apache.doris.statistics.util.StatisticsUtil; import com.google.common.collect.ImmutableMap; +import com.google.common.util.concurrent.MoreExecutors; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.AppendOnlyFileStoreTable; +import org.apache.paimon.table.CatalogEnvironment; +import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.Table; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.table.system.PartitionsTable; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.IntType; import org.junit.Assert; import org.junit.Test; import org.mockito.ArgumentMatchers; import org.mockito.MockedStatic; import org.mockito.Mockito; +import java.util.Arrays; import java.util.Collections; +import java.util.Map; import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; public class PaimonExternalTableTest { @@ -51,6 +69,7 @@ public void testSelectorFreeOptionsPreserveStatementSnapshot() { Mockito.doReturn(statementTable).when(externalTable).getPaimonTable(statementSnapshot); Mockito.when(statementTable.copy(scanParams.getMapParams())).thenReturn(statementCopy); + Mockito.when(statementCopy.options()).thenReturn(scanParams.getMapParams()); Mockito.when(baseTable.copy(scanParams.getMapParams())).thenReturn(baseCopy); try (MockedStatic mvccUtil = Mockito.mockStatic(MvccUtil.class); @@ -79,6 +98,7 @@ public void testModeOnlyLatestUsesStatementSnapshotAcrossPhases() { Mockito.doReturn(statementTable).when(externalTable).getPaimonTable(statementSnapshot); Mockito.when(statementTable.options()).thenReturn(ImmutableMap.of("scan.snapshot-id", "7")); Mockito.when(baseTable.copy(ArgumentMatchers.anyMap())).thenReturn(pinnedCopy); + Mockito.when(pinnedCopy.options()).thenReturn(ImmutableMap.of("scan.snapshot-id", "7")); try (MockedStatic mvccUtil = Mockito.mockStatic(MvccUtil.class); MockedStatic paimonUtils = Mockito.mockStatic(PaimonUtils.class)) { @@ -94,4 +114,214 @@ public void testModeOnlyLatestUsesStatementSnapshotAcrossPhases() { && options.containsKey("scan.mode") && options.get("scan.mode") == null)); } + + @Test + public void testRelationOptionsLoadSnapshotProjectionFromEffectiveTable() { + PaimonExternalTable externalTable = Mockito.mock( + PaimonExternalTable.class, Mockito.CALLS_REAL_METHODS); + Mockito.doNothing().when(externalTable).makeSureInitialized(); + FileStoreTable unsafeDataTable = newFileStoreTable( + "snapshot_projection", ImmutableMap.of("scan.manifest.parallelism", "0"), null); + TableScanParams scanParams = new TableScanParams( + TableScanParams.OPTIONS, + ImmutableMap.of("scan.manifest.parallelism", "1"), + Collections.emptyList()); + PaimonSnapshotCacheValue projection = Mockito.mock(PaimonSnapshotCacheValue.class); + + try (MockedStatic paimonUtils = Mockito.mockStatic(PaimonUtils.class)) { + paimonUtils.when(() -> PaimonUtils.getPaimonTable(externalTable)).thenReturn(unsafeDataTable); + paimonUtils.when(() -> PaimonUtils.loadSnapshotProjection( + Mockito.eq(externalTable), Mockito.any(Table.class))).thenReturn(projection); + + PaimonMvccSnapshot snapshot = (PaimonMvccSnapshot) externalTable.loadSnapshot( + Optional.empty(), Optional.of(scanParams)); + + Assert.assertSame(projection, snapshot.getSnapshotCacheValue()); + paimonUtils.verify(() -> PaimonUtils.loadSnapshotProjection( + Mockito.eq(externalTable), Mockito.argThat(table -> + "1".equals(table.options().get("scan.manifest.parallelism"))))); + } + } + + @Test + public void testReaderOnlyOptionsReusePartitionedMemoizedProjection() { + PaimonExternalTable externalTable = Mockito.mock( + PaimonExternalTable.class, Mockito.CALLS_REAL_METHODS); + Mockito.doNothing().when(externalTable).makeSureInitialized(); + FileStoreTable partitionedTable = newPartitionedFileStoreTable("reader_only_projection"); + TableScanParams scanParams = new TableScanParams( + TableScanParams.OPTIONS, + ImmutableMap.of("read.batch-size", "4096"), + Collections.emptyList()); + PaimonSnapshotCacheValue memoizedProjection = Mockito.mock(PaimonSnapshotCacheValue.class); + PaimonSnapshotCacheValue directProjection = Mockito.mock(PaimonSnapshotCacheValue.class); + AtomicInteger directProjectionLoads = new AtomicInteger(); + + try (MockedStatic paimonUtils = Mockito.mockStatic(PaimonUtils.class)) { + paimonUtils.when(() -> PaimonUtils.getPaimonTable(externalTable)).thenReturn(partitionedTable); + paimonUtils.when(() -> PaimonUtils.getLatestSnapshotCacheValue(externalTable)) + .thenReturn(memoizedProjection); + paimonUtils.when(() -> PaimonUtils.loadSnapshotProjection( + Mockito.eq(externalTable), Mockito.any(Table.class))).thenAnswer(invocation -> { + directProjectionLoads.incrementAndGet(); + return directProjection; + }); + + PaimonMvccSnapshot snapshot = (PaimonMvccSnapshot) externalTable.loadSnapshot( + Optional.empty(), Optional.of(scanParams)); + + Assert.assertSame(memoizedProjection, snapshot.getSnapshotCacheValue()); + Assert.assertEquals(0, directProjectionLoads.get()); + } + } + + @Test + public void testSystemTableIsBuiltFromTheValidatedSourceHandle() { + FileStoreTable safeSource = newFileStoreTable( + "safe_cached_source", ImmutableMap.of("scan.manifest.parallelism", "1"), null); + FileStoreTable unsafeReload = newFileStoreTable( + "unsafe_catalog_reload", ImmutableMap.of("scan.manifest.parallelism", "0"), null); + Table divergentWrapper = new PartitionsTable(unsafeReload); + PaimonExternalTable sourceTable = Mockito.mock(PaimonExternalTable.class); + PaimonExternalCatalog catalog = Mockito.mock(PaimonExternalCatalog.class); + PaimonExternalDatabase database = Mockito.mock(PaimonExternalDatabase.class); + Mockito.when(catalog.getId()).thenReturn(1L); + Mockito.when(database.getFullName()).thenReturn("db"); + Mockito.when(database.getRemoteName()).thenReturn("db"); + Mockito.when(sourceTable.getId()).thenReturn(10L); + Mockito.when(sourceTable.getName()).thenReturn("source"); + Mockito.when(sourceTable.getRemoteName()).thenReturn("source"); + Mockito.when(sourceTable.getCatalog()).thenReturn(catalog); + Mockito.when(sourceTable.getDatabase()).thenReturn(database); + Mockito.doReturn(safeSource).when(sourceTable).getBasePaimonTable(); + Mockito.when(catalog.getPaimonTable(Mockito.any(), Mockito.anyString(), Mockito.anyString())) + .thenReturn(divergentWrapper); + PaimonSysExternalTable systemTable = new PaimonSysExternalTable(sourceTable, "partitions"); + + Table actual = systemTable.getSysPaimonTable(); + + Assert.assertNotSame(divergentWrapper, actual); + Mockito.verify(catalog, Mockito.never()).getPaimonTable( + Mockito.any(), Mockito.anyString(), Mockito.anyString()); + } + + @Test + public void testPartitionsTableValidatesHiddenDataTableWithOverridePrecedence() throws Exception { + FileStoreTable unsafeDataTable = newFileStoreTable( + "partitions", ImmutableMap.of("scan.manifest.parallelism", "0"), null); + PaimonSysExternalTable systemTable = newSystemTable(unsafeDataTable); + + try { + systemTable.getSysPaimonTable(); + Assert.fail("$partitions must reject an unsafe hidden data table without an override"); + } catch (IllegalArgumentException e) { + Assert.assertTrue(e.getMessage().contains("scan.manifest.parallelism")); + } + + TableScanParams safeOverride = new TableScanParams( + TableScanParams.OPTIONS, + ImmutableMap.of("scan.manifest.parallelism", "1"), + Collections.emptyList()); + Assert.assertTrue(systemTable.getSysPaimonTable(safeOverride) instanceof PartitionsTable); + Assert.assertFalse(systemTable.isDataTable()); + // Descriptor serialization happens after relation binding and must not fall back to + // validating the unsafe physical handle that the relation override already replaced. + Assert.assertNotNull(systemTable.toThrift()); + } + + @Test + public void testFetchRowCountValidatesBeforePlanning() { + AtomicBoolean planningStarted = new AtomicBoolean(false); + FileStoreTable unsafeDataTable = newFileStoreTable( + "row_count", ImmutableMap.of("scan.manifest.parallelism", "0"), planningStarted); + PaimonExternalTable externalTable = Mockito.mock( + PaimonExternalTable.class, Mockito.CALLS_REAL_METHODS); + Mockito.doNothing().when(externalTable).makeSureInitialized(); + + try (MockedStatic paimonUtils = Mockito.mockStatic(PaimonUtils.class)) { + paimonUtils.when(() -> PaimonUtils.getPaimonTable(externalTable)).thenReturn(unsafeDataTable); + try { + externalTable.fetchRowCount(); + Assert.fail("row-count planning must reject unsafe manifest parallelism"); + } catch (IllegalArgumentException e) { + Assert.assertTrue(e.getMessage().contains("scan.manifest.parallelism")); + } + } + Assert.assertFalse(planningStarted.get()); + } + + @Test + public void testExternalRowCountCacheUsesGuardedPaimonFetch() { + AtomicBoolean planningStarted = new AtomicBoolean(false); + FileStoreTable unsafeDataTable = newFileStoreTable( + "row_count_cache", ImmutableMap.of("scan.manifest.parallelism", "0"), planningStarted); + PaimonExternalTable externalTable = Mockito.mock( + PaimonExternalTable.class, Mockito.CALLS_REAL_METHODS); + Mockito.doNothing().when(externalTable).makeSureInitialized(); + + try (MockedStatic paimonUtils = Mockito.mockStatic(PaimonUtils.class); + MockedStatic statisticsUtil = Mockito.mockStatic(StatisticsUtil.class)) { + paimonUtils.when(() -> PaimonUtils.getPaimonTable(externalTable)).thenReturn(unsafeDataTable); + statisticsUtil.when(() -> StatisticsUtil.findTable(1, 2, 3)).thenReturn(externalTable); + ExternalRowCountCache cache = new ExternalRowCountCache(MoreExecutors.newDirectExecutorService()); + + Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT, cache.getCachedRowCount(1, 2, 3, false)); + } + Assert.assertFalse(planningStarted.get()); + } + + private PaimonSysExternalTable newSystemTable(FileStoreTable dataTable) { + PaimonExternalTable sourceTable = Mockito.mock(PaimonExternalTable.class); + PaimonExternalCatalog catalog = Mockito.mock(PaimonExternalCatalog.class); + PaimonExternalDatabase database = Mockito.mock(PaimonExternalDatabase.class); + Mockito.when(catalog.getId()).thenReturn(1L); + Mockito.when(database.getFullName()).thenReturn("db"); + Mockito.when(database.getRemoteName()).thenReturn("db"); + Mockito.when(sourceTable.getId()).thenReturn(10L); + Mockito.when(sourceTable.getName()).thenReturn("source"); + Mockito.when(sourceTable.getRemoteName()).thenReturn("source"); + Mockito.when(sourceTable.getCatalog()).thenReturn(catalog); + Mockito.when(sourceTable.getDatabase()).thenReturn(database); + Mockito.when(sourceTable.getPaimonCatalogType()) + .thenReturn(PaimonExternalCatalog.PAIMON_FILESYSTEM); + Mockito.doReturn(dataTable).when(sourceTable).getBasePaimonTable(); + return new PaimonSysExternalTable(sourceTable, "partitions"); + } + + private FileStoreTable newFileStoreTable( + String name, Map options, AtomicBoolean planningStarted) { + TableSchema schema = new TableSchema( + 0, + Collections.singletonList(new DataField(0, "id", new IntType())), + 0, + Collections.emptyList(), + Collections.emptyList(), + options, + null); + return new AppendOnlyFileStoreTable( + Mockito.mock(FileIO.class), new Path("memory://" + name), schema, CatalogEnvironment.empty()) { + @Override + public ReadBuilder newReadBuilder() { + if (planningStarted != null) { + planningStarted.set(true); + } + return super.newReadBuilder(); + } + }; + } + + private FileStoreTable newPartitionedFileStoreTable(String name) { + TableSchema schema = new TableSchema( + 0, + Arrays.asList( + new DataField(0, "id", new IntType()), + new DataField(1, "part", new IntType())), + 1, + Collections.singletonList("part"), + Collections.emptyList(), + Collections.emptyMap(), + null); + return new AppendOnlyFileStoreTable( + Mockito.mock(FileIO.class), new Path("memory://" + name), schema, CatalogEnvironment.empty()); + } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonReaderOptionsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonReaderOptionsTest.java new file mode 100644 index 00000000000000..82cdd46e8278b8 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonReaderOptionsTest.java @@ -0,0 +1,179 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.datasource.paimon; + +import org.apache.doris.datasource.property.metastore.AbstractPaimonProperties; + +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.privilege.PrivilegeChecker; +import org.apache.paimon.privilege.PrivilegedFileStoreTable; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.AppendOnlyFileStoreTable; +import org.apache.paimon.table.CatalogEnvironment; +import org.apache.paimon.table.FallbackReadFileStoreTable; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.IntType; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.Map; + +public class PaimonReaderOptionsTest { + + @Test + void testSupportsSafeBatchReadOptions() { + Assertions.assertEquals(ImmutableSet.of( + CoreOptions.READ_BATCH_SIZE.key(), + CoreOptions.FILE_READER_ASYNC_THRESHOLD.key(), + CoreOptions.FILE_INDEX_READ_ENABLED.key(), + CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), + CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(), + CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), + CoreOptions.SCAN_PLAN_SORT_PARTITION.key()), + PaimonReaderOptions.supportedOptions()); + + Assertions.assertDoesNotThrow(() -> PaimonReaderOptions.validateCatalogProperties(ImmutableMap.of( + AbstractPaimonProperties.TABLE_OPTION_PREFIX + "file-index.read.enabled", "false", + AbstractPaimonProperties.TABLE_OPTION_PREFIX + "source.split.target-size", "64 MB", + AbstractPaimonProperties.TABLE_OPTION_PREFIX + "source.split.open-file-cost", "1 MB", + AbstractPaimonProperties.TABLE_OPTION_PREFIX + "scan.manifest.parallelism", "1", + AbstractPaimonProperties.TABLE_OPTION_PREFIX + "scan.plan-sort-partition", "true"))); + } + + @Test + void testRejectsInvalidSafeBatchReadOptions() { + for (Map options : new Map[] { + ImmutableMap.of("file-index.read.enabled", "not-a-boolean"), + ImmutableMap.of("source.split.target-size", "0 B"), + ImmutableMap.of("source.split.open-file-cost", "-1 B"), + ImmutableMap.of("scan.plan-sort-partition", "not-a-boolean") + }) { + Assertions.assertThrows(IllegalArgumentException.class, + () -> PaimonReaderOptions.validateEffectiveTableOptions(options)); + } + } + + @Test + void testRejectUnsafeOptionsFromCreateOrAlterProperties() { + for (Map properties : new Map[] { + ImmutableMap.of(AbstractPaimonProperties.TABLE_OPTION_PREFIX + "branch", "archive"), + ImmutableMap.of(AbstractPaimonProperties.TABLE_OPTION_PREFIX + "read.batch-size", "0"), + ImmutableMap.of(AbstractPaimonProperties.TABLE_OPTION_PREFIX + + "file-reader-async-threshold", "2 GB") + }) { + Assertions.assertThrows(IllegalArgumentException.class, + () -> PaimonReaderOptions.validateCatalogProperties(properties)); + } + } + + @Test + void testRejectUnsafeEffectivePhysicalTableOptions() { + for (Map options : new Map[] { + ImmutableMap.of("read.batch-size", "0"), + ImmutableMap.of("file-reader-async-threshold", "512 KB"), + ImmutableMap.of("scan.manifest.parallelism", "0"), + ImmutableMap.of("scan.manifest.parallelism", + String.valueOf(Runtime.getRuntime().availableProcessors() + 1)) + }) { + Assertions.assertThrows(IllegalArgumentException.class, + () -> PaimonReaderOptions.validateEffectiveTableOptions(options)); + } + } + + @Test + void testRejectUnsafeOptionOnlyAfterFinalTableCopy() { + Table physicalTable = Mockito.mock(Table.class); + Table finalTable = Mockito.mock(Table.class); + Mockito.when(physicalTable.copy(Collections.emptyMap())).thenReturn(finalTable); + Mockito.when(finalTable.options()).thenReturn(ImmutableMap.of("read.batch-size", "0")); + + Assertions.assertThrows(IllegalArgumentException.class, + () -> PaimonScanParams.applyOptions(physicalTable, Collections.emptyMap())); + } + + @Test + void testSafeRelationOptionOverridesUnsafePhysicalOption() { + Table physicalTable = Mockito.mock(Table.class); + Table finalTable = Mockito.mock(Table.class); + Map relationOptions = ImmutableMap.of("read.batch-size", "4096"); + Mockito.when(physicalTable.options()).thenReturn(ImmutableMap.of("read.batch-size", "0")); + Mockito.when(physicalTable.copy(relationOptions)).thenReturn(finalTable); + Mockito.when(finalTable.options()).thenReturn(relationOptions); + + Assertions.assertSame(finalTable, PaimonScanParams.applyOptions(physicalTable, relationOptions)); + } + + @Test + void testRejectUnsafeHiddenFallbackTableAfterCopy() { + FileStoreTable main = newFileStoreTable("main", Collections.emptyMap()); + FileStoreTable fallback = newFileStoreTable( + "fallback", ImmutableMap.of("scan.manifest.parallelism", "0")); + Table fallbackReadTable = new FallbackReadFileStoreTable(main, fallback); + + Assertions.assertThrows(IllegalArgumentException.class, + () -> PaimonScanParams.applyOptions(fallbackReadTable, Collections.emptyMap())); + } + + @Test + void testSafeRelationOptionOverridesUnsafeHiddenFallbackTable() { + FileStoreTable main = newFileStoreTable("main", Collections.emptyMap()); + FileStoreTable fallback = newFileStoreTable( + "fallback", ImmutableMap.of("scan.manifest.parallelism", "0")); + Table fallbackReadTable = new FallbackReadFileStoreTable(main, fallback); + + Assertions.assertDoesNotThrow(() -> PaimonScanParams.applyOptions( + fallbackReadTable, ImmutableMap.of("scan.manifest.parallelism", "1"))); + } + + @Test + void testRejectUnsafeFallbackHiddenByPrivilegeDelegate() { + FileStoreTable main = newFileStoreTable("privileged_main", Collections.emptyMap()); + FileStoreTable fallback = newFileStoreTable( + "privileged_fallback", ImmutableMap.of("scan.manifest.parallelism", "0")); + FileStoreTable fallbackReadTable = new FallbackReadFileStoreTable(main, fallback); + FileStoreTable privilegedTable = PrivilegedFileStoreTable.wrap( + fallbackReadTable, + Mockito.mock(PrivilegeChecker.class), + Identifier.create("db", "table")); + + Assertions.assertThrows(IllegalArgumentException.class, + () -> PaimonReaderOptions.validateEffectiveTable(privilegedTable)); + } + + private FileStoreTable newFileStoreTable(String name, Map options) { + TableSchema schema = new TableSchema( + 0, + Collections.singletonList(new DataField(0, "id", new IntType())), + 0, + Collections.emptyList(), + Collections.emptyList(), + options, + null); + return new AppendOnlyFileStoreTable( + Mockito.mock(FileIO.class), new Path("memory://" + name), schema, CatalogEnvironment.empty()); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonScanParamsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonScanParamsTest.java index b003acac955fa7..eacf56a372bc95 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonScanParamsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonScanParamsTest.java @@ -44,6 +44,72 @@ public void testValidateKnownScanOptions() { "scan.plan-sort-partition", "true")); } + @Test + public void testValidateRelationScopedReaderOptions() { + PaimonScanParams.validateOptions(ImmutableMap.of( + "read.batch-size", "4096", + "file-reader-async-threshold", "16 MB", + "file-index.read.enabled", "false", + "source.split.target-size", "64 MB", + "source.split.open-file-cost", "1 MB", + "scan.manifest.parallelism", "1", + "scan.plan-sort-partition", "true")); + + for (Map options : new Map[] { + ImmutableMap.of("read.batch-size", "0"), + ImmutableMap.of("read.batch-size", "-1"), + ImmutableMap.of("read.batch-size", "65537"), + ImmutableMap.of("file-reader-async-threshold", "512 KB"), + ImmutableMap.of("file-reader-async-threshold", "2 GB") + }) { + Assert.assertThrows(IllegalArgumentException.class, + () -> PaimonScanParams.validateOptions(options)); + } + } + + @Test + public void testPlanningOptionsDoNotReuseMetadataProjection() { + Assert.assertTrue(PaimonScanParams.hasOnlyReaderOptions(ImmutableMap.of( + "file-index.read.enabled", "false", + "source.split.target-size", "64 MB"))); + Assert.assertFalse(PaimonScanParams.hasOnlyReaderOptions( + ImmutableMap.of("scan.manifest.parallelism", "1"))); + Assert.assertFalse(PaimonScanParams.hasOnlyReaderOptions( + ImmutableMap.of("scan.plan-sort-partition", "true"))); + } + + @Test + public void testManifestParallelismCannotMutateGlobalPoolCapacity() { + int availableProcessors = Runtime.getRuntime().availableProcessors(); + PaimonScanParams.validateOptions(ImmutableMap.of( + "scan.manifest.parallelism", String.valueOf(availableProcessors))); + + for (int invalid : new int[] {0, -1, availableProcessors + 1}) { + IllegalArgumentException exception = Assert.assertThrows( + IllegalArgumentException.class, + () -> PaimonScanParams.validateOptions(ImmutableMap.of( + "scan.manifest.parallelism", String.valueOf(invalid)))); + Assert.assertTrue(exception.getMessage().contains("scan.manifest.parallelism")); + } + } + + @Test + public void testRelationReaderOptionsAreAppliedWithoutMutatingBaseTable() { + Table table = Mockito.mock(Table.class); + Table copied = Mockito.mock(Table.class); + Mockito.when(table.copy(ArgumentMatchers.anyMap())).thenReturn(copied); + Mockito.when(copied.options()).thenReturn(ImmutableMap.of( + "read.batch-size", "8192", + "file-reader-async-threshold", "32 MB")); + + Assert.assertSame(copied, PaimonScanParams.applyOptions(table, ImmutableMap.of( + "read.batch-size", "8192", + "file-reader-async-threshold", "32 MB"))); + Mockito.verify(table).copy(ImmutableMap.of( + "read.batch-size", "8192", + "file-reader-async-threshold", "32 MB")); + } + @Test public void testRejectUnknownAndConflictingOptions() { IllegalArgumentException typo = Assert.assertThrows( @@ -98,6 +164,7 @@ public void testApplyPositionClearsInheritedStartupState() { Table table = Mockito.mock(Table.class); Table copied = Mockito.mock(Table.class); Mockito.when(table.copy(ArgumentMatchers.anyMap())).thenReturn(copied); + Mockito.when(copied.options()).thenReturn(Collections.emptyMap()); Assert.assertSame(copied, PaimonScanParams.applyOptions( table, ImmutableMap.of("scan.creation-time-millis", "1000"))); @@ -122,7 +189,9 @@ && containsNull(applied, "incremental-between-scan-mode") @Test public void testApplyModeClearsInheritedPositions() { Table table = Mockito.mock(Table.class); - Mockito.when(table.copy(ArgumentMatchers.anyMap())).thenReturn(Mockito.mock(Table.class)); + Table copied = Mockito.mock(Table.class); + Mockito.when(table.copy(ArgumentMatchers.anyMap())).thenReturn(copied); + Mockito.when(copied.options()).thenReturn(Collections.emptyMap()); PaimonScanParams.applyOptions(table, ImmutableMap.of("scan.mode", "latest")); @@ -141,7 +210,9 @@ && containsNull(applied, "scan.file-creation-time-millis") @Test public void testIsolationClearsFallbackReadStateKeys() { Table table = Mockito.mock(Table.class); - Mockito.when(table.copy(ArgumentMatchers.anyMap())).thenReturn(Mockito.mock(Table.class)); + Table copied = Mockito.mock(Table.class); + Mockito.when(table.copy(ArgumentMatchers.anyMap())).thenReturn(copied); + Mockito.when(copied.options()).thenReturn(Collections.emptyMap()); PaimonScanParams.applyOptions(table, ImmutableMap.of("scan.snapshot-id", "1")); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java index b520573de6ad8b..15ac70ca6daeea 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java @@ -630,6 +630,7 @@ public void testSystemTablePassesIncrementalOptionsToPaimonTable() throws Except expectedOptions.put("incremental-to-auto-tag", null); expectedOptions.put("incremental-between", "1,2"); Mockito.when(baseTable.copy(expectedOptions)).thenReturn(copiedTable); + Mockito.when(copiedTable.options()).thenReturn(Collections.emptyMap()); try { Assert.assertSame(copiedTable, invokePrivateMethod(node, "getProcessedTable")); @@ -732,6 +733,7 @@ public void testSystemTablePassesDynamicOptionsToPaimonTable() throws Exception node.setScanParams(new TableScanParams( TableScanParams.OPTIONS, options, Collections.emptyList())); Mockito.when(baseTable.copy(ArgumentMatchers.anyMap())).thenReturn(copiedTable); + Mockito.when(copiedTable.options()).thenReturn(options); try { Assert.assertSame(copiedTable, invokePrivateMethod(node, "getProcessedTable")); @@ -786,6 +788,25 @@ public void testDataTableQueryOptionsOverrideDefaultsWithoutMutation() throws Ex Assert.assertFalse(baseTable.options().containsKey("scan.snapshot-id")); } + @Test + public void testRejectsUnsafePhysicalOptionsAtFinalPlanningBoundary() throws Exception { + PaimonScanNode node = newTestNode(new PlanNodeId(0), new TupleId(0), sv); + PaimonSource source = Mockito.mock(PaimonSource.class); + PaimonExternalTable externalTable = Mockito.mock(PaimonExternalTable.class); + Table unsafePhysicalTable = Mockito.mock(Table.class); + Mockito.when(source.getExternalTable()).thenReturn(externalTable); + Mockito.when(source.getPaimonTable()).thenReturn(unsafePhysicalTable); + Mockito.when(unsafePhysicalTable.options()).thenReturn(ImmutableMap.of("read.batch-size", "0")); + node.setSource(source); + + try { + invokePrivateMethod(node, "getProcessedTable"); + Assert.fail("The final planning boundary must reject an effective zero batch size"); + } catch (java.lang.reflect.InvocationTargetException e) { + Assert.assertTrue(e.getTargetException().getMessage().contains("read.batch-size")); + } + } + @Test public void testDataTableOptionsUseRelationScopedCatalogHandle() throws Exception { PaimonScanNode node = newTestNode(new PlanNodeId(0), new TupleId(0), sv); @@ -803,6 +824,7 @@ public void testDataTableOptionsUseRelationScopedCatalogHandle() throws Exceptio Collections.emptyList()); node.setScanParams(scanParams); Mockito.when(source.getPaimonTable(scanParams)).thenReturn(relationScopedTable); + Mockito.when(relationScopedTable.options()).thenReturn(Collections.emptyMap()); Assert.assertSame(relationScopedTable, invokePrivateMethod(node, "getProcessedTable")); Mockito.verify(source).getPaimonTable(scanParams); @@ -897,6 +919,7 @@ public void testBackendSerializationUsesDynamicOptionsTable() throws Exception { node.setScanParams(new TableScanParams( TableScanParams.OPTIONS, options, Collections.emptyList())); Mockito.when(baseTable.copy(ArgumentMatchers.anyMap())).thenReturn(copiedTable); + Mockito.when(copiedTable.options()).thenReturn(options); try { invokePrivateMethod(node, "serializeProcessedTable"); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/AbstractPaimonPropertiesTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/AbstractPaimonPropertiesTest.java index e9391e9e6d49cc..03bb2298fe4da7 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/AbstractPaimonPropertiesTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/AbstractPaimonPropertiesTest.java @@ -17,14 +17,15 @@ package org.apache.doris.datasource.property.metastore; +import org.apache.doris.datasource.CatalogProperty; import org.apache.doris.datasource.storage.StorageAdapter; +import org.apache.doris.persist.gson.GsonUtils; import org.apache.paimon.catalog.Catalog; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -93,40 +94,42 @@ void testExtractAndValidateTableOptions() { input.put("warehouse", "s3://tmp/warehouse"); input.put("paimon.jni.enable_jni_io_manager", "true"); input.put("paimon.table-option.read.batch-size", "4096"); - input.put("paimon.table-option.file.compression.per.level", "0:lz4,1:zstd"); + input.put("paimon.table-option.file-reader-async-threshold", "16 MB"); + input.put("paimon.table-option.file-index.read.enabled", "false"); + input.put("paimon.table-option.source.split.target-size", "64 MB"); + input.put("paimon.table-option.source.split.open-file-cost", "1 MB"); + input.put("paimon.table-option.scan.manifest.parallelism", "1"); + input.put("paimon.table-option.scan.plan-sort-partition", "true"); TestPaimonProperties testProps = new TestPaimonProperties(input); testProps.initNormalizeAndCheckProps(); testProps.buildCatalogOptions(); Assertions.assertEquals("4096", testProps.getTableOptionsMap().get("read.batch-size")); - Assertions.assertEquals( - "0:lz4,1:zstd", testProps.getTableOptionsMap().get("file.compression.per.level")); + Assertions.assertEquals("16 MB", + testProps.getTableOptionsMap().get("file-reader-async-threshold")); + Assertions.assertEquals("false", testProps.getTableOptionsMap().get("file-index.read.enabled")); + Assertions.assertEquals("64 MB", testProps.getTableOptionsMap().get("source.split.target-size")); + Assertions.assertEquals("1 MB", testProps.getTableOptionsMap().get("source.split.open-file-cost")); + Assertions.assertEquals("1", testProps.getTableOptionsMap().get("scan.manifest.parallelism")); + Assertions.assertEquals("true", testProps.getTableOptionsMap().get("scan.plan-sort-partition")); Assertions.assertFalse(testProps.getCatalogOptionsMap().containsKey("table-option.read.batch-size")); Assertions.assertFalse(testProps.getCatalogOptionsMap().containsKey("jni.enable_jni_io_manager")); } @Test - void testPaimonTableOptionsTakePrecedenceOverCatalogOptions() { + void testCatalogReaderOptionsTakePrecedenceOverPhysicalTableOptions() { Map input = new HashMap<>(); input.put("warehouse", "s3://tmp/warehouse"); input.put("paimon.table-option.read.batch-size", "4096"); - input.put("paimon.table-option.write.batch-size", "2048"); - input.put("paimon.table-option.file.compression.per.level", "0:lz4,1:zstd"); + input.put("paimon.table-option.file-reader-async-threshold", "16 MB"); TestPaimonProperties testProps = new TestPaimonProperties(input); testProps.initNormalizeAndCheckProps(); - Map currentTableOptions = new HashMap<>(); - currentTableOptions.put("read.batch-size", "1024"); - currentTableOptions.put("orc.write.batch-size", "512"); - currentTableOptions.put("file.compression.per.level", "0:snappy"); - - Map optionsForCopy = - testProps.getTableOptionsForCopy(currentTableOptions); + Map optionsForCopy = testProps.getTableOptionsForCopy(); - Assertions.assertFalse(optionsForCopy.containsKey("read.batch-size")); - Assertions.assertFalse(optionsForCopy.containsKey("write.batch-size")); - Assertions.assertFalse(optionsForCopy.containsKey("file.compression.per.level")); + Assertions.assertEquals("4096", optionsForCopy.get("read.batch-size")); + Assertions.assertEquals("16 MB", optionsForCopy.get("file-reader-async-threshold")); } @Test @@ -137,50 +140,99 @@ void testCatalogTableOptionsFillMissingPaimonTableOptions() { TestPaimonProperties testProps = new TestPaimonProperties(input); testProps.initNormalizeAndCheckProps(); - Map optionsForCopy = - testProps.getTableOptionsForCopy(Collections.singletonMap( - "path", "s3://tmp/warehouse/test.db/test")); + Map optionsForCopy = testProps.getTableOptionsForCopy(); Assertions.assertEquals("4096", optionsForCopy.get("read.batch-size")); } @Test - void testRejectUnknownTableOption() { + void testPersistedUnknownTableOptionDoesNotPreventCatalogLoading() { Map input = new HashMap<>(); input.put("warehouse", "s3://tmp/warehouse"); input.put("paimon.table-option.option-does-not-exist", "value"); TestPaimonProperties testProps = new TestPaimonProperties(input); - IllegalArgumentException exception = Assertions.assertThrows( - IllegalArgumentException.class, testProps::initNormalizeAndCheckProps); + testProps.initNormalizeAndCheckProps(); - Assertions.assertTrue(exception.getMessage().contains("option-does-not-exist")); + Assertions.assertTrue(testProps.getTableOptionsMap().isEmpty()); } @Test - void testRejectPrefixMapTableOption() { + void testPersistedPrefixMapTableOptionDoesNotPreventCatalogLoading() { Map input = new HashMap<>(); input.put("warehouse", "s3://tmp/warehouse"); input.put("paimon.table-option.file.compression.per.level.0", "lz4"); TestPaimonProperties testProps = new TestPaimonProperties(input); - IllegalArgumentException exception = Assertions.assertThrows( - IllegalArgumentException.class, testProps::initNormalizeAndCheckProps); + testProps.initNormalizeAndCheckProps(); - Assertions.assertTrue(exception.getMessage().contains("file.compression.per.level.0")); + Assertions.assertTrue(testProps.getTableOptionsMap().isEmpty()); } @Test - void testRejectInvalidTableOptionValue() { + void testPersistedInvalidReaderOptionIsIgnoredDuringCatalogLoading() { Map input = new HashMap<>(); input.put("warehouse", "s3://tmp/warehouse"); input.put("paimon.table-option.read.batch-size", "not-an-integer"); TestPaimonProperties testProps = new TestPaimonProperties(input); - IllegalArgumentException exception = Assertions.assertThrows( - IllegalArgumentException.class, testProps::initNormalizeAndCheckProps); + testProps.initNormalizeAndCheckProps(); + + Assertions.assertTrue(testProps.getTableOptionsMap().isEmpty()); + } + + @Test + void testPersistedUnsafeTableOptionsAreIgnoredDuringCatalogLoading() { + for (String option : new String[] { + "branch", "path", "scan.tag-name", "scan.snapshot-id", + "write.batch-size", "file.compression.per.level" + }) { + Map input = new HashMap<>(); + input.put("warehouse", "s3://tmp/warehouse"); + input.put("paimon.table-option." + option, "1"); + + TestPaimonProperties testProps = new TestPaimonProperties(input); + testProps.initNormalizeAndCheckProps(); + Assertions.assertTrue(testProps.getTableOptionsMap().isEmpty(), option); + } + } + + @Test + void testPersistedOutOfRangeReaderOptionsAreIgnoredDuringCatalogLoading() { + Map invalidOptions = new HashMap<>(); + invalidOptions.put("read.batch-size", "0"); + invalidOptions.put("read.batch-size-negative", "-1"); + invalidOptions.put("read.batch-size-too-large", "65537"); + invalidOptions.put("file-reader-async-threshold", "0 B"); + invalidOptions.put("file-reader-async-threshold-too-small", "512 KB"); + invalidOptions.put("file-reader-async-threshold-too-large", "2 GB"); + + invalidOptions.forEach((caseName, value) -> { + String option = caseName.startsWith("read.batch-size") + ? "read.batch-size" : "file-reader-async-threshold"; + Map input = new HashMap<>(); + input.put("warehouse", "s3://tmp/warehouse"); + input.put("paimon.table-option." + option, value); + + TestPaimonProperties testProps = new TestPaimonProperties(input); + testProps.initNormalizeAndCheckProps(); + Assertions.assertTrue(testProps.getTableOptionsMap().isEmpty(), caseName); + }); + } + + @Test + void testImageRoundTripKeepsLegacyCatalogLoadableButDoesNotApplyUnsafeOption() { + Map input = new HashMap<>(); + input.put("warehouse", "s3://tmp/warehouse"); + input.put("paimon.table-option.write.batch-size", "2048"); + CatalogProperty persisted = new CatalogProperty(null, input); + + CatalogProperty restored = GsonUtils.GSON.fromJson( + GsonUtils.GSON.toJson(persisted), CatalogProperty.class); + TestPaimonProperties testProps = new TestPaimonProperties(restored.getProperties()); + testProps.initNormalizeAndCheckProps(); - Assertions.assertTrue(exception.getMessage().contains("read.batch-size")); + Assertions.assertTrue(testProps.getTableOptionsMap().isEmpty()); } @Test diff --git a/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisManagerTest.java b/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisManagerTest.java index 30b7bd55ed5010..0e1e95025fc8ae 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisManagerTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisManagerTest.java @@ -29,6 +29,10 @@ import org.apache.doris.common.DdlException; import org.apache.doris.common.Pair; import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.datasource.paimon.PaimonExternalCatalog; +import org.apache.doris.datasource.paimon.PaimonExternalDatabase; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.datasource.paimon.PaimonUtils; import org.apache.doris.nereids.trees.expressions.ExprId; import org.apache.doris.nereids.trees.expressions.Slot; import org.apache.doris.nereids.trees.expressions.SlotReference; @@ -38,9 +42,20 @@ import org.apache.doris.statistics.AnalysisInfo.AnalysisType; import org.apache.doris.statistics.AnalysisInfo.JobType; import org.apache.doris.statistics.AnalysisInfo.ScheduleType; +import org.apache.doris.statistics.util.StatisticsUtil; import org.apache.doris.thrift.TQueryColumn; import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.AppendOnlyFileStoreTable; +import org.apache.paimon.table.CatalogEnvironment; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.IntType; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.mockito.MockedStatic; @@ -53,6 +68,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; // CHECKSTYLE OFF @@ -118,6 +134,55 @@ public void testBuildAnalysisJobInfoAutoSampleCommandCollectsHotValue() { } } + @Test + public void testManualAnalysisValidatesPaimonRowCountBeforePlanning() { + AtomicBoolean planningStarted = new AtomicBoolean(false); + FileStoreTable unsafeTable = new AppendOnlyFileStoreTable( + Mockito.mock(FileIO.class), + new Path("memory://manual_analysis_guard"), + new TableSchema( + 0, + Collections.singletonList(new DataField(0, "id", new IntType())), + 0, + Collections.emptyList(), + Collections.emptyList(), + ImmutableMap.of("scan.manifest.parallelism", "0"), + null), + CatalogEnvironment.empty()) { + @Override + public ReadBuilder newReadBuilder() { + planningStarted.set(true); + return super.newReadBuilder(); + } + }; + PaimonExternalCatalog catalog = Mockito.mock(PaimonExternalCatalog.class); + PaimonExternalDatabase database = Mockito.mock(PaimonExternalDatabase.class); + Mockito.when(catalog.getId()).thenReturn(1L); + Mockito.when(database.getFullName()).thenReturn("db"); + Mockito.when(database.getRemoteName()).thenReturn("db"); + PaimonExternalTable table = Mockito.spy(new TestPaimonExternalTable(catalog, database)); + Mockito.doReturn(-1L).when(table).getRowCount(); + Mockito.doReturn(Collections.emptySet()).when(table).getColumnIndexPairs(Mockito.any()); + AnalyzeTableCommand command = mockAnalyzeCommand(AnalysisMethod.FULL, ScheduleType.ONCE, false, false); + Mockito.when(command.getTable()).thenReturn(table); + AnalysisManager manager = new AnalysisManager(); + Env env = Mockito.mock(Env.class); + + try (MockedStatic envMockedStatic = Mockito.mockStatic(Env.class); + MockedStatic paimonUtils = Mockito.mockStatic(PaimonUtils.class); + MockedStatic statisticsUtil = Mockito.mockStatic( + StatisticsUtil.class, Mockito.CALLS_REAL_METHODS)) { + envMockedStatic.when(Env::getCurrentEnv).thenReturn(env); + Mockito.when(env.getNextId()).thenReturn(1L); + paimonUtils.when(() -> PaimonUtils.getPaimonTable(table)).thenReturn(unsafeTable); + statisticsUtil.when(() -> StatisticsUtil.isEmptyTable(table, AnalysisMethod.FULL)).thenReturn(false); + + Assertions.assertThrows(IllegalArgumentException.class, + () -> manager.buildAnalysisJobInfo(command)); + } + Assertions.assertFalse(planningStarted.get()); + } + @Test public void testUpdateTaskStatus() { BaseAnalysisTask task1 = Mockito.mock(BaseAnalysisTask.class); @@ -473,4 +538,14 @@ private AnalyzeTableCommand mockAnalyzeCommand(AnalysisMethod analysisMethod, Sc Mockito.when(command.getAnalyzeProperties()).thenReturn(analyzeProperties); return command; } + + private static class TestPaimonExternalTable extends PaimonExternalTable { + private TestPaimonExternalTable(PaimonExternalCatalog catalog, PaimonExternalDatabase database) { + super(30001L, "table", "table", catalog, database); + } + + @Override + protected synchronized void makeSureInitialized() { + } + } } diff --git a/regression-test/data/external_table_p0/paimon/test_paimon_jni_reader_guardrails.out b/regression-test/data/external_table_p0/paimon/test_paimon_jni_reader_guardrails.out new file mode 100644 index 00000000000000..b149c4535db0e6 --- /dev/null +++ b/regression-test/data/external_table_p0/paimon/test_paimon_jni_reader_guardrails.out @@ -0,0 +1,44 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !scanner_v2_all_identifiers -- +1 east,01 hash-one first row 华东 nested-one east,01 a:1 +2 west,02 hash-two second row 华西 nested-two west,02 b:2 + +-- !scanner_v1_quoted_nested -- +1 east,01 nested-one east,01 a:1 +2 west,02 nested-two west,02 b:2 + +-- !scanner_v1_empty_identifier -- +empty-name + +-- !scanner_v2_quoted_nested -- +1 east,01 nested-one east,01 a:1 +2 west,02 nested-two west,02 b:2 + +-- !scanner_v2_empty_identifier -- +empty-name + +-- !catalog_override_physical_batch -- +1 + +-- !relation_reader_options -- +1 +2 + +-- !relation_option_isolation -- +1 1 +2 2 + +-- !failed_alter_preserves_catalog -- +2 + +-- !relation_override_physical_batch -- +1 + +-- !partitioned_relation_override_physical_batch -- +1 20 + +-- !relation_override_physical_manifest -- +1 10 + +-- !system_table_descriptor_with_safe_override -- +1 diff --git a/regression-test/suites/external_table_p0/paimon/test_paimon_jni_reader_guardrails.groovy b/regression-test/suites/external_table_p0/paimon/test_paimon_jni_reader_guardrails.groovy new file mode 100644 index 00000000000000..0f8780e2e6b5e7 --- /dev/null +++ b/regression-test/suites/external_table_p0/paimon/test_paimon_jni_reader_guardrails.groovy @@ -0,0 +1,231 @@ +// 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. + +suite("test_paimon_jni_reader_guardrails", "p0,external,paimon") { + String enabled = context.config.otherConfigs.get("enablePaimonTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable paimon test") + return + } + + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String catalogName = "test_paimon_jni_reader_guardrails" + String physicalCatalogName = "test_paimon_jni_reader_physical_guardrails" + String dbName = "paimon_jni_guardrails_db" + def scannerV2Rows = sql "show variables like 'enable_file_scanner_v2'" + String originalScannerV2 = scannerV2Rows[0][1] + + def catalogDdl = { String name, String extraProperty -> + return """ + create catalog ${name} properties ( + 'type'='paimon', + 'warehouse'='s3://warehouse/wh', + 's3.endpoint'='http://${externalEnvIp}:${minioPort}', + 's3.access_key'='admin', + 's3.secret_key'='password', + 's3.path.style.access'='true' + ${extraProperty} + ) + """ + } + + for (def invalid : [ + ["invalid_batch", ", 'paimon.table-option.read.batch-size'='0'", "read.batch-size"], + ["unsafe_branch", ", 'paimon.table-option.branch'='archive'", "branch"] + ]) { + String invalidCatalog = "${catalogName}_${invalid[0]}" + sql "drop catalog if exists ${invalidCatalog}" + try { + test { + sql(catalogDdl(invalidCatalog, invalid[1])) + exception invalid[2] + } + } finally { + sql "drop catalog if exists ${invalidCatalog}" + } + } + + sql "drop catalog if exists ${catalogName}" + sql(catalogDdl(catalogName, """ + , 'paimon.table-option.read.batch-size'='1024' + , 'paimon.table-option.file-reader-async-threshold'='16 MB' + , 'paimon.table-option.file-index.read.enabled'='false' + , 'paimon.table-option.source.split.target-size'='64 MB' + , 'paimon.table-option.source.split.open-file-cost'='1 MB' + , 'paimon.table-option.scan.manifest.parallelism'='1' + , 'paimon.table-option.scan.plan-sort-partition'='true' + """)) + + try { + spark_paimon_multi """ + create database if not exists paimon.${dbName}; + drop table if exists paimon.${dbName}.quoted_reader_options; + create table paimon.${dbName}.quoted_reader_options ( + id int, + `region,code` string, + `hash#name` string, + `display name` string, + `地区 名` string, + `nested#value` struct<`hash#name`:string,`region,code`:string,`colon:name`:string> + ) using paimon + tblproperties ('file.format'='parquet', 'read.batch-size'='512'); + insert into paimon.${dbName}.quoted_reader_options values + (1, 'east,01', 'hash-one', 'first row', '华东', + named_struct('hash#name', 'nested-one', 'region,code', 'east,01', 'colon:name', 'a:1')), + (2, 'west,02', 'hash-two', 'second row', '华西', + named_struct('hash#name', 'nested-two', 'region,code', 'west,02', 'colon:name', 'b:2')); + drop table if exists paimon.${dbName}.unsafe_physical_batch; + create table paimon.${dbName}.unsafe_physical_batch (id int) using paimon + tblproperties ('read.batch-size'='0'); + insert into paimon.${dbName}.unsafe_physical_batch values (1); + drop table if exists paimon.${dbName}.unsafe_partitioned_batch; + create table paimon.${dbName}.unsafe_partitioned_batch (id int, part int) using paimon + partitioned by (part) + tblproperties ('read.batch-size'='0'); + insert into paimon.${dbName}.unsafe_partitioned_batch values (1, 20); + drop table if exists paimon.${dbName}.unsafe_physical_manifest; + create table paimon.${dbName}.unsafe_physical_manifest (id int, part int) using paimon + partitioned by (part) + tblproperties ('scan.manifest.parallelism'='0'); + insert into paimon.${dbName}.unsafe_physical_manifest values (1, 10); + drop table if exists paimon.${dbName}.empty_identifier; + create table paimon.${dbName}.empty_identifier (`` string) using paimon; + insert into paimon.${dbName}.empty_identifier values ('empty-name'); + """ + + sql "switch ${catalogName}" + sql "use ${dbName}" + sql "set force_jni_scanner=true" + sql "set enable_file_scanner_v2=true" + + order_qt_scanner_v2_all_identifiers """ + select id, `region,code`, `hash#name`, `display name`, `地区 名`, + `nested#value`.`hash#name`, `nested#value`.`region,code`, + `nested#value`.`colon:name` + from quoted_reader_options + where `region,code` in ('east,01', 'west,02') + order by id + """ + + sql "set enable_file_scanner_v2=false" + order_qt_scanner_v1_quoted_nested """ + select id, `region,code`, `nested#value`.`hash#name`, + `nested#value`.`region,code`, `nested#value`.`colon:name` + from quoted_reader_options + order by id + """ + qt_scanner_v1_empty_identifier "select * from empty_identifier" + sql "set enable_file_scanner_v2=true" + order_qt_scanner_v2_quoted_nested """ + select id, `region,code`, `nested#value`.`hash#name`, + `nested#value`.`region,code`, `nested#value`.`colon:name` + from quoted_reader_options + order by id + """ + qt_scanner_v2_empty_identifier "select * from empty_identifier" + + // The safe catalog value must override the physical read.batch-size=0 value. + order_qt_catalog_override_physical_batch "select * from unsafe_physical_batch order by id" + + order_qt_relation_reader_options """ + select id from quoted_reader_options@options( + 'read.batch-size'='4096', + 'file-reader-async-threshold'='32 MB', + 'file-index.read.enabled'='true', + 'source.split.target-size'='32 MB', + 'source.split.open-file-cost'='2 MB', + 'scan.manifest.parallelism'='1', + 'scan.plan-sort-partition'='false') + order by id + """ + order_qt_relation_option_isolation """ + select small.id, large.id + from quoted_reader_options@options('read.batch-size'='1') small + join quoted_reader_options@options('read.batch-size'='8192') large + on small.id = large.id + order by small.id + """ + + for (def invalidOption : [ + ["read.batch-size", "0"], + ["read.batch-size", "65537"], + ["file-reader-async-threshold", "512 KB"], + ["file-reader-async-threshold", "2 GB"], + ["scan.manifest.parallelism", "0"], + ["scan.manifest.parallelism", "2147483647"] + ]) { + test { + sql """ + select * from quoted_reader_options@options( + '${invalidOption[0]}'='${invalidOption[1]}') + """ + exception invalidOption[0] + } + } + + test { + sql """ + alter catalog ${catalogName} set properties ( + 'paimon.table-option.read.batch-size'='0') + """ + exception "read.batch-size" + } + // A failed ALTER must not leave read.batch-size=0 behind; on the old path this follow-up + // JNI scan can stop making progress instead of completing. + qt_failed_alter_preserves_catalog "select count(*) from quoted_reader_options" + + sql "drop catalog if exists ${physicalCatalogName}" + sql(catalogDdl(physicalCatalogName, "")) + sql "switch ${physicalCatalogName}" + sql "use ${dbName}" + test { + sql "select * from unsafe_physical_batch" + exception "read.batch-size" + } + order_qt_relation_override_physical_batch """ + select * from unsafe_physical_batch@options('read.batch-size'='4096') order by id + """ + test { + sql "select * from unsafe_partitioned_batch" + exception "read.batch-size" + } + order_qt_partitioned_relation_override_physical_batch """ + select * from unsafe_partitioned_batch@options('read.batch-size'='4096') order by id + """ + test { + sql "select * from unsafe_physical_manifest" + exception "scan.manifest.parallelism" + } + order_qt_relation_override_physical_manifest """ + select * from unsafe_physical_manifest@options('scan.manifest.parallelism'='1') order by id + """ + test { + sql "select count(*) from unsafe_physical_manifest\$partitions" + exception "scan.manifest.parallelism" + } + qt_system_table_descriptor_with_safe_override """ + select count(*) from unsafe_physical_manifest\$partitions + @options('scan.manifest.parallelism'='1') + """ + } finally { + sql "set force_jni_scanner=false" + sql "set enable_file_scanner_v2=${originalScannerV2}" + sql "drop catalog if exists ${physicalCatalogName}" + sql "drop catalog if exists ${catalogName}" + } +}