PaimonReaderOptions.java

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package org.apache.doris.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.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.OptionalInt;
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;
    // Keep catalog replay deterministic while bounding a single option's JVM-wide thread impact.
    public static final int MAX_MANIFEST_PARALLELISM = 256;
    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<String> 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<String> 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<String> supportedOptions() {
        return SUPPORTED_OPTIONS;
    }

    public static Set<String> 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<String, String> 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<String, String> compatibleCatalogOptions(Map<String, String> properties) {
        Map<String, String> 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 validateReaderOptions(Map<String, String> options) {
        SUPPORTED_OPTIONS.stream()
                .filter(options::containsKey)
                .forEach(key -> validate(key, options.get(key)));
    }

    public static void validateEffectiveTableOptions(Map<String, String> options) {
        validateReaderOptions(options);
        validateIfPresentForRuntime(options, CoreOptions.SCAN_MANIFEST_PARALLELISM.key());
    }

    public static Map<String, String> runtimeSafeCopyOptions(Table table, Map<String, String> copyOptions) {
        Map<String, String> safeOptions = new LinkedHashMap<>(copyOptions);
        String key = CoreOptions.SCAN_MANIFEST_PARALLELISM.key();
        if (safeOptions.containsKey(key)) {
            String configured = safeOptions.get(key);
            if (configured == null) {
                return safeOptions;
            }
            validateManifestParallelism(configured);
            int requested = Integer.parseInt(configured);
            int localCapacity = Runtime.getRuntime().availableProcessors();
            if (requested > localCapacity) {
                safeOptions.put(key, String.valueOf(localCapacity));
            }
            return safeOptions;
        }

        List<Integer> configuredValues = new ArrayList<>();
        collectManifestParallelism(table, configuredValues);
        if (configuredValues.isEmpty()) {
            return safeOptions;
        }
        int localCapacity = Runtime.getRuntime().availableProcessors();
        if (configuredValues.stream().anyMatch(value -> value > localCapacity)) {
            // Keep persisted semantics stable across heterogeneous FEs, but cap the execution copy
            // conservatively across every nested planner hidden by a wrapper.
            int safeParallelism = configuredValues.stream()
                    .mapToInt(Integer::intValue)
                    .min()
                    .orElse(localCapacity);
            safeOptions.put(key, String.valueOf(Math.min(safeParallelism, localCapacity)));
        }
        return safeOptions;
    }

    public static Table runtimeSafeTable(Table table) {
        Map<String, String> safeOptions = runtimeSafeCopyOptions(table, Collections.emptyMap());
        return safeOptions.isEmpty() ? table : table.copy(safeOptions);
    }

    public static Table runtimeSafeTable(Table table, Map<String, String> copyOptions) {
        return table.copy(runtimeSafeCopyOptions(table, copyOptions));
    }

    public static OptionalInt runtimeSafeManifestParallelism(Table table) {
        List<Integer> configuredValues = new ArrayList<>();
        collectManifestParallelism(table, configuredValues);
        if (configuredValues.isEmpty()) {
            return OptionalInt.empty();
        }
        return OptionalInt.of(Math.min(
                configuredValues.stream().mapToInt(Integer::intValue).min().getAsInt(),
                Runtime.getRuntime().availableProcessors()));
    }

    private static void collectManifestParallelism(Table table, List<Integer> configuredValues) {
        Map<String, String> options = table.options();
        String key = CoreOptions.SCAN_MANIFEST_PARALLELISM.key();
        if (options != null && options.containsKey(key) && options.get(key) != null) {
            String configured = options.get(key);
            validateManifestParallelism(configured);
            configuredValues.add(Integer.parseInt(configured));
        }
        if (table instanceof FallbackReadFileStoreTable) {
            collectManifestParallelism(((FallbackReadFileStoreTable) table).fallback(), configuredValues);
        }
        if (table instanceof DelegatedFileStoreTable) {
            collectManifestParallelism(((DelegatedFileStoreTable) table).wrapped(), configuredValues);
        }
    }

    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.
        validateIfPresentForRuntime(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<String, String> options, String key) {
        if (options.containsKey(key)) {
            validate(key, options.get(key));
        }
    }

    private static void validateIfPresentForRuntime(Map<String, String> options, String key) {
        validateIfPresent(options, key);
        if (options.containsKey(key)) {
            validateRuntimeManifestParallelism(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);
        }
        // Catalog properties are replayed on every FE, so validity must not depend on the CPU count
        // of whichever node happens to deserialize the image or become leader.
        requireRange(CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), parallelism, 1, MAX_MANIFEST_PARALLELISM);
    }

    private static void validateRuntimeManifestParallelism(String value) {
        int parallelism = Integer.parseInt(value);
        requireRange(CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), parallelism, 1,
                Runtime.getRuntime().availableProcessors());
    }

    private static <T> T parse(String key, String value, ConfigOption<T> 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);
        }
    }
}